Recipe

Python AnyIO primer

AnyIO is the structured-concurrency layer that lets a single Python codebase run on both asyncio and Trio backends. For Meridian agents that fan out HTTP calls, supervise child tasks, and cancel cleanly on deadline, AnyIO gives you the supervision tree without rewriting against a specific event loop.

1. Install and select a backend

Add anyio to your project. The default backend is asyncio, which keeps you compatible with the rest of the ecosystem. Switch to Trio later by passing backend="trio" to anyio.run.

2. Use task groups, not gather

A task group owns its children. If any task raises, siblings get cancelled and the exception propagates out of the async with block. This is the structured-concurrency contract: no orphaned background tasks, no swallowed errors.

3. Apply deadlines with move_on_after

Wrap risky work in anyio.move_on_after(seconds) to cap latency. When the cancel scope fires, the inner code raises a cancellation that the task group cleans up for you. Pair it with a fallback path so the caller always gets a response.

Example

import anyio

async def fetch(url: str) -> str:
    async with await anyio.connect_tcp(url, 443, tls=True) as stream:
        await stream.send(b"GET / HTTP/1.0\r\n\r\n")
        return (await stream.receive(4096)).decode()

async def main() -> None:
    async with anyio.create_task_group() as tg:
        for host in ["meridian.ai", "getnimbus.net"]:
            tg.start_soon(fetch, host)

anyio.run(main)