Concurrent code becomes difficult to reason about when tasks can outlive the operation that created them. A request handler may return while background tasks are still running, or one task may fail while its siblings continue doing work that is no longer useful.

Python’s asyncio.TaskGroup, available since Python 3.11, provides structured concurrency for related asynchronous tasks. Tasks created inside the group belong to a clear lifetime: leaving the async with block waits for them, and failures are handled as a group rather than as detached background events.

The lifetime of child tasks is explicit

A task group is an asynchronous context manager. Create related tasks inside it and use their results after the context exits:

import asyncio

async def fetch_label(name: str, delay: float) -> str:
    await asyncio.sleep(delay)
    return name.upper()

async def main() -> None:
    async with asyncio.TaskGroup() as tg:
        first = tg.create_task(fetch_label("alpha", 0.05))
        second = tg.create_task(fetch_label("beta", 0.02))

    print(first.result(), second.result())

asyncio.run(main())

The context does not finish until both child tasks finish. After a normal exit, the stored task objects are complete, so calling result() is straightforward.

This differs from creating tasks and then forgetting to retain or await them. A TaskGroup makes ownership visible in the control flow.

A child failure cancels sibling work

The most important behavior appears when one child fails. If a task raises an exception other than asyncio.CancelledError, the group cancels its remaining unfinished tasks and waits for them before leaving the context.

import asyncio

async def worker(name: str, delay: float, fail: bool = False) -> str:
    try:
        await asyncio.sleep(delay)
        if fail:
            raise ValueError(f"{name} failed")
        return name
    finally:
        print(f"cleanup: {name}")

async def main() -> None:
    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(worker("fast", 0.05, fail=True))
            tg.create_task(worker("slow", 10))
    except* ValueError as group:
        for error in group.exceptions:
            print(error)

asyncio.run(main())

When fast raises ValueError, the group cancels slow. The finally block still runs, which gives the cancelled task an opportunity to release resources.

Cancellation is cooperative. A coroutine receives asyncio.CancelledError at an appropriate suspension point, such as an await. Code that performs long CPU-bound work without yielding cannot react promptly to task cancellation.

Do not swallow cancellation accidentally

Cancellation is part of asyncio control flow, not just another application error. Cleanup code should normally use try/finally:

async def consume(queue) -> None:
    resource = await open_resource()
    try:
        await process_messages(queue, resource)
    finally:
        await resource.close()

If code explicitly catches asyncio.CancelledError, it should generally re-raise it after cleanup. TaskGroup and other structured-concurrency features rely on cancellation internally, so suppressing cancellation can interfere with their behavior.

Also note that asyncio.CancelledError is a direct subclass of BaseException, not Exception. A broad except Exception: therefore does not catch it.

Multiple failures are reported together

Concurrent tasks can fail close enough together that more than one non-cancellation exception must be reported. After all child tasks have finished, TaskGroup raises the relevant failures in an ExceptionGroup or BaseExceptionGroup.

Python’s except* syntax can select matching exceptions from such a group:

try:
    async with asyncio.TaskGroup() as tg:
        tg.create_task(load_primary())
        tg.create_task(load_secondary())
except* TimeoutError as timeouts:
    for error in timeouts.exceptions:
        log_timeout(error)

Do not assume an exception group always contains only one failure. Error handling should make sense when several tasks fail independently.

KeyboardInterrupt and SystemExit receive special treatment: the task group still cancels and waits for remaining children, then re-raises the original base exception instead of wrapping it in the usual exception-group result.

TaskGroup and gather have different failure semantics

asyncio.gather() remains useful when its semantics match the problem, but it is not interchangeable with TaskGroup.

With the default return_exceptions=False, gather() propagates the first exception to the awaiting caller, but other submitted awaitables are not automatically cancelled merely because one failed. A task group instead treats sibling tasks as one operation: a non-cancellation failure triggers cancellation of unfinished siblings.

That distinction matters for operations such as:

  • fetching several pieces required to build one response;
  • running validation steps where any failure invalidates the whole operation;
  • starting related service tasks that should share a lifetime.

If tasks are intentionally independent and one failure should not stop the others, a task group may not be the right abstraction without additional per-task error handling.

Keep task groups aligned with ownership boundaries

A useful task group usually corresponds to one logical operation. For example, an HTTP handler might fetch a profile and permissions concurrently because both results are required for the response.

Avoid creating one process-wide task group simply to hold every asynchronous activity. That recreates the same ownership problem at a larger scale: unrelated tasks acquire coupled failure and cancellation behavior.

Nested task groups can express nested ownership. A parent operation can own several components, while each component owns its own related children.

Put time limits around the operation, not arbitrary children

When the whole concurrent operation has one deadline, combine structured concurrency with asyncio.timeout():

async with asyncio.timeout(2.0):
    async with asyncio.TaskGroup() as tg:
        tg.create_task(fetch_profile())
        tg.create_task(fetch_permissions())

If the timeout expires, cancellation unwinds the operation and the task group waits for its children to finish cancellation and cleanup before control leaves the nested contexts.

This expresses the intended policy more clearly than assigning unrelated timeout values to every child when they all share one request deadline.

Be careful with blocking work

TaskGroup organizes asynchronous tasks; it does not make blocking functions asynchronous. Calling a blocking database driver, filesystem operation, or CPU-heavy function directly inside a coroutine can block the event-loop thread and delay every task.

Use genuinely asynchronous APIs where appropriate. For blocking I/O that must coexist with an event loop, asyncio.to_thread() can move that call to a worker thread, but thread cancellation has different limits: cancelling the awaiting coroutine does not forcibly stop arbitrary Python code already running in that thread.

For CPU-bound work, consider process-based parallelism or another architecture appropriate to the workload rather than expecting an asyncio task to provide parallel execution.

Common pitfalls

Creating detached tasks inside structured code

Calling asyncio.create_task() inside a function that otherwise uses a task group can let that task escape the group’s lifetime. Use tg.create_task() when the work belongs to the group.

Catching every child error inside the child

If every coroutine catches and suppresses all failures, the task group cannot know that the operation failed. Catch errors locally only when the child can genuinely recover or intentionally convert the error into a result.

Ignoring cleanup cancellation

Cleanup should be reliable, but it should not silently turn cancellation into success. Prefer finally blocks and propagate cancellation when explicitly caught.

Assuming cancellation is immediate

Cancellation is cooperative. Coroutines that do not reach suspension points promptly can delay group shutdown.

Using structured concurrency for unrelated background jobs

Tasks that should survive the caller need an explicit longer-lived owner, queue, supervisor, or service lifecycle. Hiding them inside a short-lived task group defeats that requirement.

Conclusion

asyncio.TaskGroup makes concurrent ownership visible in Python code. Use it when several asynchronous tasks form one logical operation and should finish within the same scope. Its sibling-cancellation and grouped-error semantics are especially valuable when partial completion is not useful.

The main design question is not whether several coroutines can run concurrently. It is whether they share a lifetime and failure boundary. When they do, a task group gives that relationship a concrete structure that is easier to reason about, test, and clean up.

Python Asyncio Concurrency Reliability