AnyIO for Python: task groups, cancellation, and timeouts

AnyIO provides a common set of asynchronous APIs for Python's asyncio and Trio backends. Its task groups keep child tasks within a defined lifetime, and its cancellation scopes make timeout and cleanup behavior explicit. Code can switch backends only when its dependencies support the selected backend too.

This guide targets AnyIO 4 and Python 3.11 or later, because the exception-handling example uses except*.

Why use AnyIO?

An asyncio-specific library does not become Trio-compatible merely because its caller uses AnyIO. What AnyIO gives library authors is a portable interface for supported operations, including concurrency and cancellation. Applications built around Starlette or FastAPI also encounter AnyIO through their framework dependencies.

If an application deliberately uses only asyncio, the standard library may be sufficient. Evaluate AnyIO when portable dependencies or its concurrency model solve a real design problem.

Install and run a coroutine

python3 -m venv .venv
source .venv/bin/activate
python -m pip install 'anyio>=4,<5'

Save and run this standalone script:

import anyio


async def main():
    print("Starting")
    await anyio.sleep(1)
    print("One second later")


anyio.run(main)

The default backend is asyncio. Install Trio support with python -m pip install 'anyio[trio]>=4,<5', then replace the final call with anyio.run(main, backend="trio"). Do not call anyio.run() from a thread that already has a running event loop; in an existing async context, await the coroutine through that context instead. See AnyIO basics.

Keep related tasks in a task group

import anyio


async def worker(name, delay):
    await anyio.sleep(delay)
    print(f"{name} finished")


async def main():
    async with anyio.create_task_group() as group:
        group.start_soon(worker, "Task A", 2)
        group.start_soon(worker, "Task B", 1)
    print("Both tasks finished")


anyio.run(main)

start_soon() schedules each worker with its arguments. The task-group context waits for its children before normal exit. If a child or the body raises an unhandled exception, the group cancels the other children and waits for their termination before propagating the failure.

This is structured concurrency: the parent's lifetime accounts for the tasks it starts. It still depends on cooperative code; a worker that blocks the event loop or suppresses cancellation can delay shutdown. See task groups.

Handle grouped failures and cancellation

The following complete example gives one task a deliberate failure while the other waits:

import anyio


async def risky_operation():
    await anyio.sleep(0.01)
    raise ValueError("Invalid result")


async def other_operation():
    try:
        await anyio.sleep_forever()
    finally:
        print("Other task cleaned up")


async def main():
    try:
        async with anyio.create_task_group() as group:
            group.start_soon(risky_operation)
            group.start_soon(other_operation)
    except* ValueError as errors:
        print("ValueError group:", errors.exceptions)


anyio.run(main)

AnyIO 4 task groups propagate ordinary non-cancellation failures in exception groups, including a single failure. Multiple tasks can fail during execution or cleanup. except* ValueError handles the matching subgroup; failures of other types still propagate. Groups containing exceptions outside Exception can use BaseExceptionGroup instead.

Cancellation takes effect at cancellation checkpoints; an arbitrary await is not a guarantee that the awaited code yields control. AnyIO uses level cancellation inside its scopes: a task can receive cancellation again at later yield points while its scope remains cancelled. If you catch the backend-specific cancellation exception, obtained through anyio.get_cancelled_exc_class(), re-raise it after cleanup.

For cleanup that must itself await, use a bounded shielded scope, such as with anyio.move_on_after(5, shield=True):, around that cleanup. Shielding avoids immediate repeated cancellation, while the separate deadline limits how long cleanup can wait. It does not forcibly stop blocking synchronous work. See cancellation and finalization.

Choose a timeout outcome

Both AnyIO timeout helpers use synchronous with, even inside an async function:

import anyio


async def slow_api():
    await anyio.sleep(10)
    return "Fresh result"


async def fetch_with_timeout():
    with anyio.move_on_after(0.05):
        return await slow_api()
    return "Fallback result"


async def strict():
    with anyio.fail_after(0.05):
        return await slow_api()


async def main():
    print(await fetch_with_timeout())
    try:
        await strict()
    except TimeoutError:
        print("The operation exceeded its deadline")


anyio.run(main)

move_on_after() suppresses the cancellation caused by its own scope and continues after the block. It does not suppress an unrelated ValueError or an outer scope's cancellation. fail_after() converts its timeout into TimeoutError outside the scope. The short values above keep the demonstration quick; choose real deadlines from the caller's budget and dependency behavior.

An outer scope can expire before an inner one. Use current_effective_deadline() when the effective deadline matters. A timeout is cooperative cancellation, not a guarantee that remote work has stopped or that elapsed time cannot exceed the configured value.

Common questions

Should an asyncio application migrate?

Not automatically. For a reusable library, backend portability can be valuable if its dependencies cooperate. For an asyncio-only application, compare the standard task-group and timeout APIs with AnyIO's semantics before changing working code.

How much overhead does AnyIO add?

Measure representative operations with the same dependencies, concurrency, and cancellation behavior. This guide makes no performance ranking or claim of negligible overhead.

Can I call a synchronous HTTP client from a coroutine?

A direct blocking call blocks the event loop. Use an async client compatible with the backend, or offload the call through anyio.to_thread.run_sync(). Set a timeout in the blocking library too. AnyIO normally shields the wait for the worker thread from cancellation; abandon_on_cancel=True allows the awaiting task to stop waiting, but the thread still runs and its result is ignored. See working with threads.

Make timeout failures diagnosable

For deployed work, log the operation, configured deadline, outcome, and request or trace identifier without copying sensitive inputs. Guance's log collector guide explains collection; verify cancellation and cleanup records before alerting on timeout counts. Expected cancellation during shutdown should not automatically count as a service failure.

The examples were exercised with both asyncio and Trio backends. No load benchmark or remote service/Guance ingestion test was performed. For scheduled work, see APScheduler in Python (Chinese).