83 - asyncio: the Plumbing an Agent Runs On

asyncio exists for one thing only: not standing still while you wait. An agent spends almost all of its time waiting for network answers — the model, the tools' APIs, the database — and during that wait the processor does nothing. With synchronous code that wait blocks everything; with a coroutine control goes back to the event loop, which in the meantime moves the other waits forward. It is not parallelism (a single thread runs, one instruction at a time): it is overlapping of waits, and for an agent it is exactly the resource that counts.

The analogy: the cook and the four burners

A synchronous cook puts the pasta on, stays in front of the pot for nine minutes, then lights the sauce, waits, then warms the bread. Thirty minutes (example times, to fix the idea). An asynchronous cook puts the pasta on, and while it boils lights the sauce and slides the bread into the oven; then moves between the burners as things get ready. Eleven minutes. It is still a single cook — it has not learned to chop two onions at once (that would be parallelism, and would require a second cook, that is, another process). It has only stopped staring at the pots.

The limit can already be deduced from this analogy: if the work is chopping onions and not waiting for the water to boil, asyncio is of no use. I/O bound work (network, disk, DB) → asyncio. CPU bound work (multiplying matrices, parsing 2 GB of JSON) → separate processes.

Three words that get confused: concurrency, parallelism, threads

Before the code it is worth sorting out the vocabulary, because most mistakes with asyncio come from a terminological confusion and not from a technical problem.

Concurrency means that several activities are in progress in the same interval of time. It says nothing about how many of them are advancing at this precise instant. The cook of the analogy is concurrent: pasta, sauce and bread are all "in progress", even if at any single moment he is doing one thing only.

Parallelism means that several activities physically advance together, on distinct computing units. Two cooks. It requires more cores and, in Python, more processes.

Threads are an operating-system mechanism to obtain concurrency (and in other languages also parallelism), in which it is the system, not your code, that decides who runs. It is the key difference from asyncio: threads are preemptive — they can be interrupted at any point, even halfway through an operation you considered indivisible — while coroutines are cooperative and yield control only at await points, which are visible in the code.

This difference has an enormous and often ignored practical consequence: with threads you cannot know, by reading a function, where it might be interrupted; with coroutines you can, because the interruption points are written down. Asynchronous code is easier to reason about for critical sections, and this makes up for the greater difficulty in debugging.

The third protagonist remains, which in Python has to be named: the Global Interpreter Lock, the lock that has historically prevented two threads from executing Python bytecode at the same time in the same process. It is the reason why, in Python, threads do not give parallelism on pure computation — they give concurrency on waiting, because the lock is released during I/O operations. The landscape is moving: Python 3.13 introduced experimental builds without the GIL and PEP 779 defines the criteria for their "supported" status (consulted on 2026-11-12). On the machine where I ran the examples of this lesson (Python 3.14.4) the GIL is still active (sys._is_gil_enabled() returns True), and in any case the practical rule for writing an agent today does not change: for waiting you use asyncio, for computation you use separate processes.

Hence the mental table, worth fixing in mind. Remember that each entry answers the same question: what is the program doing while it "waits"?

• Many network waits (calls to the model, APIs, queries): asyncio. This is the case of an agent, almost 100%.
• Heavy computation (embeddings on CPU, parsing huge files, numerical transformations): separate processes, with ProcessPoolExecutor or a separate service.
• Synchronous libraries you cannot change (a legacy database driver, an SDK with no async version): asyncio.to_thread, which moves the blocking call to a thread and gives control back to the event loop.
• A mix, which is the real case: async code for the flow, to_thread for the synchronous islands, a separate process for computation.

If the wait is on the network, asyncio; if it is computation, another process; if you have a library you cannot change, you isolate it in a thread with asyncio.to_thread without rewriting anything. The "mixed" case is the most frequent in real projects and is worth stating, because none of the three solutions alone covers a service that calls a model over the network, reads from a legacy driver and computes embeddings.

Where an agent really waits: anatomy of a latency

It is worth looking at where the time comes from in a typical run, because it determines where it pays to intervene and where it is useless. Take an agent that makes three turns and calls four tools in total, on a model served over the network. The times below are an example built for reasoning, not a measurement: yours will be different, and they are the ones you must measure.

Turn 1: call to the model              1.8 s   (network wait)
        execution of 2 tools           0.4 s   (wait: 2 DB queries, in parallel)
