python-trio-anyio-migrator

Move asyncio code to structured concurrency without changing behaviour by accident. Start by deciding whether to move at all: asyncio with TaskGroup on 3.11+ may be enough. If the answer is yes, migrate in an order that keeps the code running at every step. The rewriter handles the mechanical renames exactly. The skill covers the parts that need judgement: gather, free-floating tasks, queues, futures, callbacks and cancellation. Every migrated module should end up passing the same tests on the asyncio and trio backends, plus a CI gate that stops asyncio from creeping back in.

#Trigger Conditions

Activate this skill when the user:

• Asks to port, convert or migrate asyncio code to AnyIO or trio, or asks "should we use anyio/trio?"
• Writes a library or SDK wrapper that must work under asyncio and trio (backend-agnostic)
• Needs replacements for gather, create_task, wait_for, asyncio.Queue, Future, call_later, add_done_callback, run_in_executor or run_coroutine_threadsafe
• Hits ExceptionGroup where a plain exception used to come out, a task group that will not finish, an await in finally that raises immediately, or get_cancelled_exc_class() swallowed in an except
• Mixes FastAPI/Starlette/httpx (anyio-based) with asyncpg/aiohttp (asyncio-only)
• Moves tests from pytest-asyncio to @pytest.mark.anyio, or wants deterministic time with trio's MockClock

#Workflow

1. DECIDE    -> Migrate, or stay on asyncio + TaskGroup? (decision table below)
2. INVENTORY -> python scripts/migrate_async.py src/ --json      (mechanical vs manual per file)
3. TESTS     -> switch the suite to the anyio pytest plugin first; keep it green on asyncio
4. MECHANICAL-> python scripts/migrate_async.py src/ --diff, review, then --write per file
5. MANUAL    -> gather -> task group + results list, Queue -> memory streams, create_task -> owned
                groups, wait_for -> fail_after, futures/callbacks -> events/streams
6. CANCEL    -> re-raise get_cancelled_exc_class(), shield awaits in finally, review timeouts
7. VERIFY    -> tests on asyncio AND trio backends; migrate_async.py src/ --check as a CI gate

#Why Structured Concurrency

Under the asyncio backend, anyio builds this model on top of the asyncio loop. Native asyncio libraries still work there, and anyio's cancel scopes also cancel the native asyncio awaits inside them (verified on anyio 4.14/4.15). Details: structured-concurrency-concepts.md.

#Should You Migrate? Situation -> Right Tool

App authors may pin a backend. Library authors should depend on anyio and test on both backends. If the user cannot name a concrete benefit (correct cancellation, no lost tasks, or backend independence), recommend staying on asyncio.

#Ecosystem Compatibility

#Mapping Cheat Sheet (full table: api-mapping-asyncio-anyio-trio.md)

#Migrating the Hard Parts

gather -> task group + results list. gather without return_exceptions does not cancel the other awaitables when one fails. A task group does, and it raises an ExceptionGroup instead of the first error.

async def fan_out(prompts: list[str], limiter: anyio.CapacityLimiter) -> list[str]:
    results: list = [None] * len(prompts)

    async def run_one(i: int, prompt: str) -> None:
        async with limiter:                       # replaces gather + Semaphore
            results[i] = await call_llm(prompt)

    async with anyio.create_task_group() as tg:
        for i, prompt in enumerate(prompts):
            tg.start_soon(run_one, i, prompt)
    return results

For return_exceptions=True, catch Exception inside run_one and store it. Never catch the cancellation class there. anyio 4.15 adds anyio.gather, but it is new and trio has nothing like it, so the pattern above is the portable choice.

create_task -> an owner. Every fire-and-forget task needs a task group that outlives it, for example one opened in the FastAPI lifespan. Before anyio 4.14, and always in trio, start_soon returns None: results go into a dict, list or memory stream, and cancelling one task means cancelling its own CancelScope.

Queue -> memory object stream. The stream has a send end and a receive end. max_buffer_size gives backpressure; 0 means rendezvous. Each producer gets its own clone(). Closing every send clone ends the consumers' async for, which makes join()/task_done() and sentinel values unnecessary.

send, receive = anyio.create_memory_object_stream[str](max_buffer_size=32)
async with anyio.create_task_group() as tg:
    for _ in range(4):
        tg.start_soon(embed_worker, receive.clone())   # worker: `async with recv: async for doc in recv`
    receive.close()                                    # workers hold the clones
    async with send:
        for doc in docs:
            await send.send(doc)                       # blocks when 32 are waiting: backpressure

wait_for -> scopes. with anyio.fail_after(10): reply = await llm(p). Use move_on_after and check scope.cancelled_caught when giving up is not an error. Future/callbacks become an Event plus a value slot, or a 1-item stream. call_later becomes a child task that sleeps and then acts. asyncio.wait FIRST_COMPLETED becomes children that store a result and call tg.cancel_scope.cancel(). Full recipes: migration-playbook.md.

#Cancellation and Exception Groups (the Gotchas)

Deep dive: cancellation-and-exception-groups.md.

#Mixed Codebases and Interop

• Under anyio.run(main) (asyncio backend), asyncpg/aiohttp calls and even asyncio.wait_for keep working. Migrate module by module. Run the tests on both backends only for modules that are free of asyncio-only dependencies.
• Blocking SDKs: await anyio.to_thread.run_sync(fn, *args, limiter=limiter). From inside that worker thread, call back into the loop with anyio.from_thread.run(async_fn, *args).
• Sync code that needs async (a CLI, Celery task or Streamlit callback): use with anyio.from_thread.start_blocking_portal() as portal: portal.call(async_fn, arg).
• An asyncio-only library inside trio: wrap it with trio-asyncio, or keep the service on anyio's asyncio backend.

