anyio

What it is

anyio is a structured-concurrency layer that runs identically on top of asyncio (stdlib) and trio (Nathaniel J. Smith's library). It provides task groups, cancel scopes, file/socket primitives, sync primitives, and thread-offload helpers that work the same whether the underlying loop is asyncio's or trio's. It is the runtime adapter that lets httpx, starlette, and FastAPI's testing layer work in both worlds without code duplication.

Reach for anyio when you want trio's structured-concurrency model (async with task_group: instead of asyncio.gather, scoped cancellation, no fire-and-forget tasks) without committing to trio specifically — your library will then run inside any asyncio app too. Reach for it when writing async libraries you want to be runtime-portable. Reach for it inside an asyncio app when you want better cancellation semantics than asyncio's stdlib primitives offer.

The value prop is hard to convey without async background, so the recipes below stay concrete. If you've been bitten by asyncio.gather losing exceptions or by orphaned background tasks, anyio is the fix.

Install

pip install anyio

Output: (none — exits 0 on success)

uv add anyio

Output: dependency resolved + added to pyproject.toml

poetry add anyio

Output: updated lockfile + virtualenv install

pip install "anyio[trio]"        # adds trio so you can pick trio as the backend
pip install "anyio[doc,test]"    # docs + dev test deps (rarely needed at runtime)

Output: anyio plus the named extras. The asyncio backend is included unconditionally.

Versioning & Python support

• Current 4.x line; 4.0 released in late 2023. 4.x standardized task groups and tightened cancellation semantics.
• Python 3.9+ on 4.x; older 3.x supports back to 3.8.
• Semver is honored; minor releases occasionally tighten cancel-scope behavior to match the trio model more strictly. Pin a minor range for libraries.

Package metadata

• Maintainer: Alex Grönholm (also author of apscheduler)
• Project home: github.com/agronholm/anyio
• Docs: anyio.readthedocs.io
• PyPI: pypi.org/project/anyio
• License: MIT
• First released: 2018
• Downloads: hundreds of millions per month — transitive dep of httpx, starlette, FastAPI, openai/anthropic Python SDKs, mcp, Litestar

Optional dependencies & extras

• anyio[trio] — installs trio as an optional backend. Without it, anyio still works on asyncio.
• anyio[doc], anyio[test] — docs/tests; not runtime.

Required transitively:

• sniffio — backend detection; cheap and dependency-free itself.
• idna — for DNS handling in network primitives.
• On Python 3.10 and earlier: exceptiongroup and typing_extensions polyfills for the 3.11+ stdlib features anyio depends on.

Alternatives

Real-world recipes

anyio's value lands once you see the task-group + cancel-scope pattern. The recipes below show what changes from raw asyncio.gather and what stays the same.

Recipe 1 — anyio.run to launch the loop (backend-agnostic).

import anyio

async def main():
    print("hello from", anyio.get_current_task().name)

anyio.run(main)                          # asyncio backend by default
anyio.run(main, backend="trio")          # opt into trio (requires anyio[trio])

Output: hello from main on whichever backend you choose. anyio.run is the universal entry point — identical to asyncio.run on the asyncio backend, identical to trio.run on trio.

Recipe 2 — Task group ("nursery") for structured concurrency.

import anyio

async def worker(n: int) -> None:
    await anyio.sleep(0.1)
    print(f"worker {n}")

async def main():
    async with anyio.create_task_group() as tg:
        for i in range(3):
            tg.start_soon(worker, i)
    print("all done")

anyio.run(main)

Output: worker 0 / worker 1 / worker 2 / all done (worker order non-deterministic). The async with block exits only when every spawned task finishes. If any task raises, the group cancels the others and re-raises an ExceptionGroup — no orphaned tasks, no swallowed errors.

Recipe 3 — CancelScope for explicit timeouts.

import anyio

async def slow():
    await anyio.sleep(10)

async def main():
    with anyio.move_on_after(0.5):       # cancels the block after 0.5 s, no exception
        await slow()
        print("won't print")
    print("continuing")

    try:
        with anyio.fail_after(0.5):       # raises TimeoutError on expiry
            await slow()
    except TimeoutError:
        print("timed out")

anyio.run(main)

Output: continuing then timed out. move_on_after and fail_after are the two timeout idioms — silent vs noisy. Cancel scopes nest cleanly; the inner scope's expiry doesn't tear down the outer block.

Recipe 4 — Bounded concurrency with Semaphore.

import anyio

async def fetch(url: str, sem: anyio.Semaphore):
    async with sem:
        await anyio.sleep(0.2)
        print(f"got {url}")

async def main():
    sem = anyio.Semaphore(3)
    urls = [f"https://example.com/{i}" for i in range(10)]
    async with anyio.create_task_group() as tg:
        for u in urls:
            tg.start_soon(fetch, u, sem)

anyio.run(main)

Output: at most three got ... lines print per ~200 ms tick. anyio.Semaphore is the structured-concurrency-friendly replacement for asyncio.Semaphore.

Recipe 5 — The same code on both backends (the actual value prop).

import anyio, sniffio

async def report():
    print("running on", sniffio.current_async_library())
    async with anyio.create_task_group() as tg:
        async def hello(): print("hello!")
        tg.start_soon(hello)

anyio.run(report)                        # asyncio
anyio.run(report, backend="trio")        # trio (needs anyio[trio])

Output:

• First run: running on asyncio / hello!
• Second run: running on trio / hello!

The same report() function works on both. If you're writing a library — httpx, an SDK, a queue worker — using anyio means trio users and asyncio users both consume your library without forks.

Recipe 6 — Offload a blocking call to a worker thread.

import anyio, time

def blocking(n: int) -> int:
    time.sleep(0.5)
    return n * 2

async def main():
    result = await anyio.to_thread.run_sync(blocking, 21)
    print(result)

anyio.run(main)

Output: 42 after ~500 ms — the blocking call ran in a thread without blocking the event loop. The reverse direction (anyio.from_thread.run) lets a regular thread call back into the async loop.

Performance tuning

anyio adds a thin shim over the underlying backend. Overhead is negligible for I/O-bound code; for tight in-process loops, the backend (asyncio vs trio) matters more than anyio itself.

• Use the asyncio backend unless you have a reason. asyncio is faster on most workloads, plus uvloop drop-in replacements are still available; trio's strength is correctness, not raw throughput.
• Don't over-spawn. tg.start_soon is cheap, but each task carries a stack. Hundreds-of-thousands of tasks slow scheduling. Use semaphores to bound parallelism.
• anyio.Lock vs threading.Lock — async locks are not thread-safe and shouldn't be reused across threads. Use the right primitive.
• anyio.to_thread.run_sync(limiter=...) lets you cap worker-thread parallelism — useful when calling a library that doesn't release the GIL well.
• anyio.run(main, backend_options={"use_uvloop": True}) uses uvloop on the asyncio backend (uvloop must be installed). 2-4× throughput boost for socket-heavy work.

Version migration guide

• 3.x → 4.0 — major. current_task returns an anyio.TaskInfo with .name, not a backend-specific object. BlockingPortal API stabilized. anyio.run(main, args) no longer auto-unpacks positional args from the call — pass them after main.
• 4.0 → 4.2 — move_on_after/fail_after accept negative timeouts (zero-duration scope). Convenience for "cancel immediately if not done".
• 4.2 → 4.3 — TCP/Unix listener APIs tightened; some keyword-only changes for connect_tcp.
• 4.3 → 4.4 — async context manager re-entry tightened to match trio's expectations.

# Before (3.x)
import anyio
async def main(a, b):
    print(a, b)
anyio.run(main, "x", "y")                # positional args after main

# After (4.0+)
import anyio, functools
anyio.run(functools.partial(main, "x", "y"))    # wrap into a zero-arg callable

Output: same result; the explicit partial is the new norm.

Production deployment notes

• Pick the backend at the application boundary, not in libraries. Libraries should use anyio and never call asyncio.run or trio.run directly. Applications choose with anyio.run(main, backend="asyncio").
• Don't mix backends. A process picks one backend per run. Library code that imports asyncio.create_task will break on the trio backend — keep library code on anyio primitives only.
• Cancel scopes propagate. A SIGTERM should call a top-level CancelScope cancel; tasks tear down cleanly. Hook into your framework's lifespan.
• Watch ExceptionGroup. Task-group errors come back as ExceptionGroup (or BaseExceptionGroup) — make sure your logging path stringifies them. Sentry/Honeybadger handle this; bespoke loggers may not.
• BlockingPortal for sync code that needs to call async. Use sparingly — it's a thread that runs an event loop. Easy to leak.
• uvloop only on Linux/macOS. It's not Windows-supported; gate with sys.platform != "win32".

Security considerations

anyio itself has a small attack surface, but the surrounding async code introduces a few sharp edges.

• Cancellation is cooperative. A misbehaving coroutine can ignore cancellation by catching BaseException or holding a sync call. Use anyio.to_thread.run_sync(cancellable=True) for long sync calls you need to interrupt; native blocking calls can't be cancelled.
• Don't catch BaseException indiscriminately. That includes CancelledError/Cancelled — catching it pins the task open and breaks structured concurrency.
• anyio.from_thread.run runs code on the event loop from a thread; treat it like an internal API. Don't expose it across trust boundaries.
• Resource-limit denial of service. Unbounded tg.start_soon accepts arbitrary spawn requests. Bound with Semaphore or a pre-allocated pool.
• TLS via anyio.connect_tcp(..., ssl_context=...) — pass your own ssl.SSLContext for cert validation. Defaults follow certifi's bundle; for private CAs configure explicitly.
• sniffio integrity. Library code uses sniffio.current_async_library() to detect the backend; spoofing that string can confuse adapters but typically isn't a real security issue.

Testing & CI integration

# pip install anyio pytest pytest-anyio
import pytest, anyio

@pytest.mark.anyio
async def test_concurrent_workers():
    results: list[int] = []

    async def add(n):
        await anyio.sleep(0.01)
        results.append(n)

    async with anyio.create_task_group() as tg:
        for i in range(5):
            tg.start_soon(add, i)

    assert sorted(results) == [0, 1, 2, 3, 4]

@pytest.fixture
def anyio_backend():
    return "asyncio"        # or parametrize across both

Output: test passes. The anyio_backend fixture controls which backend the test runs on; parametrize across ["asyncio", "trio"] to verify library code on both.

pytest-anyio is the canonical test plugin; for a one-shot script you can call anyio.run(test_main) and call it a day.

Ecosystem integrations

anyio is one of the most widely depended-on async libraries in the Python ecosystem.

• httpx — uses anyio for socket and timeout primitives.
• starlette / FastAPI — anyio runs the threadpool for sync route handlers.
• Litestar — uses anyio for cancel scopes and task groups.
• openai / anthropic / google-genai Python SDKs — all use anyio under the hood for the async client.
• mcp (Model Context Protocol) — anyio for the transport runtime.
• asyncpg-pool, several DB pools — anyio task groups for connection management.
• hypercorn — ASGI server with native anyio support.
• watchfiles — async filesystem watcher built on anyio.

Troubleshooting common errors

When NOT to use this

• Pure asyncio app, no library aspirations. stdlib asyncio.TaskGroup (3.11+) and asyncio.timeout (3.11+) cover most of what anyio gives you; the extra dep may not pay off.
• Pure trio app. Use trio directly; anyio adds a shim.
• Sync-only code. Don't introduce async machinery for problems that don't need it.
• Hard-real-time or scheduler-tight inner loop. anyio's overhead is small but non-zero. Drop to raw asyncio (or even threads) when measured.
• You can't change the entry point. anyio's strength is owning the loop start; if a framework already runs the loop, you can still use anyio primitives — but the structured-concurrency story is best when anyio runs the whole show.

Compatibility matrix

See also

• Concept: async — async/await mental model
• Python: asyncio — stdlib backend anyio sits on top of
• Packages: pip-httpx — the highest-profile anyio consumer