Turn 2: call to the model              2.4 s   (wait; the prompt is longer)
        execution of 1 tool            1.1 s   (wait: a slow external API)
Turn 3: call to the model              1.3 s   (wait; final answer, short)
        your code (validation,
        serialisation, logging)       0.02 s   (computation)
                                      -------
                                       7.02 s   of which 7.00 is waiting

The ratio is what counts: over 99% of the time is waiting. From this come three operational conclusions that are not obvious.

First: optimising your code is of no use. Those 20 milliseconds of validation and logging are irrelevant. The time is in the network calls, and the only way to reduce it is to make fewer of them (fewer turns, tools that return less context) or to make them together when they are independent.

Second: the agent's loop cannot be parallelised. Turn 2 needs the answer of turn 1: there is nothing to overlap. The 5.5 seconds of model calls are a lower bound for this run, and they come down only by reducing the number of turns — which is an agent design problem, not a concurrency one.

Third: what is parallelised are the tools within the same turn and the runs among themselves. In turn 1 the two tools go together: 0.4 seconds instead of 0.7. And if you have to process fifty documents, the fifty runs are independent: with a semaphore of ten the total time is about a tenth of the sequential one (fifty runs in five waves of ten); in the example executed further down, 0.5 seconds against 5.0.

There is also a fourth conclusion, which concerns the server more than the single user, and which is the main reason why this whole module is written in asynchronous code. With synchronous code, a process serving one user stays blocked for seven seconds, during which it can do nothing for anyone else: to serve ten users at the same time you need ten processes, with ten times the memory. With asynchronous code the same process, during those seven seconds of waiting, moves forward the requests of all the others. I have not measured the advantage on a real server; the order of magnitude depends on the ratio between waiting and computation, and in a run like the one above, where waiting exceeds 99%, it is large.

The difference, drawn on a timeline

Three calls of one second each. At the top the synchronous code, in the middle the most common mistake, at the bottom the version that works. The grey bars are pure waiting: time in which the processor does nothing and could be doing something else.

The mechanism in four objects

• Coroutine. A function declared async def. Calling it does not run it: it returns a coroutine object, inert, like a recipe written on a sheet of paper. Forgetting to await it is mistake number one for beginners (Python usually warns you with a RuntimeWarning: coroutine was never awaited).
• await. The point where the coroutine says "I am about to wait, take control in the meantime". It is the only place where the event loop can move on to something else. No await inside an async function means no concurrency.
• Event loop. The cooperative scheduler: it keeps the list of waits and, when the operating system tells it that a socket is ready, resumes the corresponding coroutine. asyncio.run() creates it, runs, closes.
• Task. A scheduled coroutine. asyncio.create_task(coro()) puts it in the race immediately; from that moment it proceeds even if you are doing something else. This is the fundamental difference between "there are two coroutines" and "two things are really advancing together".

The four objects read in sequence: a coroutine is the recipe, a task is the recipe put on the stove, the event loop is the cook moving between the burners, and await is the moment when the cook can move. If one of the four is missing, concurrency disappears without warning, which is why the mistakes of this lesson produce code that works, only slowly.

The measurement that clears everything up

Let us simulate three calls to a tool that takes one second each (asyncio.sleep stands in for the network wait). Three versions, same work. Run with Python 3.14.4, twice, with the same output:

import asyncio, time

async def tool(name: str, seconds: float) -> str:
    await asyncio.sleep(seconds)          # the (fake) network wait is here
    return f"{name} ok"

async def sequential():
    return [await tool(f"t{i}", 1) for i in range(3)]

async def concurrent_wrong():
    # BUG: create_task is there, but the await comes right after -> stays sequential
    out = []
    for i in range(3):
        t = asyncio.create_task(tool(f"t{i}", 1))
        out.append(await t)
    return out

async def concurrent():
    tasks = [asyncio.create_task(tool(f"t{i}", 1)) for i in range(3)]
    return await asyncio.gather(*tasks)

async def main():
    for f in (sequential, concurrent_wrong, concurrent):
        t0 = time.perf_counter()
        await f()
        print(f"{f.__name__:24s} {time.perf_counter() - t0:.1f}s")

asyncio.run(main())

# sequential               3.0s
# concurrent_wrong         3.0s
# concurrent               1.0s

