8 AnyIO/Trio/AsyncIO Interop Patterns (No Pain)

A pragmatic field guide to mixing Python’s async stacks — without deadlocks, mystery timeouts, or framework wars.

Ship faster async Python by blending AnyIO, Trio, and asyncio safely. Eight interop patterns for task groups, timeouts, threads, testing, and shutdown.

You inherited a service that speaks asyncio, your team loves Trio nurseries, and a vendor SDK blocks like it’s 2013. Sound familiar? Good news: with a few deliberate patterns, you can run a calm, hybrid stack. Let’s make async boring — in the best way.

Pattern 1 — Put AnyIO at the boundaries

AnyIO is the compatibility layer that speaks both asyncio and Trio. Use it to standardize entry/exit points and “host” whichever backend you want per process or test run.

# main.py
import anyio

async def app():
    async with anyio.create_task_group() as tg:
        tg.start_soon(work)

async def work():
    await anyio.sleep(0.1)  # works on asyncio or Trio

if __name__ == "__main__":
    # Choose your backend on deployment or via env
    anyio.run(app, backend="asyncio")  # or backend="trio"

Why it works: Your code depends on AnyIO APIs (task groups, sockets, files), which map cleanly to either runtime. You can switch backends without rewriting business logic.

Pattern 2 — Keep nurseries/task groups central

Whether it’s Trio nurseries or AnyIO task groups, structure concurrency deliberately. Don’t free-solo create_task() everywhere.

import anyio

async def serve(listener):
    async with anyio.create_task_group() as tg:
        while True:
            stream = await listener.accept()
            tg.start_soon(handle_client, stream)

async def handle_client(stream):
    async with stream:
        await stream.send(b"hello\n")

Benefits: predictable lifetimes, inherited cancellation, and fewer “zombie tasks.” Think of a task group as a scope with cleanup guarantees baked in.

Pattern 3 — Use fail_after / move_on_after for consistent timeouts

Timeout semantics differ between libraries. AnyIO gives you a single, readable approach that works everywhere.

import anyio

async def fetch_with_timeout(op, seconds=0.5):
    with anyio.fail_after(seconds):     # cancel on expiry
        return await op()

Need soft timeouts? Swap to move_on_after() and handle the None path. Timeouts are part of your contract, not scattered magic numbers.

Pattern 4 — Bridge threads ↔ async the right way

You will need to call sync libraries. Do it safely with thread helpers. (Let’s be real: requests still sneaks in.)

import anyio

def heavy_cpu_sync(n: int) -> int:
    return sum(i*i for i in range(n))

async def compute(n: int) -> int:
    # run blocking work off the event loop
    return await anyio.to_thread.run_sync(heavy_cpu_sync, n)

# From a worker thread calling back into async:
def sync_entrypoint():
    anyio.from_thread.run(compute, 10_000_000)  # schedules into the running loop

Rules of thumb:

• Prefer to_thread.run_sync for blocking IO/CPU.
• Use from_thread.run only when you must hop back from a worker thread.
• Keep the payload small; pass big buffers by reference, not copy.

Pattern 5 — Adopt libraries via AnyIO abstractions

When picking network/file libs, prefer those that expose AnyIO ABCs (e.g., SocketStream, Listener, async file). If you must bring raw asyncio or Trio objects, wrap them at the edge.

# Edge adapter example (simplified)
import anyio
import asyncio

class AsyncioStream(anyio.abc.SocketStream):
    def __init__(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter):
        self.r, self.w = reader, writer
    async def send(self, data: bytes): self.w.write(data); await self.w.drain()
    async def receive(self, max_bytes=65536): return await self.r.read(max_bytes)
    async def aclose(self): self.w.close(); await self.w.wait_closed()

Payoff: your core logic never learns about event-loop quirks; only the boundary does.

Pattern 6 — Cancellation is a feature, design for it

Trio and AnyIO cancel by default when a scope exits. Embrace it: write idempotent cleanup and small critical sections.

