anyio review
AnyIO gives Python libraries one structured-concurrency API over asyncio and Trio. Its task groups keep child tasks inside a lifetime you can reason about, while cancel scopes, object streams, socket helpers, worker threads, subprocesses, async files, and a pytest plugin fill in the rest of the runtime layer. Version 4.14 added task handles, TaskGroup.create_task() for easier asyncio migration, custom capacity limiters for file operations, and async itertools. The 4.14.2 patch fixes a process worker deadlock caused by a full stderr pipe, a CapacityLimiter token race, Unicode hostname certificate matching, and cancellation-related CPU spin. Our Python 3.12 import completed successfully and the installed package included typing metadata.
Install AnyIO deliberately when backend-neutral library code or its structured task ownership solves a problem you have. An asyncio-only application on current Python should first see whether the standard library already covers its task groups and timeouts.
We installed it
| Install | ✓ · 0.2s | 3 packages on disk · 2 MB |
| Import | ✓ | import anyio in 0.28s · pure Python · py.typed · requires Python >=3.10 |
| Known vulns | 0 | (pip-audit) |
Answers from our run
Does anyio install cleanly?
Yes. In a fresh container with an empty cache, pip install anyio finished in 0.2s, leaving 3 packages and 2 MB on disk. pip-audit reported no known vulnerabilities.
What does anyio need to run?
Python >=3.10, and nothing compiled: it is pure Python. In our run import anyio succeeded in 0.28s, and the package ships py.typed for type checkers.
anyio or trio: which should you use?
trio: Use it when you control the application and want Trio's native structured-concurrency model without an asyncio backend. Install AnyIO deliberately when backend-neutral library code or its structured task ownership solves a problem you have.
When should you not use anyio?
Your application is asyncio-only and Python 3.11 or newer; asyncio.TaskGroup and asyncio.timeout may cover the concurrency structure you need without another API layer
Use it if
- You maintain a library that must run unchanged on asyncio and Trio, including its tests
- You want task groups, composable cancellation, and bounded concurrency to govern background work instead of keeping loose asyncio.Task references
- Your service needs one set of stream, socket, subprocess, file, and sync-to-async bridge APIs across supported backends
- You want pytest fixtures and test functions to run under more than one async backend
- Your application is asyncio-only and Python 3.11 or newer; asyncio.TaskGroup and asyncio.timeout may cover the concurrency structure you need without another API layer
- Your application is Trio-only and does not publish a reusable library; direct Trio code avoids backend-neutral naming and compatibility behavior
- Your exception handling assumes one task raises one ordinary exception; concurrent task failures leave a task group as an exception group and need except* handling on Python 3.11 or newer
- You mix raw asyncio cancellation with AnyIO cancel scopes without defining ownership; the documentation warns that cancel scopes must be exited in stack order and by the task that entered them
- You expect async file methods to make disk I/O nonblocking at the operating-system level; AnyIO runs those operations in worker threads and each call consumes limiter capacity
Setup reality
Our fresh Python 3.12 sandbox installed AnyIO 4.14.2 in 0.2 seconds. The environment held 3 packages and used 2 MB afterward. pip-audit found 0 known vulnerabilities. The package declares 4 direct dependencies, requires Python 3.10 or newer, is pure Python, and ships py.typed. import anyio worked in 0.28 seconds. PyPI did not provide a license value for this release, so treat the package metadata license as unknown and check the repository terms if your policy requires a machine-readable declaration.
The default install is enough for asyncio. Running on Trio requires the trio extra and a backend selection such as anyio.run(main, backend="trio"). There are no credentials or application config files. The pytest plugin arrives with the package; mark tests with pytest.mark.anyio or set anyio_mode in pytest configuration. If pytest-asyncio also runs in auto mode, both plugins can try to own the same async tests, so keep pytest-asyncio in strict mode or choose one plugin.
Task groups wait for their children and combine concurrent failures into an exception group. A plain except ValueError will not catch a ValueError nested inside that group. Cancellation is delivered at checkpoints, and an await inside an already-cancelled scope can be cancelled immediately. Cleanup that must await usually needs a short shielded scope. Cancel scopes also rely on strict nesting; manually entering one scope and exiting it from another task is unsupported.
File and path helpers use worker threads. to_thread.run_sync() also has a capacity limiter, so blocking calls can queue when the pool is full. Cancelling the waiting coroutine does not stop arbitrary synchronous code already executing in its thread. Process workers have separate serialization and startup costs; 4.14.2 prevents a worker from deadlocking when its stderr output fills a pipe, but it still discards worker stderr as documented. Subinterpreter support needs Python 3.13 or newer and follows the interpreter limitations in the AnyIO docs.
Patterns
Run an async entrypoint run-an-entrypoint
import anyio
async def main() -> None:
await anyio.sleep(0.1)
print("ready")
anyio.run(main)Pass backend="trio" only after installing the trio extra; asyncio is the default backend.
Own child tasks with a task group start-concurrent-tasks
import anyio
async def fetch(name: str) -> None:
await anyio.sleep(0.1)
print(name)
async def main() -> None:
async with anyio.create_task_group() as tg:
first = tg.start_soon(fetch, "alpha")
second = tg.start_soon(fetch, "beta")
print(first.done(), second.done())
anyio.run(main)Version 4.14 makes start_soon() return a TaskHandle. Leaving the context waits for both children, and one failure cancels its siblings.
Create a coroutine task and keep its handle migrate-asyncio-task-creation
import anyio
async def calculate() -> int:
await anyio.sleep(0)
return 42
async def main() -> None:
async with anyio.create_task_group() as tg:
handle = tg.create_task(calculate())
print(handle.result())create_task() was added in 4.14 for asyncio-style migration. The task still belongs to the task group and cannot outlive it.
Wait until a child reports that it started wait-for-service-readiness
import anyio
from anyio.abc import TaskStatus
async def serve(*, task_status: TaskStatus[int] = anyio.TASK_STATUS_IGNORED) -> None:
listener = await anyio.create_tcp_listener(local_port=0)
port = listener.extra(anyio.abc.SocketAttribute.local_address)[1]
task_status.started(port)
await listener.serve(handle_client)
async def main() -> None:
async with anyio.create_task_group() as tg:
port = await tg.start(serve)
print(port)tg.start() waits for started(). If the child exits without calling it, the parent receives an error instead of assuming the service is ready.
Choose whether a timeout raises enforce-a-timeout
import anyio
async def main() -> None:
with anyio.move_on_after(2) as scope:
await optional_refresh()
if scope.cancelled_caught:
print("refresh skipped")
with anyio.fail_after(2):
await required_refresh()These are synchronous context managers. move_on_after() continues after expiry, while fail_after() raises TimeoutError.
Handle one error type from concurrent tasks handle-task-errors
import anyio
async def main() -> None:
try:
async with anyio.create_task_group() as tg:
tg.start_soon(parse_one)
tg.start_soon(parse_two)
except* ValueError as group:
for error in group.exceptions:
print(error)except* syntax requires Python 3.11 or newer. AnyIO supports Python 3.10, where exception-group compatibility helpers are needed instead.
Send objects through a bounded stream send-items-with-backpressure
import anyio
async def producer(send: anyio.abc.ObjectSendStream[int]) -> None:
async with send:
for value in range(5):
await send.send(value)
async def main() -> None:
send, receive = anyio.create_memory_object_stream[int](2)
async with anyio.create_task_group() as tg:
tg.start_soon(producer, send)
async with receive:
async for value in receive:
print(value)Close every send clone or the receiving iterator will not know that the stream has ended. The buffer size controls how far producers can get ahead.
Cap access to a scarce resource limit-concurrency
import anyio
limiter = anyio.CapacityLimiter(8)
async def query(item: str) -> None:
async with limiter:
await call_service(item)CapacityLimiter tracks borrowers, which is useful when one logical operation acquires and releases on behalf of another. Use Semaphore for simpler permit counting.
Move a blocking call to a worker thread run-blocking-code
from functools import partial
import anyio
async def main() -> None:
text = await anyio.to_thread.run_sync(
partial(read_report, encoding="utf-8"),
"report.txt",
)
print(text)Keyword arguments need functools.partial. Cancelling the await does not forcibly terminate Python code that is already running in the worker thread.
Shield a short cleanup operation protect-async-cleanup
import anyio
async def use_connection(connection) -> None:
try:
await do_work(connection)
finally:
with anyio.CancelScope(shield=True):
await connection.aclose()Keep the shield narrow. A shield delays outer cancellation, so unrelated work should stay outside it.
Run a TCP echo listener serve-tcp-clients
import anyio
async def echo(stream: anyio.abc.SocketStream) -> None:
async with stream:
while data := await stream.receive():
await stream.send(data)
async def main() -> None:
listener = await anyio.create_tcp_listener(local_host="127.0.0.1", local_port=9000)
await listener.serve(echo)
anyio.run(main)serve() keeps running until cancelled or the listener fails. Put it in an owning task group when the application needs an explicit shutdown path.
Run one pytest test on asyncio and Trio test-multiple-backends
import pytest
@pytest.fixture(params=["asyncio", "trio"])
def anyio_backend(request):
return request.param
@pytest.mark.anyio
async def test_lookup() -> None:
assert await lookup("id-7") is not NoneInstall the trio extra before selecting that backend. Configure pytest-asyncio carefully if it is present, because two auto-mode async plugins can conflict.
Alternatives
| Package | Registry | Pick it when |
|---|---|---|
| trio | PyPI | Use it when you control the application and want Trio's native structured-concurrency model without an asyncio backend. |
| uvloop | PyPI | Use it when your code should stay on asyncio and the goal is a replacement event loop rather than a new concurrency API. |
| pytest-asyncio | PyPI | Use it when the only missing piece is asyncio test and fixture support in pytest. |
More web backend guides
urllib3 · requests · ws · undici · httpx · express · the whole shelf →
How this guide is made: grounded in the library's documentation, release notes, changelog, and issue history, on a fixed rubric — not a hands-on install of every release. The 50 most-downloaded entries are additionally install-verified in clean containers. Corrections: contact the desk.