The middle version is the most common bug of all: it creates the tasks and awaits them one by one, so it gains nothing. The operating rule: create all the tasks, then await. If between creation and waiting there is an await inside a loop, you have written synchronous code with asynchronous syntax. Over five further runs, the times printed with three decimals were 3.003-3.009 s for the first two versions and 1.001-1.004 s for the third: the difference between "wrong" and "sequential" is nil, the one with the third is two seconds.

Where it really matters in an agent

The agent's loop is sequential by nature: turn 2 depends on the answer of turn 1, there is nothing to overlap. The opportunities are three others, and they are the ones that count. The first: parallel tool calls in the same turn, because the model can ask for three tools in a single answer, and running them together brings the turn time down to that of the slowest tool. The second: several agents, or the same agent on several inputs, because twenty documents to classify are twenty independent runs. The third: serving several users with one process, the reason async HTTP libraries exist: with synchronous code a worker blocked waiting for the model cannot serve anyone else.

Here is the first opportunity, with two fake tools: one answers, the other raises an error. The calls have the form {"name": ..., "args": ...}, a simplification of the one in the messages of the previous lesson. The block does not launch itself: it is run by the main you will find after the controlled-concurrency examples, together with the other tests.

import asyncio, time

async def run_calls(tools: dict, calls: list) -> list:
    tasks = [asyncio.create_task(tools[c["name"]](**c["args"])) for c in calls]
    outcomes = await asyncio.gather(*tasks, return_exceptions=True)
    out = []
    for c, e in zip(calls, outcomes):
        if isinstance(e, Exception):
            out.append({"tool": c["name"], "error": f"{type(e).__name__}: {e}"})
        else:
            out.append({"tool": c["name"], "result": e})
    return out

async def stock(sku):
    await asyncio.sleep(0.2)
    return {"sku": sku, "stock": 4}

async def incoming(sku):
    await asyncio.sleep(0.3)
    raise KeyError(sku)                      # a tool that fails

async def try_calls():
    t0 = time.perf_counter()
    out = await run_calls({"stock": stock, "incoming": incoming},
                          [{"name": "stock", "args": {"sku": "A1"}},
                           {"name": "incoming", "args": {"sku": "A1"}}])
    print(out)
    print(f"duration {time.perf_counter() - t0:.1f}s (sequential: 0.5s)")

# [{'tool': 'stock', 'result': {'sku': 'A1', 'stock': 4}}, {'tool': 'incoming', 'error': "KeyError: 'A1'"}]
# duration 0.3s (sequential: 0.5s)

The detail that makes the difference is return_exceptions=True. Without it, the first exception makes the whole gather fail and you throw away the results of the tools that went well — in an agent, where a failing tool is useful information to give back to the model, it is exactly the behaviour you do not want. The output confirms it: the successful tool delivers its result, the failed one delivers a readable error, and the duration is that of the slowest (0.3 seconds) instead of the sum (0.5).

Controlled concurrency: semaphore, timeout, cancellation

"Do everything at once" works with three calls and gets you banned with three hundred. Three tools, all in the standard library. The examples, together with the parallel-calls one, are sections of a single file that I ran twice (Python 3.14.4, identical output) and the main at the bottom launches them in sequence. Each block has its own async def try_... that exercises it; the overall output is given after the main.

Semaphore — a ceiling on the number of simultaneous requests. The semaphore is a counter of seats: each async with sem takes one, and beyond the ceiling the coroutine waits at that line. In the test I launch 50 jobs of 0.1 seconds with a ceiling of 10 and count how many are really in flight at the same instant.

import asyncio, time

async def map_limited(coro_f, items, max_concurrent=5):
    sem = asyncio.Semaphore(max_concurrent)
    async def with_limit(x):
        async with sem:              # beyond the cap, we wait here
            return await coro_f(x)
    return await asyncio.gather(*(with_limit(x) for x in items),
                                return_exceptions=True)

in_flight = 0
peak = 0

async def job(x):
    global in_flight, peak
    in_flight += 1
    peak = max(peak, in_flight)
    try:
        await asyncio.sleep(0.1)
        return x
    finally:
        in_flight -= 1

async def try_semaphore():
    t0 = time.perf_counter()
    out = await map_limited(job, range(50), max_concurrent=10)
    print(len(out), "results; peak in flight:", peak,
          f"; duration {time.perf_counter() - t0:.1f}s (sequential: 5.0s)")

