anyio and Trio — structured concurrency

asyncio works. But Trio (and its compatibility layer anyio) offers a more principled API for concurrent Python. Worth knowing for interviews — and increasingly common in libraries.

Trio’s pitch

• Structured concurrency by default. Every task lives inside a nursery (Trio’s name for what asyncio later called TaskGroup). Tasks can’t escape; cancellation is transitive.
• Cancellation is hierarchical, deadline-based, and cleanly composable.
• No global state. No get_event_loop. Each function gets its scope explicitly.
• Better error messages. Tracebacks point at the original exception, not into asyncio internals.

The cost: Trio is its own ecosystem. Trio code doesn’t run on asyncio libraries directly.

anyio — the bridge

anyio is a compat layer: write code with one API, run on either Trio or asyncio. Used by Starlette, FastAPI (since 0.95), HTTPX, and many modern libraries.

import anyio

async def main():
    async with anyio.create_task_group() as tg:
        tg.start_soon(fetch_a)
        tg.start_soon(fetch_b)

anyio.run(main)                          # runs on asyncio by default
anyio.run(main, backend="trio")          # runs on Trio

Same code, two backends.

Trio nurseries vs asyncio TaskGroup

# Trio
async with trio.open_nursery() as nursery:
    nursery.start_soon(fetch_a)
    nursery.start_soon(fetch_b)
# Tasks finish by here. If any raised, all siblings cancelled.

# asyncio 3.11+
async with asyncio.TaskGroup() as tg:
    tg.create_task(fetch_a())
    tg.create_task(fetch_b())

Trio inspired TaskGroup directly. The semantics are intentionally identical: nothing escapes the block, exceptions become an ExceptionGroup, cancellation propagates.

Cancellation: Trio’s strong suit

import trio

async def with_deadline():
    with trio.move_on_after(10):       # cancel after 10s
        async with trio.open_nursery() as n:
            n.start_soon(slow_a)
            n.start_soon(slow_b)
        # if we get here, both finished within 10s
    # if we get here after timeout, both were cancelled, no exception

move_on_after cancels the block at the deadline but swallows the cancel — clean “best-effort, give up after N seconds” semantics.

fail_after is the same but raises TooSlowError on timeout. Pick based on whether the timeout is recoverable.

Compare to asyncio: asyncio.timeout(10) does this on 3.11+. Before 3.11 you used wait_for which has rough edges.

anyio in real code

FastAPI uses anyio under the hood:

# from FastAPI's source — runs the synchronous endpoint in a worker thread
import anyio
result = await anyio.to_thread.run_sync(sync_endpoint, ...)

For your own code:

import anyio

async def main():
    # Run a sync function in a thread, with cancellation support
    result = await anyio.to_thread.run_sync(blocking_call)

    # Run an async function from a thread (rare but useful)
    # anyio.from_thread.run(async_func) -- inside a sync context

    # Task group
    async with anyio.create_task_group() as tg:
        tg.start_soon(fetch_a)
        tg.start_soon(fetch_b)

    # Cancel scopes
    with anyio.move_on_after(5):
        await long_operation()

When to pick which

For most application code in 2024+, asyncio + TaskGroup is fine. For libraries, anyio is the right choice.

Common patterns ported

Race two operations, take the winner

import anyio

async def race(a_co, b_co):
    winner = None
    async with anyio.create_task_group() as tg:
        async def run_a():
            nonlocal winner
            winner = await a_co
            tg.cancel_scope.cancel()
        async def run_b():
            nonlocal winner
            winner = await b_co
            tg.cancel_scope.cancel()
        tg.start_soon(run_a)
        tg.start_soon(run_b)
    return winner

cancel_scope.cancel() cancels the rest of the group. Clean and obvious.

Bounded concurrency

import anyio

async def process_all(items):
    limit = anyio.Semaphore(20)
    async with anyio.create_task_group() as tg:
        for item in items:
            tg.start_soon(_process_one, item, limit)

async def _process_one(item, limit):
    async with limit:
        await process(item)

Streaming I/O

import anyio
async with anyio.open_file("data.bin", "rb") as f:
    async for chunk in f:
        process(chunk)

anyio gives async file I/O via threads — useful where async fs operations don’t exist natively.

Why this matters in interviews

Trio’s structured-concurrency thinking has won. asyncio adopted it; major frameworks now build on anyio. Being able to articulate why this matters (lifetime-bounded tasks, automatic cancellation, no leaks) signals you’ve used async at scale.

Interview angle

• “What’s the difference between Trio and asyncio?” — Trio is structured-concurrency-first: tasks live in nurseries, cancellation is transitive, no global event loop accessor. asyncio added TaskGroup in 3.11 to bring the same primitives. Trio’s API is cleaner; asyncio has the bigger ecosystem.
• “What’s anyio and when would you use it?” — compat layer that runs on either Trio or asyncio. Used by Starlette, FastAPI, HTTPX. Write a library once, users pick the backend. For app code, also fine — you get TaskGroup-equivalent on Python 3.10.
• “How does cancellation work in Trio?” — cancel scopes are hierarchical contexts. move_on_after(t) cancels the inner block at time t, swallowing the cancel (no exception). fail_after(t) raises TooSlowError. Cancellation propagates into all child tasks in any nursery within the scope.
• “Trio nursery vs asyncio TaskGroup?” — same semantics: structured concurrency, ExceptionGroup on failure, automatic cancellation of siblings. Different API surface. TaskGroup is asyncio’s adoption of Trio’s idea.
• “Why did asyncio take so long to get structured concurrency?” — asyncio was designed pre-Trio (2014). Trio (2017) introduced nurseries and showed the benefits. PEP 654 (ExceptionGroup) and asyncio.TaskGroup (3.11, 2022) ported the model. Trio influenced the language itself.