Structured Concurrency with TaskGroup

Why TaskGroup?

asyncio.gather is useful, but it can make lifecycle and error handling messy.

asyncio.TaskGroup (Python 3.11+) provides structured concurrency:

• tasks are scoped to a block
• if one task fails, the group cancels the rest
• errors are grouped in an ExceptionGroup

Example

import asyncio
 
 
async def job(name: str, delay: float):
    await asyncio.sleep(delay)
    return f"{name} done"
 
 
async def main():
    async with asyncio.TaskGroup() as tg:
        t1 = tg.create_task(job("A", 0.2))
        t2 = tg.create_task(job("B", 0.1))
 
    # task results are available after the block
    print(t1.result())
    print(t2.result())
 
 
asyncio.run(main())

Handling failures

import asyncio
 
 
async def ok():
    await asyncio.sleep(0.1)
    return 1
 
 
async def bad():
    await asyncio.sleep(0.05)
    raise ValueError("boom")
 
 
async def main():
    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(ok())
            tg.create_task(bad())
    except* ValueError as eg:
        # ExceptionGroup support (PEP 654)
        print("handled:", eg)
 
 
asyncio.run(main())

Notes

• Requires Python 3.11+
• Consider TaskGroup for “start N tasks, all must finish” patterns

The idea in one sentence

Structured concurrency means tasks live inside a block of code, and the block does not finish until every task it started has finished. Nothing is left running in the background by accident.

Reading the examples

In taskgroup_basic.py, two tasks are created with tg.create_task(...) inside async with asyncio.TaskGroup(). The async with block ends only after both tasks are complete, which is why t1.result() and t2.result() are safe to call right after it. Task B has the shorter delay and finishes first, but the block still waits for A.

In taskgroup_error.py, bad() raises after 0.05 seconds while ok() would need 0.1. The group notices the failure, cancels ok(), waits for it to stop, and then raises an ExceptionGroup containing the ValueError. The except* syntax catches errors of a given type out of that group. It exists because several tasks can fail at the same time, and all of their errors are reported together.

Comparing with gather

With asyncio.gather, one failing task raises its error to you, but the other tasks keep running unless you cancel them yourself. That is how leaked tasks happen. A TaskGroup cannot leak: leaving the block means everything has stopped.

A practical example

import asyncio
 
 
async def square(n: int) -> int:
    await asyncio.sleep(0.05)
    return n * n
 
 
async def main() -> None:
    async with asyncio.TaskGroup() as tg:
        tasks = [tg.create_task(square(n)) for n in range(5)]
    print([t.result() for t in tasks])  # [0, 1, 4, 9, 16]
 
 
asyncio.run(main())

Common mistakes

• Reading result() inside the block. The tasks may not be finished yet. Read results after the async with ends.
• Using except ValueError instead of except* ValueError. The error arrives wrapped in an ExceptionGroup, so a plain except will not match.
• Running on Python older than 3.11. TaskGroup and except* do not exist there; use gather instead.
• Catching and hiding errors inside every task. The group can only cancel siblings if the error escapes the failing task.

When to use it

Use a TaskGroup for "start these N jobs and they all must succeed" work: fetching several pieces of data for one page, or running the steps of one request. Use gather(..., return_exceptions=True) when you want every outcome, including failures, without cancelling anything.

🧪 Try It Yourself

Exercise 1 – Create a Task

Press Run to see what your code prints.

Exercise 2 – Gather Multiple Tasks

Press Run to see what your code prints.

Exercise 3 – Task Name and Done

Press Run to see what your code prints.