Timeout — no wait is infinite. Since Python 3.11 there is asyncio.timeout, a context manager more readable than wait_for. Here TimeoutError is the built-in one: since 3.11 asyncio.TimeoutError is an alias of it. The function turns the deadline into an error value readable by the model, instead of an exception that kills the run.

async def with_deadline(coro, seconds=20):
    try:
        async with asyncio.timeout(seconds):
            return await coro
    except TimeoutError:
        return {"error": f"timeout after {seconds}s"}

async def try_timeout():
    print(await with_deadline(asyncio.sleep(5, "slow"), seconds=0.2))
    print(await with_deadline(asyncio.sleep(0.05, "fast"), seconds=0.2))

Cancellation — when a timeout fires, the coroutine receives a CancelledError at the point where it was waiting. Two non-negotiable rules: do not catch CancelledError to ignore it (if you catch it to clean up, re-raise it), and free the resources in a finally. To try it you need a client that does not answer: FakeClient sleeps for five seconds and I cancel the task after 0.2.

class FakeClient:
    async def get(self, url, timeout):
        await asyncio.sleep(5)

async def traced_call(client, url):
    start = time.perf_counter()
    try:
        r = await client.get(url, timeout=10)
        return r.json()
    except asyncio.CancelledError:
        print("cancelled, cleaning up")
        raise                        # ALWAYS: otherwise you break cancellation
    finally:
        print(f"duration {time.perf_counter() - start:.1f}s")

async def try_cancellation():
    t = asyncio.create_task(traced_call(FakeClient(), "http://x"))
    await asyncio.sleep(0.2)
    t.cancel()
    try:
        await t
    except asyncio.CancelledError:
        print("the task is cancelled:", t.cancelled())

Since Python 3.11 there is also asyncio.TaskGroup, which is the recommended form when the tasks are logically a group: if one raises, the others are cancelled and the async with block does not close before they have all finished. It is safer than gather against orphan tasks; gather remains convenient when you want the errors as values.

async def download(client, u, results):
    await asyncio.sleep(0.1)
    results[u] = f"content of {u}"

async def group(urls, client):
    results = {}
    async with asyncio.TaskGroup() as tg:          # Python 3.11+
        for u in urls:
            tg.create_task(download(client, u, results))
    return results      # here all tasks have finished, guaranteed

async def try_group():
    print(await group(["a", "b", "c"], client=None))

The main that runs the five tests in sequence, printing the name of each, and the overall output that results. It is the only point where the event loop starts: everything else is definitions, and that is how a file of asynchronous tests stays readable. Concatenate the previous blocks in the order in which they appear and add this at the bottom.

async def main():
    for p in (try_calls, try_semaphore, try_timeout, try_cancellation, try_group):
        print("##", p.__name__)
        await p()

asyncio.run(main())

# ## try_calls
# [{'tool': 'stock', 'result': {'sku': 'A1', 'stock': 4}}, {'tool': 'incoming', 'error': "KeyError: 'A1'"}]
# duration 0.3s (sequential: 0.5s)
# ## try_semaphore
# 50 results; peak in flight: 10 ; duration 0.5s (sequential: 5.0s)
# ## try_timeout
# {'error': 'timeout after 0.2s'}
# fast
# ## try_cancellation
# cancelled, cleaning up
# duration 0.2s
# the task is cancelled: True
# ## try_group
# {'a': 'content of a', 'b': 'content of b', 'c': 'content of c'}

Read the output as a check for each tool. The semaphore: fifty jobs, never more than ten together, in 0.5 seconds instead of 5 (five waves of 0.1). The timeout: the slow call returns an error dictionary, the fast one goes through. The cancellation: the line "cancelled, cleaning up" appears because the CancelledError is caught and re-raised, the finally prints the duration anyway, and the task really is cancelled. The TaskGroup delivers the complete dictionary because the block does not close until the three tasks have finished. In the original code download was not defined: here it is, which is why the block now runs.

What happens between two awaits (and why it matters)

The event loop is cooperative: it switches coroutine only at await points. Between one await and the next your code runs without interruptions, as if it were alone. This property has two opposite consequences that have to be understood together.

The first is good: between two awaits no lock is needed. A counter increment, a write into a dictionary, a read-modify-write with no await in the middle: they are atomic by construction, because no other coroutine can step in. An example that shows it is already the in_flight counter in the semaphore block: it increments and decrements without any lock, and the peak observed (10) is exact.

