AnyIO

Common errors

Warnings

Install

Install compatibility verified last tested: 2026-05-12

Imports

• run
  wrong

  asyncio.run(main)

  correct

  import anyio
  anyio.run(main)

  Use anyio.run() to stay backend-agnostic; pass backend='trio' or backend='asyncio' explicitly when needed

• create_task_group
  wrong

  await tg.spawn(fn)

  correct

  from anyio import create_task_group
  async with create_task_group() as tg:
      tg.start_soon(fn)

  TaskGroup.spawn() was removed in AnyIO 4; use the synchronous tg.start_soon() instead

• create_memory_object_stream
  wrong

  create_memory_object_stream(10, item_type=int)

  correct

  from anyio import create_memory_object_stream
  send, recv = create_memory_object_stream[int](max_buffer_size=10)

  The item_type keyword argument is deprecated; use the generic bracket syntax create_memory_object_stream[T]()

• start_blocking_portal
  wrong

  from anyio import start_blocking_portal

  correct

  from anyio.from_thread import start_blocking_portal
  with start_blocking_portal() as portal:
      portal.call(async_fn)

  start_blocking_portal lives in anyio.from_thread, not the top-level anyio namespace; must be used as a context manager

• to_thread.run_sync
  wrong

  await to_thread.run_sync(blocking_fn, cancellable=True)

  correct

  from anyio import to_thread
  await to_thread.run_sync(blocking_fn, abandon_on_cancel=True)

  The 'cancellable' parameter is a deprecated alias for 'abandon_on_cancel'; use abandon_on_cancel instead

• CancelScope

  from anyio import CancelScope
  with CancelScope(deadline=...) as scope:
      ...

  CancelScope is a plain synchronous context manager since AnyIO 3; do NOT use 'async with'

• Path

  from anyio import Path
  path = Path('/tmp/file')
  content = await path.read_text()

  anyio.Path is NOT a drop-in replacement for pathlib.Path; it cannot be passed to APIs expecting pathlib.Path

• get_cancelled_exc_class
  wrong

  except asyncio.CancelledError:

  correct

  from anyio import get_cancelled_exc_class
  try:
      await something()
  except get_cancelled_exc_class():
      raise

  The cancellation exception differs between backends; use get_cancelled_exc_class() to catch it portably

Quickstart verified last tested: 2026-04-23

Demonstrates concurrent task groups, move_on_after timeout scopes, and offloading blocking calls to a worker thread — all backend-agnostic.

import anyio

async def fetch(name: str, delay: float) -> None:
    print(f'{name}: starting')
    await anyio.sleep(delay)
    print(f'{name}: done after {delay}s')

async def main() -> None:
    # Task group: all tasks run concurrently; exceptions surface as ExceptionGroup
    async with anyio.create_task_group() as tg:
        tg.start_soon(fetch, 'task-A', 1.0)
        tg.start_soon(fetch, 'task-B', 0.5)

    # Timeout via CancelScope
    with anyio.move_on_after(2.0) as scope:
        await anyio.sleep(5)
    if scope.cancelled_caught:
        print('timed out (moved on)')

    # Run a blocking call in a thread
    import time
    result = await anyio.to_thread.run_sync(time.strftime, '%X')
    print('wall-clock time from thread:', result)

if __name__ == '__main__':
    # Default backend is asyncio; swap to 'trio' with no code changes
    anyio.run(main, backend='asyncio')