Async Python

The problem: An AI service may wait on several models, tools, databases, or data sources. If independent waits happen one after another, their delays add together while the computer is mostly idle.

What you will learn: Async Python lets an event loop run another ready task while one task is waiting. You will learn async, await, concurrent tasks, cancellation, and concurrency limits using agent and RAG examples. The essential boundary is that async overlaps waiting; it does not automatically make CPU-heavy model computation faster.

1. Why Async Programming Exists

Async helps when a program spends time waiting, especially for networks, databases, files, or model responses. It does not make every calculation faster.

task A computes → waits for API ─────────→ resumes
                         ↓ event loop runs ready work
task B                   computes → waits → resumes

There may still be only one thread executing Python code at a moment. The event loop cooperatively switches tasks when one reaches await. If a task performs long CPU work without yielding, it blocks the loop and delays every other task.

The Waiting Problem

Imagine your AI service needs to call three different tools to answer one question: a web search, a database lookup, and a calculator. Each takes about 1 second — mostly spent waiting for a network response, not actually computing anything.

# Bad / naive approach — synchronous, one at a time
import time

def call_tool(name):
    time.sleep(1)   # simulates waiting on a network response
    return f"{name} result"

start = time.time()
result_a = call_tool("web_search")
result_b = call_tool("database")
result_c = call_tool("calculator")
print(f"Took {time.time() - start:.1f}s")

Expected Output:

Took 3.0s

Three seconds — even though your program spent almost the entire time just waiting, not actually doing computational work. That waiting time is completely wasted if nothing else can happen during it.

🧠 Why Does It Exist? Async programming exists to let your program do something else productive while waiting, instead of sitting idle.

2. Synchronous vs Asynchronous Programming

Real-World Analogy — the waiter

“Think of a restaurant waiter who doesn’t stand idle while one table is waiting for the kitchen.”

A synchronous waiter takes Table A’s order, walks it to the kitchen, and stands there staring at the kitchen door until the food is ready — completely ignoring Tables B and C the entire time.

An asynchronous waiter takes Table A’s order, hands it to the kitchen, and immediately goes to take Table B’s order while the kitchen works — coming back to Table A only once the food is actually ready. One waiter, three tables served far faster, because waiting time is never wasted.

Visualizing it

graph TD
    subgraph Synchronous Timeline (Total: 3s)
        S1[Start] --> SA["Task A (waits 1s)"]
        SA --> SB["Task B (waits 1s)"]
        SB --> SC["Task C (waits 1s)"]
        SC --> SE([End])
    end

    subgraph Asynchronous Timeline (Total: 1s)
        A1[Start] --> A_All["Trigger Tasks A, B, and C concurrently"]
        A_All --> AA["Task A (waits 1s)"]
        A_All --> AB["Task B (waits 1s)"]
        A_All --> AC["Task C (waits 1s)"]
        AA --> AE([End when all finish])
        AB --> AE
        AC --> AE
    end

🧠 Intuition

Synchronous code does things one after another, in strict sequence. Asynchronous code starts multiple waiting operations, and while each one waits, lets others make progress too — dramatically reducing total wall-clock time when the work is mostly “waiting on something external,” like a network call.

3. Blocking vs Non-Blocking

What Is It?

A blocking operation freezes your entire program until it finishes. A non-blocking operation lets your program continue doing other work while it’s in progress.

# Blocking: nothing else can happen during time.sleep(1)
import time
time.sleep(1)

# Non-blocking equivalent (async): other code CAN run during this wait
import asyncio
await asyncio.sleep(1)

🧠 Intuition: A blocking call is like standing in a single-file checkout line where the whole store stops until you’re done. A non-blocking call is like an order number system — you wait for your number, but everyone else keeps shopping in the meantime.

🤖 How Is This Used in AI? Calling an LLM API is a blocking operation by default — your program does nothing at all while it waits for the model to respond. Async lets you start several such calls, and let Python work on other requests while each one waits, instead of one call completely freezing everything else.

4. async and await

What Is It?

async def defines a coroutine — a special kind of function that can be paused and resumed.
await pauses that coroutine at a point where it’s waiting on something (a network call, a sleep), without blocking the rest of the program.

Syntax

import asyncio

async def call_tool(name):
    print(f"Starting {name}")
    await asyncio.sleep(1)   # pauses HERE, lets other coroutines run meanwhile
    print(f"Finished {name}")
    return f"{name} result"

async def main():
    result = await call_tool("web_search")
    print(result)

asyncio.run(main())

Expected Output:

Starting web_search
Finished web_search
web_search result

How It Works

async def marks a function as a coroutine — calling it doesn’t run it immediately; it returns a coroutine object that needs to be awaited or scheduled.
await says “pause here until this finishes, but let other coroutines make progress in the meantime.”
asyncio.run(main()) is the entry point that actually starts the async event loop and runs your top-level coroutine.