The second is bad: if you put an await inside a sequence that has to be consistent, you have opened a window in which another coroutine can change the state under your feet. Two debits of 30 on a balance of 100 must leave 40. In the broken version, the balance is read before the fraud check's await and written after; the two debits both read 100 and both write 70. check_fraud is here a fake check that sleeps for 0.05 seconds.

import asyncio

async def check_fraud(amount):
    await asyncio.sleep(0.05)               # fake external check

balance = {"value": 100}

# BROKEN: there is an await between the read and the write
async def debit_broken(amount):
    current = balance["value"]
    await check_fraud(amount)              # here another coroutine can debit
    balance["value"] = current - amount    # writes based on a stale value

# CORRECT: the critical section is protected
lock = asyncio.Lock()

async def debit(amount):
    await check_fraud(amount)              # outside the critical section
    async with lock:
        balance["value"] -= amount

async def try_balance():
    balance["value"] = 100
    await asyncio.gather(debit_broken(30), debit_broken(30))
    print("broken :", balance["value"], "(expected 40)")
    balance["value"] = 100
    await asyncio.gather(debit(30), debit(30))
    print("correct:", balance["value"], "(expected 40)")

asyncio.run(try_balance())

# broken : 70 (expected 40)
# correct: 40 (expected 40)

The result 70 instead of 40 is a debit that vanished without any error, and it is exactly the kind of bug that does not show up in a sequential test. The correct version also does another right thing: it keeps the slow call outside the lock, and inside only the modification. A lock that wraps a network await serialises all the users on the latency of that network, that is, it destroys exactly the advantage for which you wrote asynchronous code.

The complete pattern: an agent that serves many users

Let us put all the pieces together in the form you will really use. Shared client, semaphore on the model provider, timeout per request, errors as values, and clean shutdown. The real agent loop is the _loop method, left to implement (here it raises NotImplementedError): it is the function of the previous lesson, made asynchronous. The httpx.AsyncClient is created and closed but in this example it makes no network calls.

import asyncio, time
import httpx

class AgentService:
    def __init__(self, max_concurrent=8, timeout_s=45.0):
        # a single client per process: it reuses connections and TLS handshakes
        self._client = httpx.AsyncClient(timeout=httpx.Timeout(timeout_s),
                                         limits=httpx.Limits(max_connections=50))
        self._sem = asyncio.Semaphore(max_concurrent)
        self._timeout_s = timeout_s
        self._running: set[asyncio.Task] = set()       # avoids orphan tasks

    async def _loop(self, request: str) -> str:
        raise NotImplementedError("the agent loop goes here")

    async def one_run(self, request: str) -> dict:
        t0 = time.perf_counter()
        try:
            async with self._sem:                       # cap on parallel calls
                async with asyncio.timeout(self._timeout_s):
                    text = await self._loop(request)
            return {"ok": True, "text": text, "duration_s": round(time.perf_counter() - t0, 2)}
        except TimeoutError:
            return {"ok": False, "error": "timeout", "duration_s": self._timeout_s}
        except asyncio.CancelledError:
            raise                                       # never swallow it
        except Exception as e:
            return {"ok": False, "error": f"{type(e).__name__}: {e}"}

    def start(self, request: str) -> asyncio.Task:
        t = asyncio.create_task(self.one_run(request))
        self._running.add(t)
        t.add_done_callback(self._running.discard)      # reference released when the task ends
        return t

    async def close(self):
        for t in list(self._running):
            t.cancel()
        await asyncio.gather(*self._running, return_exceptions=True)
        await self._client.aclose()

Read it in this order: one_run is the only point where a request becomes a result (never an exception towards the caller, except cancellation), start is the one where the run becomes a stored task, close the one where everything ends. To try it without a model I derived a class that replaces _loop with a fake, whose argument is the number of seconds to wait, or "fault" to make it raise an error. I granted two semaphore seats and a timeout of half a second. Run with httpx 0.28.1, twice, identical output.

class TestService(AgentService):
    async def _loop(self, request):
        if request == "fault":
            raise ValueError("tool broken")
        await asyncio.sleep(float(request))
        return f"done after {request}s"

