Python asyncio TaskGroup: Structured Concurrency

When you need to run several asynchronous operations together and manage their lifecycle as a unit, asyncio.TaskGroup provides a structured approach. Introduced in Python 3.11, it gives you a clear way to spawn tasks, wait for all of them, and automatically cancel the group if any task fails. This article explains how to use python asyncio taskgroup effectively, with practical examples and the behavior you need to understand before adopting it in production code.

What asyncio.TaskGroup Solves

Before TaskGroup, the common pattern was to create tasks with asyncio.create_task() and then await them individually or use asyncio.gather(). That approach works, but it leaves the responsibility of tracking tasks and handling failures to the caller. If one task raises an exception, gather() propagates it, but other tasks continue running unless you explicitly cancel them. This can lead to orphaned tasks that keep consuming resources after an error.

TaskGroup enforces structured concurrency: tasks created inside the group are guaranteed to be awaited or cancelled before the group exits. If any task raises an exception, the group cancels all remaining tasks and then re-raises the first exception. This makes error propagation deterministic and prevents tasks from leaking.

Basic Usage: Creating a TaskGroup

To use TaskGroup, you create an instance and add tasks using its create_task() method. The group acts as an asynchronous context manager. Here is a minimal example:

import asyncio

async def worker(name: str, delay: float) -> str:
    await asyncio.sleep(delay)
    return f"{name} done"

async def main():
    async with asyncio.TaskGroup() as tg:
        task1 = tg.create_task(worker("first", 1.0))
        task2 = tg.create_task(worker("second", 2.0))
    print(task1.result())
    print(task2.result())

asyncio.run(main())

When the async with block exits normally, all tasks have completed. You can access their results via task.result(). If you need to retrieve results as they become available, you can await each task inside the block, but the group already waits for all tasks at the end.

Error Handling and Propagation

One of the main advantages of TaskGroup is how it handles exceptions. If any task inside the group raises an exception, the group immediately cancels all other tasks and waits for them to finish cancelling. After the block, the first exception is re-raised. This means you do not need to manually cancel sibling tasks when one fails.

import asyncio

async def fail():
    raise ValueError("boom")

async def slow():
    await asyncio.sleep(10)
    return "slow done"

async def main():
    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(fail())
            tg.create_task(slow())
    except* ValueError as eg:
        print(f"Caught: {eg.exceptions}")

asyncio.run(main())

Notice the except* syntax. TaskGroup raises an ExceptionGroup that can contain multiple exceptions. The except* clause lets you handle specific exception types from the group. If you use a plain except ValueError, it will not catch the exception because it is wrapped in an ExceptionGroup. This is a common source of confusion for developers new to TaskGroup.

Cancellation Semantics

When a task fails, TaskGroup cancels all other tasks. Cancellation in asyncio is cooperative: the cancelled task receives a CancelledError at the next await point. If a task is in a blocking operation that does not respond to cancellation, it may not stop immediately. However, the group waits for all tasks to finish (either by completing or being cancelled) before exiting the context manager.

You can also cancel the entire group from outside. If the async with block is cancelled (for example, by a timeout), the group cancels all tasks and waits for them to finish. This makes TaskGroup useful for implementing timeouts and graceful shutdowns.

TaskGroup vs asyncio.gather

asyncio.gather() is the older way to run multiple coroutines concurrently. It has a return_exceptions parameter that controls whether exceptions are collected or raised immediately. The table below summarizes the key differences:

Use TaskGroup when you need structured cancellation and you are on Python 3.11 or later. If you must support older Python versions, gather() remains the compatible option.

Practical Example: Running Concurrent HTTP Requests

A common use case is making several HTTP requests in parallel. Here is an example using aiohttp (you would need to install it separately) to fetch multiple URLs with a TaskGroup:

import asyncio
import aiohttp

async def fetch(session: aiohttp.ClientSession, url: str) -> str:
    async with session.get(url) as response:
        return await response.text()

async def main():
    urls = [
        "https://example.com",
        "https://httpbin.org/get",
        "https://jsonplaceholder.typicode.com/todos/1",
    ]
    async with aiohttp.ClientSession() as session:
        async with asyncio.TaskGroup() as tg:
            tasks = [tg.create_task(fetch(session, url)) for url in urls]
        results = [task.result() for task in tasks]
    for url, result in zip(urls, results):
        print(f"{url}: {len(result)} bytes")

asyncio.run(main())

If one request fails, the group cancels the others and raises an ExceptionGroup. You can catch it and decide whether to retry or degrade gracefully.

Compatibility and Migration Considerations

asyncio.TaskGroup requires Python 3.11 or later. If you are on an older version, you cannot use it without a backport. There is no official backport, so you would need to implement similar behavior manually or stick with gather(). When migrating existing code, be aware of the exception handling change: code that catches Exception directly will break because ExceptionGroup is a different type. You need to use except* or inspect the .exceptions attribute.

Another subtlety is that TaskGroup does not support passing coroutines directly; you must call create_task() with a coroutine. This is similar to create_task() but adds the group management. Also, tasks created inside the group are not accessible after the block unless you keep references, so plan accordingly.

Edge Cases and Common Pitfalls

One pitfall is using asyncio.shield() inside a task group. A shielded task is not cancelled when the group cancels, but the group still waits for it. This can cause the group to hang if the shielded task never finishes. Use shielding only when you are certain the task will complete independently.

Another issue is mixing TaskGroup with low-level loop.create_task(). The latter does not participate in the group's cancellation logic. If you create a task that way inside the group, it will not be cancelled on failure, defeating the purpose. Always use tg.create_task() for tasks that should be managed by the group.

Finally, remember that TaskGroup is not a replacement for asyncio.Semaphore or other synchronization primitives. It only manages task lifecycle, not access to shared resources. If you need to limit concurrency, combine TaskGroup with a semaphore inside the task coroutine.

Understanding these behaviors helps you use TaskGroup safely and avoid subtle bugs in concurrent code.