import anyio

async def critical_update():
    with anyio.CancelScope(shield=True):   # finish even if parent cancels
        await write_metadata()
    await flush_metrics()                  # ok to be cancelled here

Guidelines:

• Shield only the minimum truly critical work.
• Treat cancellation like an expected exception path and test it.

Pattern 7 — Graceful shutdown with one place to coordinate

Capture signals once, cancel the root scope, and let task groups clean up.

import anyio

async def run():
    async with anyio.create_task_group() as tg:
        stop = anyio.Event()
        tg.start_soon(server, stop)
        tg.start_soon(worker, stop)
        await anyio.wait_signal("SIGTERM")   # or use anyio.abc.SignalReceiver
        stop.set()
        # cancellation cascades automatically when scope exits

anyio.run(run, backend="asyncio")

Result: no race of ad-hoc signal handlers; your runtime exits calmly, even under load.

Pattern 8 — Test once, run everywhere (pytest + AnyIO)

Write a single test that can execute on both backends. Validate semantics, not loop trivia.

# conftest.py
import pytest

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

# test_app.py
import anyio
import pytest

@pytest.mark.anyio
async def test_echo_roundtrip(backend):
    async def main():
        async with anyio.create_memory_object_stream() as (send, recv):
            await send.send("hi")
            assert await recv.receive() == "hi"

    anyio.run(main, backend=backend)

Bonus: run your CI twice (or parametrize) to catch regressions that only appear on one engine.

Mini playbook (decisions you’ll actually face)

• Which backend in prod? Choose the one your team debugs fastest. With AnyIO, switching later is not a rewrite.
• Vendor client is asyncio-only. Keep the process on backend="asyncio", wrap it behind AnyIO ABCs, and keep the rest of the code portable.
• High CPU step? to_thread.run_sync or a dedicated process pool. Measure event-loop latency; if it spikes, move more off-loop.
• Weird stuck tasks? Audit for forgotten await, unbounded queues, and streams without backpressure. Task groups make leaks visible.

Common foot-guns (and how to dodge them)

• **asyncio.run() inside an already-running loop.** Don’t. Use anyio.from_thread.run or pass the running loop’s primitives explicitly.
• Swallowing cancellation. If you except Exception: ..., remember CancelledError is special—don’t hide it accidentally.
• Random sleeps. Replace await asyncio.sleep(…) vs trio.sleep(…) with anyio.sleep(…). It reads clearer, too.
• Global clients. Create them inside a task group and close on exit; it enforces lifecycle discipline.

A tiny end-to-end example

# server_client.py
import anyio

async def echo_server(listener):
    async with listener:
        async with anyio.create_task_group() as tg:
            while True:
                stream = await listener.accept()
                tg.start_soon(handle, stream)

async def handle(stream: anyio.abc.SocketStream):
    async with stream:
        data = await stream.receive(1024)
        await stream.send(data)

async def main():
    listener = await anyio.create_tcp_listener(local_port=9000)
    async with anyio.create_task_group() as tg:
        tg.start_soon(echo_server, listener)
        await anyio.sleep(0.05)  # give server a tick
        async with await anyio.connect_tcp("127.0.0.1", 9000) as s:
            await s.send(b"hey")
            assert await s.receive(1024) == b"hey"

if __name__ == "__main__":
    anyio.run(main, backend="trio")  # flip to "asyncio" and it still passes

Runs on Trio or asyncio with no code changes. That’s the power of designing to AnyIO’s contracts.

Conclusion

Interop doesn’t have to be a headache. Put AnyIO at the edges, lean on task groups, make timeouts explicit, cross threads deliberately, and test on both backends. You’ll get predictable shutdowns, fewer leaks, and the freedom to pick the runtime that fits your team today — without painting yourself into a corner tomorrow.

CTA: Which pattern will you adopt first — timeouts with fail_after or a single task-group root? Tell me in the comments, and follow for more pragmatic Python performance posts.