async def try_service():
    s = TestService(max_concurrent=2, timeout_s=0.5)
    t = [s.start(r) for r in ("0.1", "fault", "2", "0.1")]
    for x in t[:2]:
        r = await x
        r.pop("duration_s", None)
        print(r)
    print(await t[2])
    print({k: v for k, v in (await t[3]).items() if k != "duration_s"}, "(waited in the queue for a slot)")
    # a long run still in progress, then shutdown
    long_run = s.start("0.4")
    await asyncio.sleep(0.05)
    print("running before close:", len(s._running))
    await s.close()
    print("running after close:", len(s._running), "- run cancelled:", long_run.cancelled())

asyncio.run(try_service())

# {'ok': True, 'text': 'done after 0.1s'}
# {'ok': False, 'error': 'ValueError: tool broken'}
# {'ok': False, 'error': 'timeout', 'duration_s': 0.5}
# {'ok': True, 'text': 'done after 0.1s'} (waited in the queue for a slot)
# running before close: 1
# running after close: 0 - run cancelled: True

The six lines tell what the code promises. The first run succeeds. The second fails with an error converted into a value, not into an exception. The third lasts 2 seconds and is stopped by the timeout at 0.5. The fourth finds both seats taken and waits in the queue until the first one finishes, then succeeds. At the end a long run is still in flight, close cancels it (run cancelled: True) and the set of running tasks goes back to empty.

Four lines deserve a comment, because they are exactly the ones people leave out. First, self._running plus add_done_callback is the canonical way of not losing a task: you keep it in a set until it finishes, then remove it. The documentation of asyncio warns that a task not referenced elsewhere can be collected by the garbage collector even before it has finished.

Second, the order of the two context managers decides what the timeout limits. With the semaphore outside and the timeout inside, as here, the timeout starts after the seat has been obtained: it limits the duration of the run, not the wait in the queue. If you want to guarantee the user a maximum total time, queue included, reverse the order: it is the time the user perceives. There is no absolutely right order, there is the one that matches the guarantee you want to give.

Third, except asyncio.CancelledError: raise comes before except Exception. Since Python 3.8 CancelledError derives from BaseException, so technically except Exception would not catch it; writing it explicitly documents the intention and protects against the case in which someone, tomorrow, changes that block to except BaseException. Fourth, close: cancelling the tasks in flight and closing the client is what distinguishes a clean shutdown from a process that leaves half-open connections and logs full of warnings.

When the user closes the page: cancellation done right

There is a case that in an agent happens all the time and that example code almost never covers: the user closes the browser tab while the run is halfway. The server receives the disconnection, the framework cancels the task, and at that point what happens depends entirely on how you wrote the coroutines. If you wrote them badly, what remains is a call to the model that carries on and that you will pay for, a lock taken and not released, a semaphore seat occupied forever.

Cancellation in asyncio works like this: the task receives a CancelledError at the point where it was waiting, as if that await had raised an exception. From there the exception propagates upwards like any other, and all the finally blocks and context managers (async with) run their cleanup. This is the mechanism that makes cancellation safe — provided you do not interfere.

import asyncio, time

class Registry:
    async def open(self, run_id): print("registry: open", run_id)
    async def close(self, run_id, reason): print("registry: close", run_id, reason)

class Metrics:
    def observe(self, name, value): print("metric", name, "recorded")

registry, metrics, semaphore = Registry(), Metrics(), asyncio.Semaphore(2)

async def agent_loop(request):
    await asyncio.sleep(float(request))
    return {"ok": True}

async def run_with_cleanup(run_id: str, request: str) -> dict:
    start = time.monotonic()
    await registry.open(run_id)                   # opens an audit row
    try:
        async with semaphore:                     # released by the context manager
            async with asyncio.timeout(45):
                return await agent_loop(request)

    except TimeoutError:
        await registry.close(run_id, "deadline")
        return {"ok": False, "error": "timeout"}

    except asyncio.CancelledError:
        # the user is gone: record and RE-RAISE
        await registry.close(run_id, "cancelled_by_user")
        raise                                     # <- without this line you break everything

    except Exception as e:
        await registry.close(run_id, f"error:{type(e).__name__}")
        raise

    finally:
        duration = time.monotonic() - start
        metrics.observe("run_duration_s", duration)  # runs in EVERY case

