Python asyncio skill for Codex

Done well means: every task has an owner that awaits it, cancellation and timeouts reach every coroutine, concurrency is bounded, and nothing in an async def blocks the loop for more than a few milliseconds.

Before changing anything

• Read AGENTS.md and requires-python. TaskGroup and asyncio.timeout() need 3.11+; on 3.10 use anyio task groups or asyncio.wait_for instead.
• Identify the framework that owns the loop (FastAPI/Starlette, aiohttp, Django async, a worker). Never call asyncio.run() inside code that already runs in a loop.
• Check whether the project uses anyio or trio. If so, write anyio APIs, not raw asyncio.
• Run the blocking-call scanner over the code you will touch:

python3 scripts/find_blocking_calls.py src/

It parses files with ast and lists time.sleep, requests.*, subprocess.run, open(), sync DB drivers and similar calls made directly inside async def. Exit code 1 means findings. It cannot see blocking calls hidden inside your own sync helpers; follow those by hand.

Decision table

Need	Use	Avoid
Run N coroutines, fail fast if one fails	asyncio.TaskGroup	gather without handling the rest
Run N, collect every result and error	gather(*aws, return_exceptions=True) then inspect	Ignoring the exception objects in the result list
Deadline on a block of awaits	async with asyncio.timeout(5):	wait_for around each call separately
Limit concurrent requests	asyncio.Semaphore(n) or a worker pool over asyncio.Queue	Creating 10,000 tasks at once
Call blocking library code	await asyncio.to_thread(fn, *args)	Calling it directly in async def
CPU-heavy work	loop.run_in_executor(ProcessPoolExecutor(), fn, arg)	Threads (GIL) unless on free-threaded 3.13t+
Fire-and-forget background work	A long-lived TaskGroup owned by the app lifespan, or a real job queue	Bare asyncio.create_task() with no stored reference
Sync code needs one async result	asyncio.run(coro()) at the top level only	loop.run_until_complete inside running loops

Structured concurrency with TaskGroup

import asyncio
import httpx

async def fetch_all(urls: list[str], limit: int = 10) -> dict[str, int]:
    sem = asyncio.Semaphore(limit)
    results: dict[str, int] = {}

    async with httpx.AsyncClient(timeout=10) as client:
        async def fetch(url: str) -> None:
            async with sem:
                r = await client.get(url)
                results[url] = r.status_code

        async with asyncio.TaskGroup() as tg:
            for url in urls:
                tg.create_task(fetch(url))
    return results

• When any task raises, the group cancels the others and raises an ExceptionGroup. Handle it with except*:

try:
    await fetch_all(urls)
except* httpx.TimeoutException as eg:
    log.warning("timeouts: %d", len(eg.exceptions))
except* httpx.HTTPError as eg:
    raise RuntimeError("fetch failed") from eg

• One shared client per group, not one per request: connection pooling is most of the speedup.
• For thousands of items, prefer a fixed pool of workers reading from a Queue over one task per item, so memory stays flat.

Worker pool over a queue

async def worker(q: asyncio.Queue[str], out: list[str]) -> None:
    while True:
        item = await q.get()
        try:
            out.append(await process(item))
        finally:
            q.task_done()

async def run(items: list[str], n: int = 8) -> list[str]:
    q: asyncio.Queue[str] = asyncio.Queue(maxsize=n * 2)
    out: list[str] = []
    async with asyncio.TaskGroup() as tg:
        workers = [tg.create_task(worker(q, out)) for _ in range(n)]
        for item in items:
            await q.put(item)          # backpressure via maxsize
        await q.join()
        for w in workers:
            w.cancel()
    return out

On 3.13+, q.shutdown() replaces the cancel loop and makes get() raise QueueShutDown.

Cancellation rules

