AnyIO Reference: Replacing asyncio Idioms

Skill mandate: import asyncio is BANNED. Use import anyio exclusively. This reference targets AnyIO 4.x (2026 Python projects).

1. Task Groups (The Core Primitive)

AnyIO uses structured concurrency via task groups. A task group is an async context manager that guarantees all child tasks finish before the block exits.

`start_soon` — fire-and-forget

python

import anyio

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

async def main() -> None:
    async with anyio.create_task_group() as tg:
        for i in range(3):
            tg.start_soon(worker, i)
    print("all tasks finished")

anyio.run(main)

Signature: tg.start_soon(func, *args, name=None)

func must be a coroutine function (not a coroutine object).
name is optional, for introspection/debugging.
No return value; exceptions propagate as ExceptionGroup on exit.

`start` — wait for ready signal

Use when a task must initialize before the caller proceeds (e.g., starting a server and then connecting to it).

python

from anyio import TASK_STATUS_IGNORED, create_task_group, run
from anyio.abc import TaskStatus

async def start_server(port: int, *, task_status: TaskStatus[None] = TASK_STATUS_IGNORED) -> None:
    listener = await anyio.create_tcp_listener(local_host="127.0.0.1", local_port=port)
    task_status.started()  # unblocks tg.start()
    await listener.serve(handler)

async def main() -> None:
    async with create_task_group() as tg:
        await tg.start(start_server, 8080)  # blocks until task_status.started()
        # server is guaranteed ready here
        async with await anyio.connect_tcp("127.0.0.1", 8080) as client:
            ...

run(main)

Rule of thumb:

Use start_soon when you don't need to know when the task is ready.
Use start when the task must signal readiness before you continue.

`create_task` — retrieving return values (AnyIO 4.14+)

python

async def add(x: int, y: int) -> int:
    return x + y

async def main() -> None:
    async with anyio.create_task_group() as tg:
        handle = tg.create_task(add(2, 4))
        result = await handle  # == 6
        print(handle.return_value)  # also 6

anyio.run(main)

Signature: tg.create_task(coro, *, name=None, context=None) -> TaskHandle[T]

Returns a TaskHandle you can await for the result.
If the task raises, awaiting raises TaskFailed (or TaskCancelled).
This is the canonical replacement for asyncio.gather when you need results.

2. asyncio → anyio Cheat Sheet

asyncio	anyio	Notes
`asyncio.gather(a, b, c)`	`tg.create_task(a); tg.create_task(b); tg.create_task(c); results = [await h for h in handles]`	No direct gather; structured concurrency requires explicit task group scope. For fire-and-forget, use `tg.start_soon`.
`asyncio.create_task(coro)`	`tg.start_soon(func, *args)` or `tg.create_task(coro)`	`start_soon` takes a coroutine function + args. `create_task` takes a coroutine object and returns a handle.
`asyncio.sleep(n)`	`anyio.sleep(n)`	Identical semantics.
`asyncio.wait_for(coro, timeout)`	`with anyio.fail_after(timeout): await coro`	Raises `TimeoutError`. Use `move_on_after` for silent timeout.
`asyncio.Event()`	`anyio.Event()`	AnyIO events are not reusable; create a new one instead of `.clear()`.
`asyncio.Lock()`	`anyio.Lock()`	Use `async with lock:`. Pass `fast_acquire=True` if performance-critical.
`asyncio.Semaphore(n)`	`anyio.Semaphore(n)`	Same. Pass `fast_acquire=True` if performance-critical.
`asyncio.Condition()`	`anyio.Condition()`	Same semantics.
`asyncio.run(main())`	`anyio.run(main)`	Backend-agnostic entry point.
`asyncio.Queue(maxsize=N)`	`anyio.create_memory_object_stream[T](max_buffer_size=N)`	Returns `(send_stream, receive_stream)`. Supports `async for` on receive end.
`asyncio.to_thread(fn, *args)`	`anyio.to_thread.run_sync(fn, *args)`	Supports `abandon_on_cancel=True` and custom `limiter`.
`asyncio.run_coroutine_threadsafe(coro, loop)`	`anyio.from_thread.run(func, *args)`	Call async code from a worker thread.
`loop.call_soon_threadsafe(callback)`	`anyio.from_thread.run_sync(func, *args)`	Call sync code in event loop thread from worker thread, with return value.
`asyncio.shield(coro)`	`with anyio.CancelScope(shield=True): ...`	AnyIO shielding does not orphan tasks.
`asyncio.timeout(delay)`	`with anyio.fail_after(delay): ...`	AnyIO uses level cancellation, not edge cancellation.
`asyncio.CancelledError`	`anyio.get_cancelled_exc_class()`	Use this to catch cancellation portably across backends.

3. Cancellation & CancelScope

AnyIO uses level cancellation (inspired by Trio), not asyncio's edge cancellation.

