Python AnyIO Portability — Core Concepts

The Problem AnyIO Solves

Python has two major async frameworks:

• asyncio — built into the standard library, the default choice
• Trio — an alternative that emphasizes structured concurrency and safety

They share the async/await syntax but have incompatible APIs for task spawning, cancellation, networking, and synchronization. Code written for one doesn’t run on the other.

AnyIO provides a unified API that works on both. It’s not a new framework — it delegates to whichever backend is currently running.

Key Abstractions

Task Groups

AnyIO’s task groups work like asyncio’s TaskGroup and Trio’s nurseries:

import anyio

async def fetch_all(urls):
    results = {}
    
    async def fetch_one(url):
        async with anyio.open_url(url) as response:
            results[url] = await response.read()
    
    async with anyio.create_task_group() as tg:
        for url in urls:
            tg.start_soon(fetch_one, url)
    
    return results

When the async with block exits, all tasks are guaranteed to be complete. If any task raises an exception, remaining tasks are cancelled.

Cancellation Scopes

AnyIO models timeouts and cancellation using scopes:

async def with_timeout():
    with anyio.fail_after(5):
        await some_slow_operation()
    
    # Or with a soft timeout (returns None instead of raising)
    with anyio.move_on_after(5):
        result = await some_slow_operation()
        return result
    return default_value

• fail_after(seconds) — raises TimeoutError if the block takes too long
• move_on_after(seconds) — silently exits the block if time runs out

Streams

AnyIO provides typed memory streams for passing data between tasks:

async def producer_consumer():
    send_stream, receive_stream = anyio.create_memory_object_stream[str]()
    
    async with anyio.create_task_group() as tg:
        async def producer():
            async with send_stream:
                for i in range(10):
                    await send_stream.send(f"item-{i}")
        
        async def consumer():
            async with receive_stream:
                async for item in receive_stream:
                    print(item)
        
        tg.start_soon(producer)
        tg.start_soon(consumer)

Networking

AnyIO wraps TCP/UDP networking in a backend-agnostic interface:

async def tcp_client():
    async with await anyio.connect_tcp("example.com", 443) as stream:
        await stream.send(b"Hello")
        response = await stream.receive(4096)

Running AnyIO Code

AnyIO auto-detects the running backend, but you can also start one explicitly:

# Run with asyncio (default)
anyio.run(main)

# Run with Trio
anyio.run(main, backend="trio")

Inside an already-running event loop, just use await normally — AnyIO detects which backend is active.

When to Use AnyIO

Use it when:

• Building a library that others will import
• You want structured concurrency patterns on asyncio (AnyIO backports Trio-style task groups)
• Your team hasn’t committed to asyncio vs Trio yet

Skip it when:

• Building an application (not a library) and you’ve chosen asyncio
• You need asyncio-specific features (like loop.run_in_executor with specific executors)
• Performance is critical and the abstraction overhead matters

Popular Libraries Using AnyIO

Several major Python libraries use AnyIO internally for portability:

• Starlette / FastAPI — ASGI framework (via AnyIO)
• httpx — HTTP client
• Encode projects — various async tools

This means when you use FastAPI, you’re already running AnyIO under the hood.

Common Misconception

“AnyIO adds significant overhead.” In practice, AnyIO’s abstraction layer is thin — it’s mostly function dispatch to the right backend. The overhead is negligible for I/O-bound code, which is the primary use case for async programming. The real cost is learning one more API, but it’s intentionally similar to both asyncio and Trio.

One thing to remember: AnyIO is a thin compatibility layer, not a new framework. It lets library authors write async code once and support both asyncio and Trio users. If you’re using FastAPI or httpx, you’re already using AnyIO.