⚠️ Common Beginner Mistake: Calling an async function without await (or asyncio.run):
async def call_tool():
    ...

call_tool()   # Nothing happens! Returns a coroutine object, doesn't run it.
Python will actually warn you: RuntimeWarning: coroutine 'call_tool' was never awaited. An async def function must be awaited (from inside another coroutine) or run with asyncio.run(...) to actually execute.

5. asyncio

asyncio is Python’s built-in library for writing and running async code — the event loop that actually manages “which coroutine runs when.”

import asyncio

async def greet(name, delay):
    await asyncio.sleep(delay)
    print(f"Hello, {name}!")

asyncio.run(greet("Claude", 1))

🧠 Intuition: Think of asyncio’s event loop as the restaurant manager coordinating the waiter — deciding which paused task to resume the moment it’s ready to continue, so nothing sits idle longer than necessary.

6. Running Multiple Tasks — the real payoff

This is where async actually earns its complexity. Compare the earlier synchronous 3-second example to its async equivalent:

import asyncio
import time

async def call_tool(name):
    await asyncio.sleep(1)   # simulates waiting on a network response
    return f"{name} result"

async def main():
    start = time.time()

    # Run all three "at the same time" — start them all, then wait together
    results = await asyncio.gather(
        call_tool("web_search"),
        call_tool("database"),
        call_tool("calculator"),
    )

    print(results)
    print(f"Took {time.time() - start:.1f}s")

asyncio.run(main())

Expected Output:

['web_search result', 'database result', 'calculator result']
Took 1.0s

Why the better approach matters in AI

One second, not three — even though we made three separate 1-second calls. asyncio.gather(...) starts all three coroutines, and while each one is await asyncio.sleep(1)-ing (waiting), the others get to run too. The waiting overlaps; only the truly sequential parts add up.

🧠 Intuition

asyncio.gather(a, b, c) is literally the async waiter from our analogy: start Table A’s order, immediately start Table B’s, immediately start Table C’s, then collect all three results once they’re all ready — instead of handling each table completely before moving to the next.

💡 Triggering Background Tasks: `asyncio.create_task`

In AI applications, you may want to start a slow operation (like saving a prompt/response pair to a database, or sending a metric to an analytics API) but return the response to the user immediately without waiting for the slow operation to finish.

To do this, use asyncio.create_task() to schedule a coroutine to run in the background:

import asyncio

async def save_to_database(data: dict):
    await asyncio.sleep(2)  # simulate slow database write
    print("Saved successfully to DB!")

async def handle_user_query(query: str):
    print("Generating response...")
    await asyncio.sleep(1)  # simulate model call
    response = "Paris is the capital."

    # Fire-and-forget: starts in the background, does NOT pause execution here!
    asyncio.create_task(save_to_database({"query": query, "response": response}))

    return response  # returned immediately after 1s, database continues in background

7. Concurrent API Calls

The single most valuable async pattern for AI development — calling multiple LLM/tool APIs concurrently instead of one after another:

import asyncio

async def call_llm(prompt, delay=1):
    print(f"Sending prompt: '{prompt}'")
    await asyncio.sleep(delay)   # simulates network latency
    return f"Response to: {prompt}"

async def process_batch(prompts):
    tasks = [call_llm(p) for p in prompts]
    responses = await asyncio.gather(*tasks)
    return responses

prompts = [
    "Summarize this document",
    "Translate this sentence",
    "Extract key entities",
]

responses = asyncio.run(process_batch(prompts))
for r in responses:
    print(r)

Expected Output:

Sending prompt: 'Summarize this document'
Sending prompt: 'Translate this sentence'
Sending prompt: 'Extract key entities'
Response to: Summarize this document
Response to: Translate this sentence
Response to: Extract key entities

🤖 How Is This Used in AI?

This is exactly the pattern behind:

Processing a batch of documents for embedding, all at once, instead of one at a time
Running multiple evaluation questions against a model concurrently
An agent calling several tools simultaneously (e.g., a web search AND a database lookup) instead of waiting for one to finish before starting the next

At real scale — hundreds or thousands of API calls — the difference between sequential and concurrent execution isn’t a matter of seconds; it’s the difference between minutes and hours.

[!IMPORTANT] Production Alert: Rate Limits and Concurrency Safety (asyncio.Semaphore) While running concurrent calls via asyncio.gather(*tasks) is incredibly fast, it can be dangerous in production. If you have a batch of 1,000 documents to embed, asyncio.gather will attempt to launch all 1,000 API requests simultaneously.

Doing this will instantly trigger the AI provider’s rate limits, returning HTTP 429 Rate Limit Exceeded errors, and could even get your IP temporarily blocked.