Edge cancellation (asyncio): A CancelledError is injected once. If caught and not re-raised, the task keeps running.
Level cancellation (anyio): As long as a task is inside an effectively cancelled scope, every yield point raises a new cancellation exception.

Basic CancelScope

python

from anyio import CancelScope, create_task_group, get_cancelled_exc_class, sleep, run

async def worker() -> None:
    try:
        await sleep(10)
    except get_cancelled_exc_class():
        print("cancelled!")
        raise  # ALWAYS re-raise cancellation exceptions

async def main() -> None:
    async with create_task_group() as tg:
        tg.start_soon(worker)
        await sleep(0.1)
        tg.cancel_scope.cancel()  # cancels all children

run(main)

Shielding

Shield a block from external cancellation. Essential for cleanup.

python

from anyio import CancelScope, create_task_group, sleep, run

async def main() -> None:
    async with create_task_group() as tg:
        with CancelScope(shield=True):
            tg.start_soon(some_task)
            tg.cancel_scope.cancel()  # shielded block is protected
            await sleep(1)  # this still runs

run(main)

Combine with timeouts for graceful shutdown:

python

from anyio import CancelScope, move_on_after

async def do_something(resource) -> None:
    try:
        await run_async_stuff()
    except BaseException:
        # Allow up to 10s for cleanup, then move on
        with move_on_after(10, shield=True):
            await resource.aclose()
        raise

Structured Concurrency Guarantee

A task group contains its own CancelScope. If any child task raises an exception:

The task group's cancel scope is cancelled.
All other child tasks receive cancellation.
The task group waits for all children to finish.
The original exception (wrapped in ExceptionGroup if multiple) is re-raised.

4. Timeouts

Two context managers. Both create a CancelScope internally.

`fail_after` — raises on timeout

python

from anyio import fail_after, sleep, run

async def main() -> None:
    try:
        with fail_after(5) as scope:
            await sleep(10)
    except TimeoutError:
        print("timed out")
        print(scope.cancelled_caught)  # True

run(main)

`move_on_after` — silent timeout

python

from anyio import move_on_after, sleep, run

async def main() -> None:
    with move_on_after(5) as scope:
        await sleep(10)
        print("this never prints")

    print("exited scope, cancelled =", scope.cancelled_caught)

run(main)

Combined with shielding

python

from anyio import move_on_after

# Give cleanup 10 seconds, but don't let outer cancellation interrupt it
with move_on_after(10, shield=True):
    await resource.aclose()

5. Memory Object Streams (Queue Replacement)

Replaces asyncio.Queue with a safer, typed, structured-concurrency-friendly construct.

python

from anyio import create_task_group, create_memory_object_stream, run
from anyio.streams.memory import MemoryObjectReceiveStream

async def consumer(stream: MemoryObjectReceiveStream[str]) -> None:
    async with stream:  # closes receive end on exit
        async for item in stream:
            print("received", item)

async def main() -> None:
    # Type-annotated stream creation (AnyIO 4+ syntax)
    send_stream, receive_stream = create_memory_object_stream[str](max_buffer_size=10)

    async with create_task_group() as tg:
        tg.start_soon(consumer, receive_stream)
        async with send_stream:
            for i in range(5):
                await send_stream.send(f"item {i}")
        # send_stream closed → consumer's async for loop exits naturally

run(main)

Key differences from asyncio.Queue:

Bounded by default: max_buffer_size=0 means send blocks until a receiver is ready.
Cloneable: Each producer/consumer can close its own clone. The stream only ends when all clones of one end are closed.
Async iterable: async for item in receive_stream: works out of the box.
Type-safe: Generic create_memory_object_stream[T]().
Synchronous close: Both close() and async with work.

6. Backend Selection

AnyIO is backend-agnostic. Code written against AnyIO APIs runs on both asyncio and Trio.

python

import anyio

async def main() -> None:
    print("running on", anyio.current_async_library())
    await anyio.sleep(1)

# Default backend (asyncio)
anyio.run(main)

# Explicit backend
anyio.run(main, backend="trio")
anyio.run(main, backend="asyncio", backend_options={"debug": True})

Library design rule: Never hardcode a backend. Let the application choose via anyio.run(). Libraries should only import anyio and avoid backend-specific APIs.

7. Compatibility with asyncio-only libraries

Using asyncio libraries under the asyncio backend

If a third-party library exposes only an asyncio interface (returns asyncio coroutine objects), it works directly under the asyncio backend because AnyIO runs on top of asyncio's event loop:

python

import anyio
import some_asyncio_only_lib  # returns asyncio.Future/coroutine objects

async def main() -> None:
    # This works because under the asyncio backend, await passes through
    result = await some_asyncio_only_lib.fetch_data()

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

Important: This only works on the asyncio backend. On the trio backend, asyncio-native objects will not work.

When you MUST use asyncio APIs

Some APIs have no AnyIO equivalent and require direct event loop access:

