Skip to content

Writing Backend-Agnostic Code with AnyIO

AnyIO is an implementation of trio's structured concurrency that runs on either asyncio or trio, chosen at startup. For an application that is already committed to asyncio, that portability is usually not worth an extra layer. For a library, it is the difference between being usable by half the async Python ecosystem and all of it — which is why httpx, Starlette and pytest's async plugin are built on it. The portability claim is also testable: the task-group, sleep, timeout and error-propagation code below ran unchanged on both backends, with identical timings and one visible difference — the class of the cancellation exception.

Prerequisites

  • Python 3.11+ with anyio (pip install anyio), plus trio to exercise the second backend; measurements use anyio 4.15 and trio 0.34.
  • Structured concurrency from Exception Groups & TaskGroups — AnyIO's task groups are the same model.
  • Cancellation semantics from Cancellation Patterns.
The asyncio API and its AnyIO equivalent A grid of 5 rows by 2 columns. The asyncio API and its AnyIO equivalent asyncio AnyIO note asyncio.TaskGroup() anyio.create_task_group() start_soon, not create_task asyncio.sleep() anyio.sleep() identical semantics asyncio.timeout() anyio.fail_after() move_on_after to swallow it asyncio.Queue memory object streams bounded by construction asyncio.to_thread() anyio.to_thread.run_sync() capacity limiter included Verified: the same task-group, sleep and timeout code ran unchanged on asyncio and on trio.

1. Learn the five substitutions

Most porting work is mechanical:

import anyio

await anyio.sleep(0.1)                                 # asyncio.sleep

async with anyio.create_task_group() as tg:            # asyncio.TaskGroup
    tg.start_soon(worker, arg)                         # create_task(worker(arg))

with anyio.fail_after(5):                              # asyncio.timeout
    await slow()

with anyio.move_on_after(5) as scope:                  # timeout that does not raise
    await slow()
if scope.cancelled_caught:
    ...

result = await anyio.to_thread.run_sync(blocking_call, arg)   # asyncio.to_thread

Two differences in spelling matter. start_soon takes the function and its arguments, not a coroutine object — tg.start_soon(worker, arg), never tg.start_soon(worker(arg)) — which prevents the "coroutine was never awaited" class of mistake entirely. And AnyIO's scopes are synchronous context managers (with, not async with), because entering them does not await anything.

Verified on both backends: five tasks in a task group completed in 0.05 s, and move_on_after(0.1) around a five-second sleep returned with cancelled_caught=True after 0.10 s, identically.

Verify: the same module imports and runs under anyio.run(main) and anyio.run(main, backend="trio").

2. Get the same structured behaviour everywhere

A failing child cancels its siblings and the group raises an ExceptionGroup — on both backends:

[asyncio] a failing child -> ExceptionGroup with ['ValueError']; siblings: [(2, 'CancelledError'), (1, 'CancelledError')]
[trio]    a failing child -> ExceptionGroup with ['ValueError']; siblings: [(1, 'Cancelled'), (2, 'Cancelled')]

The semantics are identical; only the cancellation exception class differs — asyncio.CancelledError against trio.Cancelled. That is the one place backend-agnostic code has to be careful:

try:
    await work()
except anyio.get_cancelled_exc_class():                # the portable spelling
    await cleanup()
    raise

anyio.get_cancelled_exc_class() returns whichever class the current backend uses. Catching asyncio.CancelledError by name in AnyIO code silently stops working under trio, and the failure is a cleanup block that never runs.

Verify: cancellation cleanup runs on both backends, tested explicitly.

What actually differs between the backends 2 columns contrasting the same on both, different underneath. What actually differs between the backends the same on both your code task groups and scopes sleeps and timeouts streams and synchronisation thread offloading different underneath the runtime CancelledError vs trio.Cancelled the ecosystem of libraries debugging tools and profilers performance characteristics Measured: identical code, identical timings, different cancellation exception classes.

3. Test on both backends, always

Portability that is not tested is portability that has already broken. pytest-anyio — bundled with AnyIO — parametrises the backend:

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


@pytest.mark.anyio
async def test_worker_cancellation():
    with anyio.move_on_after(0.1) as scope:
        await run_worker()
    assert scope.cancelled_caught

Every test now runs twice, and the second run is what catches the asyncio. import that crept in. For a library this is the single most valuable thing in the AnyIO toolkit: it converts "should work on trio" into a CI signal.

Detecting the backend at runtime is occasionally necessary for a fallback path, and sniffio — which AnyIO uses internally — is the way:

import sniffio
if sniffio.current_async_library() == "asyncio":
    ...                                                # an asyncio-only optimisation

Use it sparingly; every branch on it is a path one of your backends does not exercise.

Verify: the suite passes under both anyio_backend values in CI.