To prevent this, production code uses asyncio.Semaphore to set a limit on how many concurrent requests are allowed to run at any single moment:
# Limit to at most 10 concurrent requests at any time
semaphore = asyncio.Semaphore(10)

async def call_llm_safe(prompt):
    async with semaphore: # only 10 tasks can enter this block concurrently
        print(f"Sending prompt: '{prompt}'")
        await asyncio.sleep(1)
        return f"Response to: {prompt}"
Using a semaphore acts as a concurrency safety valve, giving you the speed of async without crashing into API rate limits!

8. Async HTTP Requests

The standard requests library (Module 11) is blocking — it doesn’t work with async/await. For real async HTTP calls, you’d use a library like httpx or aiohttp:

import asyncio
import httpx   # pip install httpx

async def fetch_completion(client, prompt):
    response = await client.post(
        "https://api.anthropic.com/v1/messages",
        headers={"x-api-key": "your_api_key_here", "anthropic-version": "2023-06-01"},
        json={"model": "claude-sonnet-4-6", "max_tokens": 200,
              "messages": [{"role": "user", "content": prompt}]},
    )
    return response.json()

async def main():
    prompts = ["What is Python?", "What is async programming?"]
    async with httpx.AsyncClient() as client:
        tasks = [fetch_completion(client, p) for p in prompts]
        results = await asyncio.gather(*tasks)
    return results

# results = asyncio.run(main())

🧠 Notice async with httpx.AsyncClient() as client: — an async context manager (Module 10’s context managers, adapted for async code) that properly sets up and tears down the HTTP connection pool around your concurrent calls.

9. Async AI Applications and Async LLM Calls

Most modern AI SDKs (including Anthropic’s and OpenAI’s) ship an async client alongside the regular one:

import asyncio
from anthropic import AsyncAnthropic   # note: Async prefix

async def ask_claude(client, prompt):
    response = await client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=300,
        messages=[{"role": "user", "content": prompt}],
    )
    return response.content[0].text

async def main():
    client = AsyncAnthropic(api_key="your_api_key_here")
    prompts = [
        "Explain RAG in one sentence.",
        "Explain embeddings in one sentence.",
        "Explain async programming in one sentence.",
    ]
    tasks = [ask_claude(client, p) for p in prompts]
    answers = await asyncio.gather(*tasks)
    for a in answers:
        print(a)

# asyncio.run(main())

🤖 This is the real, idiomatic pattern for evaluating a model against many prompts, or handling many simultaneous user requests in a production AI service — all three calls happen concurrently, dramatically reducing total time compared to awaiting each one sequentially.

10. Async Agent Tools

Agent frameworks often define tools as async functions so an agent can run multiple tool calls concurrently when they don’t depend on each other:

import asyncio

async def web_search(query):
    await asyncio.sleep(0.8)
    return f"Search results for: {query}"

async def database_lookup(user_id):
    await asyncio.sleep(0.5)
    return f"Database record for user: {user_id}"

async def run_agent_step(query, user_id):
    # These two tools don't depend on each other — run them concurrently
    search_result, db_result = await asyncio.gather(
        web_search(query),
        database_lookup(user_id),
    )
    return {
        "search": search_result,
        "database": db_result,
    }

result = asyncio.run(run_agent_step("latest AI news", "user_42"))
print(result)

Expected Output:

{'search': 'Search results for: latest AI news', 'database': 'Database record for user: user_42'}

🧠 Intuition: If one tool’s input depends on another tool’s output, they must run sequentially (await one, then the other). But independent tools — like a web search and an unrelated database lookup — should run concurrently with asyncio.gather, since there’s no reason to make one wait for the other.

11. Explaining What Async Does NOT Do

“Explain that async does NOT mean ‘everything magically becomes faster.’”

This is the single most important misconception to clear up.

import asyncio
import time

# CPU-bound work — async gives NO benefit here
async def compute_heavy_thing():
    total = 0
    for i in range(10_000_000):   # pure computation, no waiting involved
        total += i
    return total

async def main():
    start = time.time()
    await asyncio.gather(
        compute_heavy_thing(),
        compute_heavy_thing(),
    )
    print(f"Took {time.time() - start:.2f}s")

asyncio.run(main())

This will not run twice as fast, even though we used asyncio.gather. Why? Because compute_heavy_thing() never actually waits on anything — it’s pure CPU computation, with no await point where control can be handed off to another task. Python’s async model only helps when tasks spend time waiting (on network calls, disk I/O, timers) — it does not run multiple CPU-heavy calculations in true parallel (that requires multiprocessing, a different, more advanced topic outside this course’s scope).

[!NOTE] The Global Interpreter Lock (GIL) Connection The reason Python threads and coroutines cannot run CPU-heavy math in parallel on multiple CPU cores is Python’s Global Interpreter Lock (GIL).

