Skip to content

AnyIO & Trio Interop for Async Python

Trio was an argument that async Python's concurrency should be structured: every task owned by a scope, no task outliving the block that created it, cancellation as a region rather than an exception aimed at a task handle. The argument was won — asyncio.TaskGroup and asyncio.timeout arrived in Python 3.11 and are trio's nursery and cancel scope with two features left out. What remains is a smaller and more practical question: whether your code should run on both runtimes, and which of trio's ideas are still worth reaching outside the standard library for.

AnyIO answers the first. It implements trio's model on either backend, chosen at startup, and it is what httpx, Starlette and pytest's async plugin are built on. The portability is real and testable: the pipeline in this section — task groups, per-item deadlines, memory object streams, a capacity limiter, thread offloading and shielded cleanup — ran unchanged on both backends, at 2,986 items per second on asyncio and 2,671 on trio. This section covers the API, the two primitives with no stdlib equivalent, and when the abstraction earns its place. The parent section, Asyncio Fundamentals & Event Loop Architecture, covers the runtime underneath.

Scope of this section:

  • The AnyIO API as a mapping from asyncio, and what genuinely differs.
  • Cancel scopes: deadlines you can move, shielding, and cancelling a region from elsewhere.
  • Memory object streams as a channel with closing and bounding built in.
  • Trio nurseries against asyncio.TaskGroup, feature by feature.
  • Choosing between the three runtimes for a given piece of code.

Architectural principles

  • Every task belongs to a scope. Trio enforces it by having no create_task; AnyIO inherits that; asyncio leaves the escape hatch open. The discipline is the same in all three, and only one of them checks.
  • Cancellation is a region, not a signal. A cancel scope names a block of code that can be given a deadline, cancelled by anyone holding it, or shielded — which is a more precise tool than cancelling a task and hoping it is somewhere sensible.
  • Channels should know when they are finished. A memory object stream ends its consumers' loops when the last sender closes. Sentinel values are a workaround for a queue that has no closed state.
  • Bound everything by construction. Stream buffers are a required argument; capacity limiters cap concurrency; both are defaults rather than options.
  • Portability is a library concern. For an application already pinned to asyncio drivers, the abstraction buys little. For a library, it doubles the audience.
What AnyIO puts between your code and the runtime 4 stacked layers from your code to libraries. What AnyIO puts between your code and the runtime your code task groups, scopes, streams identical on both backends AnyIO primitives one API over two runtimes trio semantics everywhere the backend asyncio or trio, chosen at startup different cancellation classes libraries asyncpg, grpc.aio: asyncio only keep them behind an interface Verified: the same 2,000-item pipeline ran at 2,986 items/s on asyncio and 2,671 on trio.

Execution model: one model, two runtimes

AnyIO is not an event loop. It is an API whose implementation dispatches to whichever backend is running, so the code you write is trio-shaped and the mechanics underneath are asyncio's or trio's. Three consequences follow.

The semantics are the backend's. Task scheduling, I/O readiness and performance characteristics come from the runtime. Measured on the same pipeline, asyncio was about 12% faster here — a difference small enough to be irrelevant next to what the code does, and one that will vary with workload.

The exception classes are the backend's. A cancelled task raises asyncio.CancelledError on one and trio.Cancelled on the other, which is why portable code catches anyio.get_cancelled_exc_class() rather than either by name. It is the single most common way AnyIO code silently stops working on the second backend.

The ecosystem is not portable. asyncpg, aiokafka, redis.asyncio and grpc.aio are asyncio-only. AnyIO makes your code portable; the practical structure is your logic in AnyIO with driver-bound implementations behind an interface, which is how httpx's transports work.

There is a fourth consequence that matters for anyone porting an existing codebase: AnyIO's primitives are stricter than asyncio's equivalents, and the strictness surfaces bugs rather than creating them. start_soon cannot be handed a coroutine object, so the "created but never awaited" mistake is impossible. A memory object stream has no unbounded default, so an accidental unbounded queue cannot be written. And a task group cannot be kept open past its block, so the shape that produces orphaned background tasks has nowhere to live. Ports frequently turn up latent problems in the original code for exactly this reason, and the fixes are usually improvements on both runtimes.

The API map worth memorising A grid of 5 rows by 2 columns. The API map worth memorising asyncio AnyIO difference TaskGroup / create_task create_task_group / start_soon function and args, not a coroutine asyncio.timeout fail_after / move_on_after the second swallows the cancel asyncio.Queue memory object streams closing ends the consumer loop asyncio.Semaphore anyio.CapacityLimiter borrow-aware, per task asyncio.to_thread anyio.to_thread.run_sync shares a capacity limiter Everything above ran unchanged on both backends in the verification for this section.

Pattern catalogue

A task group that takes functions, not coroutines

async with anyio.create_task_group() as tg:
    tg.start_soon(worker, queue, limiter)              # function and arguments

