PyPIWeb Backendupdated 05 Aug 2026

anyio

Item: anyio
Rating: 4.3
Author: Keyoor

AnyIO is an async concurrency and networking layer that runs the same code on either asyncio or Trio. It brings Trio-style structured concurrency to asyncio: task groups that own their children, cancel scopes with timeouts, memory object streams for passing data between tasks, worker threads, async file I/O, and a pytest plugin for async tests. Most people have it installed without choosing it, because httpx, Starlette, and FastAPI are built on top of it; its quarter-billion weekly downloads are mostly transitive.

anyio · v4.14.2 · 259.5M/wk · ★ 2,519 · pushed 04 Aug 2026 · agronholm/anyio · docs · reviewed by Keyoor

Verdict

The right foundation for libraries that must serve both asyncio and Trio, and a genuinely better concurrency model than raw asyncio if your team commits to it. For asyncio-only applications on modern Python, stdlib TaskGroup plus asyncio.timeout covers most of the same ground with one less dependency.

API stability4/5The 4.x line has been current since late 2023 with additive releases; the 3-to-4 major did break things (exception groups everywhere, memory stream typing), so majors are real migrations even if minors are calm.

Docs4/5anyio.readthedocs.io has proper narrative docs (tasks, cancellation, streams, testing) plus API reference and a versionhistory page worth reading; smaller corners like blocking portals are thinner and rely on the API reference.

Maintenance4/5Pushed August 2026 with steady releases (4.14.2 current) and responsive triage, but it rests largely on one maintainer, Alex Gronholm, with 104 open issues and PRs.

Ecosystem5/5About 259M weekly downloads because httpx, Starlette, FastAPI, and the OpenAI SDK stack sit on it; being infrastructure for the modern Python web stack means examples and third-party knowledge are everywhere.

Use it if

You are writing a library and want it to work for both asyncio and Trio users without maintaining two code paths; this is exactly why httpx and Starlette use it
You want structured concurrency on asyncio: task groups that cannot leak background tasks, plus cancel scopes and timeouts that compose, on any Python from 3.10 up
You need its pytest plugin: async test functions and async fixtures that can be parametrized to run against both backends
You need primitives asyncio lacks or does awkwardly: memory object streams with backpressure, CapacityLimiter, happy-eyeballs TCP connects, and blocking-portal bridges between sync and async code

Skip it if

You are writing an application that will only ever run on asyncio with Python 3.11+: stdlib asyncio now has TaskGroup and asyncio.timeout, which cover the core patterns without an extra dependency and a second set of semantics
You are all-in on Trio: using Trio directly gives you the original design without a translation layer
Your team is not ready for exception groups: every task group failure surfaces as an ExceptionGroup, so existing except SomeError handlers silently stop matching and you need except* (3.11+) or the exceptiongroup helpers
You depend on cancellation behaving like vanilla asyncio: AnyIO uses Trio-style level cancellation inside its scopes, and mixing it with raw asyncio task code can produce surprising cancel timing
Bus factor matters to you: it is effectively a single-maintainer project (with sponsorship), carrying 104 open issues and PRs

Setup reality

pip install anyio gets you the asyncio backend; Trio support is an extra (pip install anyio[trio], needing trio>=0.32). Python 3.10 is the floor, and on 3.10 it pulls the exceptiongroup backport since except* syntax only exists from 3.11. The pytest plugin activates automatically and will fight pytest-asyncio if both try to collect the same async tests, so pick one or scope them by marker. The real setup cost is conceptual: cancel scopes, checkpoints, and ExceptionGroup handling are a different mental model from create_task-and-forget asyncio, and half-migrating a codebase leaves you debugging both models at once.

Patterns

Run concurrent tasks in a task grouprun-task-group

import anyio

async def fetch(name: str) -> None:
    await anyio.sleep(1)
    print(f"done {name}")

async def main() -> None:
    async with anyio.create_task_group() as tg:
        tg.start_soon(fetch, "a")
        tg.start_soon(fetch, "b")
    # exiting the block waits for all tasks

anyio.run(main)

If any task raises, the others are cancelled and the error surfaces as an ExceptionGroup; keyword arguments cannot be passed to start_soon, use functools.partial.

Apply timeouts with cancel scopestimeouts-and-cancel-scopes

import anyio

async def main() -> None:
    with anyio.move_on_after(2) as scope:
        await slow_operation()
    if scope.cancelled_caught:
        print("timed out, moving on")

    with anyio.fail_after(2):  # raises TimeoutError instead
        await slow_operation()

These are sync context managers (with, not async with). Cancellation only lands at await points, so pure CPU loops will not be interrupted.