The GIL is a mechanism that ensures only one thread executes Python code at any single instant.

For I/O tasks (waiting on network responses, files, or asyncio.sleep), Python releases the GIL, allowing other tasks to run during the wait.

For CPU tasks (running calculations or loops), Python holds onto the GIL, locking execution to a single CPU core.

Note: External libraries like NumPy or PyTorch release the GIL internally during heavy vector math, so they can run in parallel on multiple cores or GPUs!

When async is useful

Calling multiple external APIs (LLMs, tools, databases) — lots of waiting on network responses
Handling many simultaneous incoming requests in a web service
Reading/writing many files or making many small network calls

When async is NOT useful (or even makes things worse)

Pure number-crunching / heavy computation with no waiting involved
A single simple script that only ever does one thing at a time — the added complexity of async/await isn’t worth it
Code that’s already fast enough — don’t add async complexity “just in case”

⚠️ Common Beginner Mistake: Adding async/await everywhere out of habit, even for code that never actually waits on anything external. This adds real complexity (and genuinely confusing bugs, like forgetting an await) for zero performance benefit. Only reach for async when you have multiple independent, wait-heavy operations to run together.

Concurrency Is Not the Same as Parallelism

Asyncio provides concurrency: while task A waits for a network response, the event loop can let task B run. That is like one cook switching between dishes while each dish is waiting in the oven. Parallelism means work literally runs at the same time on multiple CPU cores or machines.

task A runs → awaits network ─────────────→ resumes
                    ↓ event loop
task B              runs → awaits ───────→ resumes

CPU-heavy image processing or model computation can block the event loop. Move such work to an appropriate process, worker, or accelerator rather than merely adding async.

Cancellation, Limits, and Structured Tasks

Async tasks can be cancelled when a user disconnects or a timeout expires. Use try/finally or async context managers so connections and files still close, and normally allow CancelledError to continue after cleanup.

A semaphore limits how many operations are in flight at once; it does not by itself enforce “20 requests per minute.” Use a rate limiter for a time-based quota. On modern Python, asyncio.TaskGroup keeps related tasks together: if one fails, sibling tasks are cancelled and their errors are collected in a structured way.

Module Summary

You now understand why async programming exists (overlapping wasted waiting time, not “making code faster” in general), how async/await and asyncio.gather work together to run multiple I/O-bound operations concurrently, how this applies directly to calling LLMs, tools, and APIs in an AI pipeline, and — critically — when async genuinely helps versus when it adds complexity for no benefit.

AI Connection

Real AI applications constantly juggle multiple slow, external operations: calling one or more LLMs, querying vector databases, running several agent tools. Async Python is how you run these concurrently instead of paying their full waiting cost one after another — the difference between a batch of 100 LLM calls taking 100 seconds sequentially versus a few seconds concurrently. This is precisely the mechanism underneath high-throughput AI services and agent frameworks.

Mini Practice

Write a synchronous version and an async version of three simulated 1-second API calls, and compare their total run times.
Write an async function fetch_all(prompts) that uses asyncio.gather to “call” a simulated LLM for each prompt concurrently.
Write two async tool functions that don’t depend on each other, and run them concurrently with asyncio.gather.
Write a CPU-heavy async function (a big loop, no await inside it) and explain, in your own words, why running two of them with asyncio.gather won’t be meaningfully faster than running them one after another.
Describe, in your own words, one real scenario in an AI application where async would help a lot, and one scenario where it wouldn’t help at all.

Next: Module 14 — Python for Modern AI Development — embeddings, RAG, vector databases, agents, tool calling, structured outputs, Pydantic, and where LangChain/LangGraph/MCP fit into everything you’ve learned.

Author: TechByteByByte Editorial Team
Reviewed by: TechByteByByte Admin
Published: August 31, 2026
Last reviewed: August 31, 2026

1. Why Async Programming Exists

The Waiting Problem

2. Synchronous vs Asynchronous Programming

Real-World Analogy — the waiter

Visualizing it

🧠 Intuition

3. Blocking vs Non-Blocking

What Is It?

4. async and await

What Is It?

Syntax

How It Works

5. asyncio

6. Running Multiple Tasks — the real payoff

Why the better approach matters in AI

🧠 Intuition

💡 Triggering Background Tasks: asyncio.create_task

7. Concurrent API Calls

🤖 How Is This Used in AI?

8. Async HTTP Requests

9. Async AI Applications and Async LLM Calls

10. Async Agent Tools

11. Explaining What Async Does NOT Do

When async is useful

When async is NOT useful (or even makes things worse)

Concurrency Is Not the Same as Parallelism

Cancellation, Limits, and Structured Tasks

Module Summary

AI Connection

Mini Practice

💡 Triggering Background Tasks: `asyncio.create_task`