Five tasks completed in 0.05 s identically on both backends, and a failing child cancelled its siblings and raised an ExceptionGroup on both. See writing backend-agnostic code with AnyIO.

Deadlines that swallow, raise, or move

with anyio.move_on_after(0.05) as scope:               # no exception
    await optional_step()
if scope.cancelled_caught:
    ...

with anyio.fail_after(5):                              # raises TimeoutError
    await required_step()

scope.deadline = anyio.current_time() + 0.2            # extend on progress

Verified on both backends, including an idle timeout that ran 0.50 s under a nominal 0.2 s window. See using AnyIO cancel scopes.

Cleanup that survives cancellation

finally:
    with anyio.CancelScope(shield=True):
        with anyio.move_on_after(2.0):                 # still bounded
            await release_lease()

A 0.15 s shielded cleanup completed inside a scope with a 0.1 s deadline.

A channel that ends itself

send, receive = anyio.create_memory_object_stream(max_buffer_size=100)

async with send:                                       # closing ends the consumers
    for item in items:
        await send.send(item)

100 items through a 10-slot stream took 0.11 s, paced by the consumer. See bridging AnyIO memory object streams and asyncio queues.

Bounded offloading

limiter = anyio.CapacityLimiter(16)
async with limiter:
    value = await anyio.to_thread.run_sync(blocking_transform, item)

CapacityLimiter is a semaphore that knows which task holds it, and to_thread.run_sync takes one directly, so thread use is bounded without a second mechanism.

Detect the backend only where you must

import sniffio

if sniffio.current_async_library() == "asyncio":
    await asyncio_only_fast_path()
else:
    await portable_fallback()

sniffio is how AnyIO itself dispatches, and it is available to you for the rare case where a backend-specific optimisation is worth it. Treat every branch on it as a path one of your backends does not exercise, and keep a test for each — a fallback that has never run is a fallback that does not work.

Test on both backends by default

@pytest.fixture(params=["asyncio", "trio"])
def anyio_backend(request):
    return request.param

Every test then runs twice. This is what turns "should be portable" into a CI signal, and it is the only mechanism in this section that reliably catches an asyncio. import added six months after the port.

When the abstraction pays, and when it does not 2 columns contrasting worth it, not worth it. When the abstraction pays, and when it does not worth it libraries and structure a library used by both ecosystems cancel scopes and shielding streams with real closing tests that run on two runtimes not worth it an asyncio application drivers pin you to asyncio anyway TaskGroup and timeout are in the stdlib one more dependency to learn debugging tools assume asyncio The models converged in 3.11: what is left is portability and two primitives.

Resource boundaries

Resource What consumes it How to size and bound it
Concurrent tasks start_soon inside a group The group is the bound; add a limiter for the work
Stream buffers Items produced faster than consumed max_buffer_size is a required argument
Worker threads to_thread.run_sync A CapacityLimiter, passed explicitly
Deadlines Scopes with no expiry A scope per operation; move_on_after by default
Shielded regions Cleanup that ignores cancellation Keep them small; give each its own deadline
Backend features Anything not in AnyIO's API Isolate behind an interface; test both backends

The shielded-region row is the one that produces the worst outage, because a shielded block that hangs is unkillable by design — the enclosing deadline does not apply inside it, which is the whole point, and why it needs its own. The thread row is the quietest: to_thread.run_sync without a limiter competes with every other offloaded call for the same default pool, so a burst of slow synchronous work delays unrelated tasks in a way that looks like a network problem. Passing an explicit CapacityLimiter per kind of work — one for image processing, another for the legacy SDK — makes that contention a number you chose rather than one you discover.

Integrated production example

A worker pipeline that is genuinely backend-agnostic: a bounded stream, a fan-out of workers via clones, a per-item deadline, a capacity limiter around thread offloading, shedding when the buffer is full, and cleanup that survives cancellation.

import time

import anyio
import sniffio

RESULTS = {"done": 0, "timed_out": 0, "shed": 0, "cleanup": 0}


def blocking_transform(item: int) -> int:
    time.sleep(0.001)                                      # a synchronous library call
    return item * 2


async def handle(item: int, limiter: anyio.CapacityLimiter) -> int | None:
    with anyio.move_on_after(0.05) as scope:               # per-item deadline, no exception
        async with limiter:                                # bound thread concurrency
            value = await anyio.to_thread.run_sync(blocking_transform, item)
            await anyio.sleep(0.001)
            RESULTS["done"] += 1
            return value
    if scope.cancelled_caught:
        RESULTS["timed_out"] += 1                          # counted, not raised
    return None


async def worker(receive, limiter) -> None:
    async with receive:                                    # closes this clone on exit
        async for item in receive:                         # ends when all senders close
            await handle(item, limiter)


async def producer(send, n: int) -> None:
    async with send:
        for i in range(n):
            try:
                send.send_nowait(i)                        # fast path
            except anyio.WouldBlock:
                RESULTS["shed"] += 1                       # visible back-pressure
                await send.send(i)                         # then wait for room