async def try_cleanup():
    t = asyncio.create_task(run_with_cleanup("r1", "5"))
    await asyncio.sleep(0.1)
    t.cancel()
    try:
        await t
    except asyncio.CancelledError:
        pass
    print("free slots in the semaphore:", semaphore._value, "of 2")

asyncio.run(try_cleanup())

# registry: open r1
# registry: close r1 cancelled_by_user
# metric run_duration_s recorded
# free slots in the semaphore: 2 of 2

The block uses a fake Registry, a fake Metrics and a fake semaphore, defined at the beginning, so it runs on its own (two identical runs). The output shows the three things that count after a cancellation: the audit row is closed with the right reason, the duration metric is recorded anyway, and the semaphore goes back to two free seats out of two. If after your cancellation the seats were fewer, there is a resource you are not releasing.

Three points to read carefully, because they are exactly the ones people get wrong.

The raise after catching CancelledError is not optional. Catching it and not re-raising it means telling the system "this task completed normally" when in fact it was interrupted: the caller believes everything went well, the handlers upstream do not run their cleanup, and in the worst cases a TaskGroup or a gather waits forever for something that will not arrive. The rule is absolute: you catch it only to clean up, and then you always re-raise.

The semaphore is released by the context manager, not by you. That is why async with semaphore is clearly preferable to an acquire() followed by a manual release(): with the manual form, an exception between the two lines leaves the semaphore taken, and after a few cancelled requests the system blocks on its own — a failure that shows up hours after the cause.

The finally block also runs during cancellation. It is the guarantee on which all cleanup rests, and it is also a trap: if inside the finally you put a slow await, you are slowing down the cancellation of a task that someone wants dead. In cleanup you do short things; heavy work goes elsewhere.

The check worth doing is precisely the one in the example: start a run, cancel it after a moment with task.cancel(), and look at the logs and the semaphore. If you do not see the closing line and the metric, you have resources that are not released and you did not know it.

Measuring instead of believing

Every claim about concurrency has to be verified, because asynchronous code lies easily: it looks parallel and it is not. Two tools are almost always enough. The first is the debug mode of asyncio, which reports the coroutines that occupy the event loop for too long (the default threshold is 0.1 seconds). I try it with a coroutine that calls time.sleep(0.3), that is, synchronous work inside async code.

import asyncio, time

async def blocking():
    time.sleep(0.3)                  # synchronous work inside a coroutine: stops the event loop

async def main():
    await asyncio.gather(blocking(), asyncio.sleep(0.1))

asyncio.run(main(), debug=True)

# Executing <Task finished name='Task-2' coro=<blocking() done, defined at .../g_debug.py:3> result=None created at .../asyncio/tasks.py:732> took 0.300 seconds

The message (abbreviated: the file paths are cut) names the guilty coroutine, blocking(), and how long it held the event loop still. The second tool is a real concurrency counter, five lines around the coroutine to be measured. I use it on three waits of 0.1 seconds, first one at a time then with gather.

import asyncio

in_flight = 0
peak = 0

async def instrumented(coro):
    global in_flight, peak
    in_flight += 1
    peak = max(peak, in_flight)
    try:
        return await coro
    finally:
        in_flight -= 1

async def wait_a_bit():
    await asyncio.sleep(0.1)

async def main():
    global peak
    for _ in range(3):                      # sequential: one at a time
        await instrumented(wait_a_bit())
    print("sequential, peak =", peak)
    peak = 0
    await asyncio.gather(*(instrumented(wait_a_bit()) for _ in range(3)))
    print("concurrent, peak =", peak)

asyncio.run(main())

# sequential, peak = 1
# concurrent, peak = 3

If after a hundred requests peak is 1, your code is sequential no matter how many async def you wrote. It is the check that dismantles the most common illusion, and it costs five lines.

The four questions that avoid the classic mistakes

Every time I write or reread asynchronous code for an agent, I go through these four questions. They are not a bureaucratic checklist: each corresponds to a class of bug that I have seen cost time.

1. "Is this wait really a wait, or is it computation?" An await in front of a function does not make it concurrent: if inside that function there is a loop that grinds for two seconds without ever yielding control, the event loop is stopped and all the other users wait. The symptom is characteristic and easy to recognise once you know it: the average latency is acceptable but the high percentiles are disastrous, because the requests queue up behind the block. The check is asyncio.run(main(), debug=True), which reports the coroutines that occupy the event loop for too long.