Thread sizing and executor details: threadpool builder. CPU-bound work: GIL analyzer and multiprocessing designer (anyio.to_process.run_sync exists, but a tuned process pool is usually better).

#AI-Engineering Cases

Runnable versions: ai-engineering-anyio.md.

#Testing

import importlib.util

import anyio
import pytest

HAS_TRIO = importlib.util.find_spec("trio") is not None


@pytest.fixture(params=["asyncio", pytest.param("trio", marks=pytest.mark.skipif(not HAS_TRIO,
                                                                                reason="trio not installed"))])
def anyio_backend(request):
    return request.param


@pytest.mark.anyio
async def test_fan_out_keeps_order_and_limit():
    assert await fan_out(["a", "b", "c"], anyio.CapacityLimiter(2)) == ["A", "B", "C"]

• Replace @pytest.mark.asyncio with @pytest.mark.anyio. Async fixtures work when the test requests anyio_backend; anyio_mode = "auto" in recent anyio 4.x marks every async test.
• In recent anyio 4.x the built-in anyio_backend fixture runs every installed backend. Older releases ran asyncio only. Override it explicitly (as above) so the result does not depend on the version.
• Trio gives deterministic time: trio.testing.MockClock(autojump_threshold=0) via anyio_backend = ("trio", {"clock": MockClock(autojump_threshold=0)}). A 30-second timeout test then takes milliseconds.

More: testing-with-anyio-and-trio.md.

#Symptom -> Cause -> Fix

More symptoms and step-by-step diagnosis: troubleshooting-migration.md. For hangs, see the deadlock debugger.

#Verify

1. python scripts/migrate_async.py src/ --check exits 0, or what remains is behind # noqa: MIG0xx with a reason such as "asyncpg adapter".
2. The test suite passes with anyio_backend set to asyncio, and also to trio wherever the dependencies allow.
3. Cancellation tests: cancel mid-flight and assert that cleanup ran, that nothing leaked (open streams, threads) and that no warnings appeared (-W error::ResourceWarning).
4. Load test the fan-out: concurrency stays at or below the limiter's total_tokens, and p95 latency has not regressed.

#Output Format

When advising or migrating, answer with:

1. Decision: migrate or not, to anyio or trio, and why (dependencies, library vs app, Python version).
2. Inventory: mechanical vs manual counts per file (script output), plus blockers (asyncio-only deps).
3. Rewritten code: complete before/after for every manual item, with semantics differences called out (errors grouped, siblings cancelled, cancellation waits for threads, and so on).
4. Cancellation review: every except that catches the cancel class re-raises, every await in finally is shielded with a timeout, and every timeout is placed where it belongs.
5. Tests and gate: anyio plugin setup, both-backend fixture, the --check CI step.

#Scripts

# Inventory: every asyncio usage with rule id, mechanical/manual, and the replacement
python scripts/migrate_async.py src/rag_service/
python scripts/migrate_async.py app/agent.py --target trio --json
cat worker.py | python scripts/migrate_async.py - --diff               # stdin: report or --diff only
python scripts/migrate_async.py app/agent.py --diff                    # unified diff of the exact rewrites
python scripts/migrate_async.py app/agent.py --write                   # in place, keeps app/agent.py.bak
python scripts/migrate_async.py app/agent.py --write --force           # write although manual items remain
python scripts/migrate_async.py src/ --check                           # CI gate: exit 1 while asyncio remains
python scripts/migrate_async.py src/ --strict                          # exit 1 if anything needs a human
python scripts/migrate_async.py src/ --select MIG003,MIG004 --ignore gather
python scripts/migrate_async.py --list-rules --target trio             # why + fix snippet for every rule

Mechanical rewrites happen only when they are provably equivalent at that call site: MIG001 imports (import asyncio is kept only while still used), MIG002 asyncio.run(main(a)) -> anyio.run(main, a) (kwargs via functools.partial), MIG003 sleep, MIG004 Event/Lock/Semaphore/ BoundedSemaphore/Condition, MIG005 to_thread -> to_thread.run_sync, MIG006 async with asyncio.timeout(x) -> with anyio.fail_after(x), MIG007/MIG008 TaskGroup + tg.create_task(f(a)) -> create_task_group + tg.start_soon(f, a) (only when no task result is used), MIG009 except CancelledError -> get_cancelled_exc_class(), and MIG010 asyncio.TimeoutError. If a call site is not safe (Event.clear() used, as cm on a timeout, a stored task, a builtin TimeoutError handler with --target trio), the item is reported as manual with the reason. Manual-only rules: MIG011-MIG033 (gather, wait_for, create_task, Queue, join/task_done, run_in_executor, loop access, callbacks, futures, done callbacks, wait, as_completed, shield, subprocess, sockets, signals, run_coroutine_threadsafe, introspection, task.cancel, swallowed cancellation, advanced timeouts, coroutine checks) and MIG099 for anything else. # noqa: MIG013 skips one line, including its rewrite. Every rewrite is checked with ast.parse. --write refuses a file that has manual items unless --force is given. Exit codes: 0 ok; 1 --check found asyncio, --strict found manual items, or --write refused a file; 2 usage error, unreadable or unparsable file named on the command line (files found while walking a directory are skipped with a warning), or a rewrite that fails ast.parse.

#Additional Resources