Porting a module to AnyIO 5 ordered steps. Porting a module to AnyIO swap the primitives sleep, locks, events task group instead of gather start_soon per child fail_after / move_on_after instead of asyncio.timeout run the tests on both anyio_backend parametrised isolate library calls the parts that stay asyncio The tests running on both backends are what proves the port, and what keeps it true.

4. Isolate the parts that cannot be portable

The limit of the abstraction is the ecosystem. asyncpg, aiokafka, redis.asyncio, grpc.aio and most database drivers are asyncio-only — they use asyncio.get_event_loop() and asyncio futures directly, and no shim makes them run on trio.

The practical structure is a port at the boundary:

class UserStore(Protocol):
    async def get(self, user_id: int) -> User: ...


class AsyncpgUserStore:                                # asyncio only
    async def get(self, user_id: int) -> User: ...

Your logic stays AnyIO-portable and testable on both backends; the driver-bound implementation is one class that trio users replace. Libraries that do this well — httpx is the canonical example, with its pluggable transports — end up usable everywhere without pretending the ecosystem is portable.

Where a genuinely asyncio-only call must run inside trio, trio-asyncio runs an asyncio loop inside trio, at the cost of a second runtime and its own bridging rules. It works, and it is a bigger commitment than replacing the dependency.

Verify: the modules that import asyncio directly are a short, deliberate list.

5. Decide whether you need it at all

Three cases, with different answers:

  • A library others embed. AnyIO, almost always. It costs you little and doubles your addressable users, and its structured primitives are better than raw asyncio regardless.
  • An application built on asyncio libraries. Plain asyncio. The portability buys nothing when asyncpg and redis.asyncio already pin you, and asyncio.TaskGroup and asyncio.timeout cover most of what AnyIO adds.
  • An application that wants trio's semantics. AnyIO or trio directly. The value here is not portability but the model: no un-owned tasks, cancel scopes that compose, and cancellation that cannot be silently swallowed.

The honest summary is that AnyIO's structured concurrency reached the standard library in 3.11 — TaskGroup and timeout are the same ideas — so the remaining reasons to adopt it are library portability and the parts that have no stdlib equivalent, chiefly cancel scopes and memory object streams.

Verify: you can state which of the three cases you are in, and why.

Is AnyIO the right choice here? A decision on What are you building with 3 outcomes. Is AnyIO the right choice here? What are you building? a library others embed AnyIO works in both worlds an app on asyncio libraries plain asyncio a layer you do not need an app wanting trio semantics AnyIO or trio structured concurrency by default AnyIO’s strongest case is a library: one implementation that trio users can also adopt.

Verification

Backend-agnostic code is genuinely portable when:

  • Nothing imports asyncio directly outside an explicitly asyncio-only module.
  • Cancellation is caught via anyio.get_cancelled_exc_class().
  • Tests run on both backends in CI, not just locally.
  • start_soon takes a function and arguments, never a coroutine object.
  • Driver-bound code sits behind an interface.
  • Backend detection is rare and each branch is tested.

Pitfalls & edge cases

  • tg.start_soon(worker(arg)). Passing a coroutine object raises; pass the function and its arguments.
  • except asyncio.CancelledError in portable code. Never matches under trio, so cleanup silently stops running.
  • Mixing asyncio.sleep and anyio.sleep. The former fails under trio; keep the import list clean.
  • Assuming library portability. Most async drivers are asyncio-only; check before promising trio support.
  • anyio.run() inside a running loop. Like asyncio.run, it wants to own the runtime.
  • Testing only on asyncio. The second backend is the entire point of the abstraction.

Frequently Asked Questions

What is AnyIO and why would I use it?

An implementation of trio-style structured concurrency that runs on either asyncio or trio, selected at startup. For libraries it means one codebase serving both ecosystems — which is why httpx, Starlette and pytest's async plugin use it. For applications already tied to asyncio drivers, it mostly adds a layer.

Does AnyIO code really run unchanged on asyncio and trio?

Yes for AnyIO's own primitives. The same task-group, sleep, timeout and error-propagation code ran identically on both in testing — five tasks in 0.05 s, move_on_after firing at 0.10 s, and an ExceptionGroup containing the same ValueError. The visible difference was the cancellation exception class.

How do I catch cancellation in backend-agnostic code?

Use except anyio.get_cancelled_exc_class(), which resolves to asyncio.CancelledError or trio.Cancelled depending on the backend. Catching asyncio.CancelledError by name works on asyncio and silently never matches under trio, so cleanup blocks stop running.

Can I use asyncpg or redis.asyncio with AnyIO?

On the asyncio backend, yes — they are ordinary asyncio libraries. On trio, no: they use asyncio primitives directly. Keep driver-bound code behind an interface so your logic stays portable, or run an asyncio loop inside trio with trio-asyncio if you must.

Is AnyIO still worth it now that asyncio has TaskGroup and timeout?

For applications, often not — the standard library covers the main patterns since 3.11. For libraries it still is, because it makes them usable by trio users, and it provides cancel scopes and memory object streams, which have no direct stdlib equivalent.