Catch errors from a task grouphandle-exception-groups

import anyio

async def main() -> None:
    try:
        async with anyio.create_task_group() as tg:
            tg.start_soon(might_fail)
            tg.start_soon(might_also_fail)
    except* ValueError as eg:      # Python 3.11+
        for exc in eg.exceptions:
            print("task failed:", exc)

A plain except ValueError will NOT match; failures arrive wrapped in ExceptionGroup. On Python 3.10 use exceptiongroup.catch() since except* is 3.11+ syntax.

Pass items between tasks with backpressurememory-object-streams

import anyio

async def producer(send) -> None:
    async with send:
        for i in range(5):
            await send.send(i)

async def main() -> None:
    send, receive = anyio.create_memory_object_stream[int](max_buffer_size=10)
    async with anyio.create_task_group() as tg:
        tg.start_soon(producer, send)
        async with receive:
            async for item in receive:
                print(item)

anyio.run(main)

Default max_buffer_size is 0, meaning every send blocks until someone receives. Close the send stream (async with) or the receiving loop never ends.

Run blocking code in a worker threadrun-blocking-in-thread

import anyio
import functools

async def main() -> None:
    data = await anyio.to_thread.run_sync(read_big_file, "dump.bin")
    # keyword args need partial:
    result = await anyio.to_thread.run_sync(
        functools.partial(requests_get, timeout=10), url
    )

Thread count is capped by a default CapacityLimiter of 40; pass limiter= for your own. The thread is not cancellable mid-call, only before it starts or at return.

Call back into the event loop from a threadcall-async-from-thread

import anyio

def sync_worker() -> None:
    # runs inside to_thread.run_sync
    anyio.from_thread.run(notify_progress, 0.5)  # await an async fn
    anyio.from_thread.run_sync(update_counter)   # call a sync fn in the loop

async def main() -> None:
    await anyio.to_thread.run_sync(sync_worker)

from_thread only works in threads that AnyIO spawned; for foreign threads use anyio.from_thread.start_blocking_portal instead.

Serve and connect over TCPtcp-server-and-client

import anyio

async def handle(client) -> None:
    async with client:
        async for chunk in client:
            await client.send(chunk)  # echo

async def server() -> None:
    listener = await anyio.create_tcp_listener(local_port=9000)
    await listener.serve(handle)

async def client() -> None:
    async with await anyio.connect_tcp("127.0.0.1", 9000) as stream:
        await stream.send(b"hello")
        print(await stream.receive())

connect_tcp with a hostname uses happy-eyeballs across IPv4/IPv6. listener.serve() runs forever; put it in a task group with a cancel scope to stop it.

Start a task and wait until it signals readinesswait-until-task-ready

import anyio
from anyio.abc import TaskStatus

async def serve(port: int, *, task_status: TaskStatus = anyio.TASK_STATUS_IGNORED):
    listener = await anyio.create_tcp_listener(local_port=port)
    task_status.started(listener)
    await listener.serve(handle)

async def main() -> None:
    async with anyio.create_task_group() as tg:
        listener = await tg.start(serve, 9000)
        # here the port is guaranteed to be bound

tg.start() blocks until the task calls task_status.started(); if the task raises first, the exception propagates to the caller instead of a half-started service.

Protect cleanup from cancellationshield-cleanup

import anyio

async def worker(conn) -> None:
    try:
        await do_work(conn)
    finally:
        with anyio.CancelScope(shield=True):
            await conn.aclose()  # runs even during cancellation

Without the shield, an await in a finally block inside a cancelled scope is itself cancelled immediately, so connections leak; shield sparingly and keep the block short.

Test async code on both backendspytest-async-tests

import pytest

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

@pytest.mark.anyio
async def test_fetch():
    result = await fetch("a")
    assert result is not None

Without the fixture override, tests run on asyncio only. If pytest-asyncio is also installed, configure it to strict mode or the two plugins will both claim async tests.

Read and write files without blocking the loopasync-file-io

import anyio

async def main() -> None:
    path = anyio.Path("report.txt")
    await path.write_text("hello\n")
    text = await path.read_text()
    async with await anyio.open_file("big.log") as f:
        async for line in f:
            process(line)

These wrap blocking I/O in worker threads, so they add overhead per call; batch small reads instead of awaiting per byte.

Alternatives

Package	Registry	Pick it when
trio	PyPI	You control the whole application and want the original structured concurrency design rather than a compatibility layer.
uvloop	PyPI	You are staying on plain asyncio and just want a faster event loop, not different concurrency semantics.
pytest-asyncio	PyPI	You only need async test support for asyncio-only code and none of AnyIO's runtime features.