2. "Between creation and waiting, is there an await?" It is the concurrent_wrong bug seen above, and it is the most frequent of all because the code looks correct: there are the create_task calls, there is the mental gather, and the syntax is entirely asynchronous. The rule for not getting it wrong is syntactic, so it can be applied at a glance: if in a for loop there is an await on something that could have started earlier, you have written sequential code. First you create all the tasks, then you await.

3. "Who holds the reference to this task?" A task that is created and not kept can be collected by the garbage collector and vanish without running and without signalling anything. It is the worst bug of the category, because it leaves no trace: no error, no log, simply something that did not happen. The defence is mechanical: every create_task goes into a list or a set, or you use a TaskGroup, which holds the references for you and guarantees that no task outlives the block.

4. "What happens if this never answers?" Every network operation must have a timeout, and every timeout must have a defined behaviour. It is not pessimism: a tool that queries an external service will answer slowly at least once, and in the meantime it holds a connection, a seat in the semaphore and a user's patience. The question to ask is not "do I add a timeout?" but "how long, and what do I return when it fires?" — and the right answer is almost always an error value readable by the model, not an exception that kills the run.

One note worth more than all four, which ties this lesson to the one on tools: in an agent, a concurrency error does not show up as an error. You do not get a crash, you get a system slower than expected, or a tool whose result does not arrive and is silently replaced by an exception inside gather, or a result that the model interprets as data. The non-determinism of the model and the non-determinism of concurrency add up, and telling the two apart afterwards is hard. That is why the five-line measurements of the previous section — the real concurrency counter, the debug mode — are worth more than an afternoon of reasoning about the code.

Trade-offs

Four costs have to be accounted for. You gain time, not computing power: on CPU bound work asyncio makes things worse, because a function that grinds for two seconds without an await blocks the whole event loop, that is, all the other users; that goes into asyncio.to_thread or a process pool. Contamination of the code: an async function is called only from async code, and introducing asyncio halfway through a synchronous project means rewriting the chain of callers up to the boundary.

Harder debugging: stack traces cross the event loop and a task that fails without being awaited can disappear silently; the mitigations are asyncio.run(main(), debug=True) and keeping a reference to every task created. Order not guaranteed: gather preserves the order of the results, not that of completion, so if your code assumes that the first to finish is the first requested, it is broken.

The traps people really fall into

1. Calling a synchronous library inside a coroutine. A requests.get() or a time.sleep(5) in async code blocks everything for five seconds, users included. Inside async code you use async clients (for HTTP, for example httpx.AsyncClient), or you wrap the synchronous call in await asyncio.to_thread(function, arg).
2. Orphan tasks. asyncio.create_task(...) without keeping the reference: the garbage collector may collect it and the task vanishes without running. Keep the tasks in a list, or use TaskGroup.
3. A new HTTP client for every request. You throw away the connection pool and pay a TLS handshake at every call. Create one AsyncClient per process and pass it around.
4. Exceptions silenced inside gather. With return_exceptions=True errors become values: if you do not check with isinstance(x, Exception), a failure passes for a valid result and ends up in the agent's context as if it were data.
5. Nested asyncio.run. Calling it when a loop is already running (typical inside a notebook or a server) raises RuntimeError. Inside code that is already async you use await, not run.
6. Believing that asyncio makes code thread-safe. There are no threads, but there is interleaving: between two awaits the shared state can be changed by another coroutine. Critical sections must be protected with asyncio.Lock.

The six traps have the same structure: the program keeps working. The synchronous library inside a coroutine, the orphan task, the recreated client, the exception mistaken for a result and the shared state touched between two awaits do not produce a crash, they produce slowness or wrong data. That is why the two checks of the "Measuring instead of believing" section are worth more than reading the code.

What you can do now

An agent is code that waits: asyncio is the way to make those waits pay off without threads. Three lines to remember: create all the tasks before awaiting; put a semaphore and a timeout on anything that touches the network; never catch CancelledError without re-raising it. You can now choose among asyncio, processes and threads according to where a program waits, write the three variants of execution and see the difference with a stopwatch, and protect a service with a semaphore, a timeout and cancellation. In the next lesson we use all of this to run the first working agent, first with a managed SDK and then with a local model.

In the series, the previous lesson is Anatomy of an Agent: the Loop, the Instructions, the Tools, the Stop Condition.