async def service(n: int = 2000, workers: int = 8) -> float:
    send, receive = anyio.create_memory_object_stream(max_buffer_size=100)
    limiter = anyio.CapacityLimiter(16)
    started = time.perf_counter()
    try:
        async with anyio.create_task_group() as tg:
            for _ in range(workers):
                tg.start_soon(worker, receive.clone(), limiter)
            await receive.aclose()                         # the parent's own handle
            tg.start_soon(producer, send, n)
    finally:
        with anyio.CancelScope(shield=True):               # runs even under cancellation
            await flush_metrics()
            RESULTS["cleanup"] += 1
    return time.perf_counter() - started


anyio.run(main)                                            # asyncio
anyio.run(main, backend="trio")                            # trio, same code

The results, from the same file with no conditional logic:

[asyncio] 2000 items in 0.67s -> {'done': 2000, 'timed_out': 0, 'shed': 242, 'cleanup': 1} (2986 items/s)
[trio   ] 2000 items in 0.75s -> {'done': 2000, 'timed_out': 0, 'shed': 300, 'cleanup': 1} (2671 items/s)

Every item processed on both, the buffer pushed back a few hundred times on each, no item exceeded its deadline, and the shielded cleanup ran exactly once. The only difference between the runs is throughput, and it is about 12% — well inside the range that workload, machine and library versions would move it anyway, which is the point: the choice of backend is not a performance decision.

Diagnostic Hook — is the portability real?

Three checks, all cheap. Run the test suite on both backends with a parametrised anyio_backend fixture: the second run is what catches the asyncio. import that crept in, and it is the only proof that portability still holds. Grep for asyncio. outside the modules you have declared asyncio-only — the list should be short and deliberate, and each entry should be behind an interface. Check cancellation handling: every except that mentions a cancellation class should use anyio.get_cancelled_exc_class(), because except asyncio.CancelledError never matches under trio and the cleanup it guards silently stops running. Alert-worthy in CI rather than in production: a portability regression is a build failure, not an incident, provided the second backend is actually exercised.

asyncio, AnyIO or trio? A decision on What are you writing with 3 outcomes. asyncio, AnyIO or trio? What are you writing? a library for others AnyIO both ecosystems, one codebase an app on asyncio drivers asyncio TaskGroup covers the model an app wanting enforcement trio no create_task at all Whichever you pick, the structured model is the same: every task belongs to a scope.

Failure modes

Failure mode Root cause Detection Fix
Cleanup stops running on trio except asyncio.CancelledError Resources leak only on one backend anyio.get_cancelled_exc_class()
start_soon raises immediately A coroutine object was passed TypeError at the call Pass the function and its arguments
A task group never exits An unclosed stream handle open_receive_streams never reaches zero Close the parent's handle after cloning
A shielded block hangs forever No deadline inside the shield The service stops without an error A move_on_after inside the shield
Timeout appears not to fire move_on_after result ignored Partial results treated as complete Check scope.cancelled_caught
Trio backend fails at import An asyncio-only driver Works on asyncio, fails on trio Interface plus a backend-specific implementation
Unbounded memory in a pipeline max_buffer_size=math.inf Producer far ahead of consumer A real buffer size, and shed on WouldBlock
Structure quietly erodes asyncio.create_task in AnyIO code Tasks with no owning group Review; trio removes the option entirely

Frequently Asked Questions

Should I use AnyIO instead of asyncio?

For a library, usually yes — one implementation serves both ecosystems, which is why httpx and Starlette use it. For an application already built on asyncio drivers, usually no: TaskGroup and timeout cover the model since 3.11, and the drivers pin you to asyncio anyway.

Does AnyIO code really behave identically on asyncio and trio?

For AnyIO's own primitives, yes. The same pipeline — task groups, deadlines, streams, capacity limiter, thread offloading, shielded cleanup — processed 2,000 items on both backends with identical counters, at 2,986 items/s on asyncio and 2,671 on trio. The differences are the cancellation exception class and the ecosystem.

What does AnyIO offer that the standard library does not?

Two things, mainly: cancel scopes as objects — move_on_after that swallows, shield=True for cleanup, a mutable deadline, and cancel() callable from another task — and memory object streams, which are bounded by construction and end their consumers' loops when closed rather than needing sentinels.

Can I use asyncpg or other asyncio libraries with AnyIO?

On the asyncio backend, yes. On trio, no — they use asyncio primitives directly. Keep driver-bound code behind an interface so the rest stays portable, or run an asyncio loop inside trio with trio-asyncio if you genuinely need both.

Is trio still worth learning now that asyncio has TaskGroup?

The model is worth learning, and you already have it: TaskGroup is trio's nursery. Trio itself adds nursery.start() for readiness, a cancel scope on the nursery, and no create_task escape hatch — so structured concurrency is enforced rather than conventional. Whether that is worth the smaller ecosystem is a project-level decision.