Python Structured Concurrency: Trio/AnyIO Patterns for Resilient Services

How structured concurrency in Trio and AnyIO helps you build Python services that fail fast, recover cleanly, and don’t leak background tasks.

Learn Python structured concurrency with Trio and AnyIO, and apply real-world patterns to build resilient, observable, and predictable async services in production.

You know that feeling when your async Python service is technically “running”, but you no longer trust what tasks are still alive inside it?

Some request got cancelled. Some background job “might” still be running. You’re staring at logs, hoping nothing important is silently stuck.

That’s the exact mess structured concurrency was invented to avoid.

In this article, we’ll look at Python structured concurrency using Trio and AnyIO, and how it changes the way you design resilient services — services that shut down gracefully, handle failure predictably, and don’t leave orphaned tasks drifting in the event loop.

Why “Just Use asyncio” Stops Being Enough

Python’s asyncio gave us the primitives:

• async def / await
• asyncio.create_task
• event loops, futures, tasks

Powerful, yes. But also dangerously easy to misuse.

The most common anti-pattern?

async def handle_request(...):
    asyncio.create_task(do_background_thing())
    return {"status": "ok"}

Congratulations, you’ve just spawned a task that:

• Is not tied to the lifetime of the request
• Might keep running after the client disconnects
• Might fail with an exception you never see
• Might still be doing I/O when your service is trying to shut down

Multiply that by a few years of “quick fixes” and you end up with a service that behaves like a haunted house: tasks appear and disappear, logs scream at random times, and nobody knows who owns what.

Structured concurrency says: no more free-floating tasks.

What Is Structured Concurrency, Really?

Think of structured concurrency as “scoped async”.

Every spawned task must live inside a well-defined block of code, with a clear parent, lifetime, and cancellation story.

Instead of tasks being spawned and forgotten, you:

• Start tasks inside a task group (or nursery in Trio).
• The parent scope waits for children to finish (or cancels them) before it exits.
• If one task fails, you decide how that failure affects its siblings (fail-fast, cancel, handle).

In other words: your async control flow now has shape.

Trio: Nurseries as the Core Abstraction

Trio is built around this idea. You don’t call “create_task” in the wild. You open a nursery.

A minimal Trio example

import trio

async def worker(name, delay):
    await trio.sleep(delay)
    print(f"{name} done after {delay}s")

async def main():
    async with trio.open_nursery() as nursery:
        nursery.start_soon(worker, "task-1", 1)
        nursery.start_soon(worker, "task-2", 2)
    print("All workers finished")

trio.run(main)

Key points:

• open_nursery() creates a scope.
• start_soon schedules child tasks inside that scope.
• The async with block doesn’t exit until all children are done (or cancelled).
• If one child raises an exception, Trio will propagate and manage it in a controlled way.

No stray tasks. No “I forgot to await that”. No secret background demons.

AnyIO: Structured Concurrency, but Portable

You might be thinking: “Trio is cool, but a lot of the ecosystem is asyncio-first.”

That’s where AnyIO comes in.

AnyIO gives you:

• A structured concurrency API (TaskGroup)
• The ability to run on either asyncio or Trio backends
• A consistent set of primitives: task groups, cancel scopes, timeouts, streams

So you get the design benefits of Trio, while still playing nicely with asyncio-based frameworks (FastAPI, Starlette, etc.).

AnyIO TaskGroup example

import anyio

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

async def main():
    async with anyio.create_task_group() as tg:
        await tg.spawn(worker, "task-1", 1)
        await tg.spawn(worker, "task-2", 2)
    print("All workers finished")

anyio.run(main)

Same idea as Trio’s nursery, different names:

• create_task_group() ↔ open_nursery()
• tg.spawn() ↔ nursery.start_soon()

Under the hood, AnyIO guarantees that:

• The task group scope only exits when all tasks complete or are cancelled.
• Exceptions in one task are propagated in a predictable way.

Architecture: From Free-Fire Zone to Supervision Tree

Let’s visualize the difference.

Classic “unstructured” async

[request handler]
     |
     +--> create_task(A)  (no parent)
     +--> create_task(B)  (no parent)
     +--> return response

Tasks A and B keep running somewhere in the loop. If A fails, maybe it logs something. Maybe not. Your handler is long gone.

Structured concurrency version

[service main]
   |
   v
+------------------------+
| Task Group (supervisor)|
|   |                    |
|   +--> worker A        |
|   +--> worker B        |
+------------------------+

All workers live under a supervisor task group:

• On shutdown, you cancel the supervisor, which cancels children.
• On failure, you decide: propagate, restart, or degrade gracefully.

Loosely, it resembles the “supervision tree” model in Erlang/Elixir, just adapted to Python async.

Pattern #1: Graceful Shutdown for a Service

Imagine a service that:

• Listens on a socket or HTTP server
• Consumes from a queue
• Periodically runs maintenance jobs

You want all of that to stop cleanly when the process receives a shutdown signal.

AnyIO-style supervision

import anyio
import signal

async def http_server():
    # pretend this runs an app framework
    await anyio.sleep_forever()

async def queue_consumer():
    while True:
        # read from queue, process message
        await anyio.sleep(1)

async def maintenance_job():
    while True:
        # do periodic cleanup
        await anyio.sleep(60)

async def main():
    async with anyio.create_task_group() as tg:
        # Cancel the group when SIGINT or SIGTERM is received
        with anyio.open_signal_receiver(signal.SIGINT, signal.SIGTERM) as signals:
            await tg.spawn(http_server)
            await tg.spawn(queue_consumer)
            await tg.spawn(maintenance_job)

            async for signum in signals:
                print(f"Received signal {signum}, shutting down...")
                tg.cancel_scope.cancel()
                break

anyio.run(main)

What’s nice here:

• All long-running tasks live in one TaskGroup.
• A signal triggers one cancellation point.
• Children exit in an orderly fashion, honoring their awaits.

No more “did we forget to stop the queue worker?” panic.

Pattern #2: Fail-Fast APIs With Timeouts

Let’s be real: in modern services, the network is your biggest source of pain.

You don’t want a slow dependency to:

• Hang a request forever
• Leak tasks that never finish
• Leave half-done work behind

Structured concurrency plus timeouts gives you a clear story.

Using AnyIO cancel scopes

import anyio

async def call_downstream_service():
    # slow HTTP call or DB query
    await anyio.sleep(5)
    return "ok"

async def handle_request():
    try:
        with anyio.move_on_after(2) as scope:
            result = await call_downstream_service()
        if scope.cancel_called:
            # Timeout happened
            return {"error": "Upstream timeout"}, 504
        return {"result": result}, 200
    except Exception as exc:
        # Log and handle failures in a predictable way
        return {"error": str(exc)}, 500

move_on_after(2) means:

• Give this block up to 2 seconds.
• If it doesn’t complete, cancel all operations inside.
• Execution moves on; nothing keeps running in the background.

Compare that to asyncio.wait_for, which is easy to sprinkle everywhere without a coherent cancellation strategy.

Pattern #3: Concurrency Limits Without Chaos

You might be tempted to spin up 1,000 tasks to process a big batch. Then your database or API starts crying.

Instead, use structured concurrency + a bounded pattern.

Trio semaphore + nursery example

import trio

async def process_item(item, limiter):
    async with limiter:
        # Only N of these run concurrently
        await trio.sleep(0.5)
        print(f"Processed {item}")

async def main():
    limiter = trio.Semaphore(10)  # Max 10 concurrent workers
    items = range(100)

    async with trio.open_nursery() as nursery:
        for item in items:
            nursery.start_soon(process_item, item, limiter)

trio.run(main)

All tasks:

• Are managed by the nursery.
• Obey the concurrency limit via the semaphore.
• Finish before the main scope completes.

No surprise background work. No silent overload.

Structured Concurrency and Web Frameworks

If you’re using frameworks like FastAPI, Starlette, or any ASGI stack, you’re often already sitting on top of AnyIO or at least an event loop that can host it.

A clean pattern is:

• Let the web framework handle per-request scopes.
• Use AnyIO TaskGroups inside request handlers for sub-tasks that must finish with the request.
• Use service-level TaskGroups at startup for long-running background workers, tied to the app lifecycle.

The goal is always the same: nothing lives longer than the scope that owns it.

Mental Shift: From “Spawn and Hope” to “Own and Supervise”

The biggest change with Python structured concurrency isn’t the API surface. It’s the mindset.

Instead of asking:

“How do I start this task?”

You start asking:

“Who owns this task, and when should it die?”

Once you answer that, Trio and AnyIO give you the tools:

• Nurseries / TaskGroups for hierarchical lifetimes
• Cancel scopes for timeouts and shutdown
• Deterministic error propagation so you don’t lose failures in log noise

And your services start feeling… calmer. More predictable. Less haunted.

Wrapping Up: Resilience by Design, Not Accident

Python structured concurrency with Trio and AnyIO isn’t just a fancy async style. It’s a way to:

• Make task lifetimes explicit
• Handle failures coherently
• Implement graceful shutdown and timeouts
• Build services you can actually reason about at 3 a.m.

If you’ve ever been burned by “fire-and-forget” asyncio.create_task chaos, this is your upgrade path.

If this resonated with you:

• Drop a comment with how you’re handling async in production today.
• Follow for more deep dives on Python performance, reliability, and architecture patterns.
• Share this with that teammate who keeps promising “I’ll just add a small background task, it’ll be fine.”

Because with structured concurrency, it actually can be.