• asyncio.CancelledError derives from BaseException. except Exception does not catch it; a bare except: or except BaseException: does and must re-raise.
• Clean up in finally, then let cancellation propagate. Never return from a handler that caught CancelledError unless you also call asyncio.current_task().uncancel() and know why.
• asyncio.timeout() turns its internal cancellation into TimeoutError at the async with boundary. Catch TimeoutError, not CancelledError, around it.
• Cleanup that must finish even under cancellation (closing a transaction, releasing a lock in an external system): await asyncio.shield(cleanup()), and still bound it with a timeout.
• Long CPU loops inside a coroutine never see cancellation; add await asyncio.sleep(0) per batch, or move the work to an executor.

async def handler(conn) -> None:
    try:
        async with asyncio.timeout(30):
            await conn.process()
    except TimeoutError:
        await conn.send_error("timeout")
    finally:
        await conn.close()

Background tasks without leaks

The loop keeps only weak references to tasks. An unreferenced create_task() result can be garbage collected mid-flight, and its exception is only logged at GC time.

background: set[asyncio.Task[None]] = set()

def spawn(coro) -> None:
    task = asyncio.create_task(coro)
    background.add(task)
    task.add_done_callback(background.discard)

Better: create the TaskGroup in the application lifespan (FastAPI lifespan, aiohttp cleanup_ctx) and spawn into it, so shutdown waits for or cancels in-flight work.

Mixing sync and async

Direction	Pattern
async calls blocking function	await asyncio.to_thread(fn, arg); default pool is min(32, cpu+4) threads
sync code calls coroutine, no loop running	asyncio.run(coro()) once at the entry point
sync thread calls into the running loop	asyncio.run_coroutine_threadsafe(coro, loop).result(timeout=5)
Library has both APIs	Use the async one (httpx.AsyncClient, asyncpg, psycopg async, redis.asyncio)

asyncio.Lock, Event and Queue are not thread-safe. Across threads use loop.call_soon_threadsafe.

Debugging a hung or slow loop

PYTHONASYNCIODEBUG=1 python app.py      # logs callbacks slower than 100 ms and never-awaited coroutines
python -X dev app.py                    # dev mode also enables asyncio debug
python -m asyncio pstree <PID>          # 3.14+: live task tree of a running process

• asyncio.all_tasks() plus task.print_stack() in a debug endpoint shows where each task waits.
• “Task was destroyed but it is pending” means a lost reference or a loop closed with running tasks.
• “coroutine was never awaited” means a missing await; the call did nothing.
• Throughput no better than sequential: look for a shared lock, a per-request client, a blocking call, or a semaphore set to 1.

Testing

import pytest

@pytest.mark.asyncio            # pytest-asyncio; or use anyio's pytest plugin
async def test_fetch_all_limits_concurrency(httpx_mock):
    ...

• Assert timeouts with a tiny budget (asyncio.timeout(0.01)) rather than sleeping for real.
• Test cancellation explicitly: start the task, task.cancel(), await it inside pytest.raises(asyncio.CancelledError), then assert cleanup ran.

Gotchas

• gather() without return_exceptions=True raises the first error but leaves the other tasks running unsupervised.
• async for over a sync generator is a TypeError; sync iteration of a huge list inside a coroutine is legal and blocks the loop.
• DNS resolution, logging handlers writing to slow disks, and json.loads on 50 MB bodies all block. Measure before assuming the network is the bottleneck.
• asyncio.get_event_loop() in a coroutine: use asyncio.get_running_loop().
• Never share one AsyncClient or DB pool across different event loops (common in tests that create a loop per test).

Verification

• python3 scripts/find_blocking_calls.py <changed dirs> exits 0, or every remaining hit is justified (startup code, a CLI path).
• Run once with PYTHONASYNCIODEBUG=1: no slow-callback warnings above 100 ms on the hot path, no never-awaited warnings.
• A timing check proves the concurrency: N requests take close to max(latency) * ceil(N/limit), not sum(latency).
• Tests cover the failure path: one task raising cancels its siblings and surfaces the error.