Scenario	asyncio API	AnyIO approach
Signal handlers	`loop.add_signal_handler()`	`anyio.open_signal_receiver()`
Custom protocols	`asyncio.Protocol`	Use AnyIO streams / sockets
Direct Future manipulation	`asyncio.Future`	Avoid; use AnyIO primitives
Eager task factories	`asyncio.eager_task_factory`	Experimental in AnyIO; avoid

If you absolutely need the running loop:

python

import asyncio

async def main() -> None:
    loop = asyncio.get_running_loop()
    # ... do something loop-specific ...
    # WARNING: this breaks backend-agnosticism

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

Best practice: Wrap asyncio-only code in a backend-agnostic facade, and document that the feature requires the asyncio backend.

8. Idiomatic Code Snippets

Snippet 1: Parallel HTTP requests with timeout and cleanup

python

import anyio

async def fetch(url: str) -> bytes:
    await anyio.sleep(0.5)  # simulate
    return b"data"

async def main() -> None:
    urls = ["a", "b", "c"]
    async with anyio.create_task_group() as tg:
        with anyio.move_on_after(5):
            for url in urls:
                tg.start_soon(fetch, url)
        # All tasks are cancelled on timeout; task group waits for cleanup

anyio.run(main)

Snippet 2: Producer-consumer with memory object stream

python

import anyio
from anyio.streams.memory import MemoryObjectReceiveStream

async def producer(send_stream: anyio.streams.memory.MemoryObjectSendStream[int]) -> None:
    async with send_stream:
        for i in range(100):
            await send_stream.send(i)

async def consumer(receive_stream: MemoryObjectReceiveStream[int]) -> None:
    async with receive_stream:
        async for item in receive_stream:
            print(f"consumed {item}")

async def main() -> None:
    send, receive = anyio.create_memory_object_stream[int](max_buffer_size=5)
    async with anyio.create_task_group() as tg:
        tg.start_soon(producer, send)
        tg.start_soon(consumer, receive)

anyio.run(main)

Snippet 3: Calling sync code from async

python

import time
import anyio

async def main() -> None:
    # Run blocking function in worker thread
    result = await anyio.to_thread.run_sync(time.sleep, 2)
    print("done")

anyio.run(main)

Snippet 4: Calling async code from a worker thread

python

import anyio

def blocking_callback() -> None:
    # Inside a worker thread, call back into the event loop
    anyio.from_thread.run(anyio.sleep, 1)
    anyio.from_thread.run_sync(print, "hello from thread")

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

anyio.run(main)

Snippet 5: Graceful shutdown with shielded cleanup

python

import anyio

async def worker() -> None:
    try:
        await anyio.sleep_forever()
    except anyio.get_cancelled_exc_class():
        with anyio.CancelScope(shield=True):
            await anyio.sleep(0.5)  # cleanup
            print("cleaned up")
        raise

async def main() -> None:
    async with anyio.create_task_group() as tg:
        tg.start_soon(worker)
        await anyio.sleep(1)
        tg.cancel_scope.cancel()

anyio.run(main)

Sources

AnyIO Documentation (stable): https://anyio.readthedocs.io/en/stable/
AnyIO GitHub (HEAD cb245dba): https://github.com/agronholm/anyio
Task Groups: https://anyio.readthedocs.io/en/stable/tasks.html
Cancellation & Timeouts: https://anyio.readthedocs.io/en/stable/cancellation.html
Streams: https://anyio.readthedocs.io/en/stable/streams.html
Synchronization: https://anyio.readthedocs.io/en/stable/synchronization.html
Threads: https://anyio.readthedocs.io/en/stable/threads.html
Basics / Backends: https://anyio.readthedocs.io/en/stable/basics.html
Design Rationale (why asyncio is problematic): https://anyio.readthedocs.io/en/stable/why.html

AnyIO Reference: Replacing asyncio Idioms

AnyIO Reference: Replacing asyncio Idioms

1. Task Groups (The Core Primitive)

start_soon — fire-and-forget

start — wait for ready signal

create_task — retrieving return values (AnyIO 4.14+)

2. asyncio → anyio Cheat Sheet

3. Cancellation & CancelScope

Basic CancelScope

Shielding

Structured Concurrency Guarantee

4. Timeouts

fail_after — raises on timeout

move_on_after — silent timeout

Combined with shielding

5. Memory Object Streams (Queue Replacement)

6. Backend Selection

7. Compatibility with asyncio-only libraries

Using asyncio libraries under the asyncio backend

When you MUST use asyncio APIs

8. Idiomatic Code Snippets

Snippet 1: Parallel HTTP requests with timeout and cleanup

Snippet 2: Producer-consumer with memory object stream

Snippet 3: Calling sync code from async

Snippet 4: Calling async code from a worker thread

Snippet 5: Graceful shutdown with shielded cleanup

Sources

`start_soon` — fire-and-forget

`start` — wait for ready signal

`create_task` — retrieving return values (AnyIO 4.14+)

`fail_after` — raises on timeout

`move_on_after` — silent timeout