Python asyncio TaskGroup

asyncio.TaskGroup runs a related group of asynchronous tasks and waits for them as one structured unit. It provides stronger safety guarantees than creating several unrelated tasks and manually awaiting them.

Task groups were added in Python 3.11. Their asynchronous context manager ensures that child tasks finish, fail, or are cancelled before execution continues outside the group.

Create a Task Group

Enter a task group with async with and call its create_task() method for each coroutine.

Example

import asyncio


async def fetch_item(item_id):
    # Simulate an asynchronous operation
    await asyncio.sleep(0.2)
    return {"id": item_id, "status": "ready"}


async def main():
    async with asyncio.TaskGroup() as group:
        first = group.create_task(fetch_item(101))
        second = group.create_task(fetch_item(102))

    # The context exits only after both tasks finish
    print(first.result())
    print(second.result())


asyncio.run(main())

Creating each task schedules it to run concurrently on the event loop. Exiting the async with block implicitly waits for every task in the group.

TaskGroup Lifecycle

A task group moves through a clear lifecycle:

1. The async with statement activates the group.
2. create_task() adds and schedules child tasks.
3. More tasks may be added while the group is active.
4. Leaving the block waits for every remaining task.
5. After the final task finishes and the block exits, the group is closed.

Calling create_task() when the group is inactive closes the supplied coroutine rather than leaving it unawaited. This behavior applies in Python 3.13 and later.

Pass Task Names and Context

TaskGroup.create_task() accepts task-creation options such as name and context. Python 3.14 passes additional keyword arguments to the event loop's task creation method.

Example

import asyncio


async def process_order(order_id):
    # Return a result after yielding control
    await asyncio.sleep(0)
    return f"Processed {order_id}"


async def main():
    async with asyncio.TaskGroup() as group:
        task = group.create_task(
            process_order(42),
            name="order-42"
        )

    print(task.get_name())
    print(task.result())


asyncio.run(main())

Collect Results in Input Order

A task group waits for tasks as a unit but does not return a result list. Store the returned task objects and read their results after successful group completion.

Example

import asyncio


async def square(number):
    # Different delays demonstrate independent completion times
    await asyncio.sleep(0.1 * (4 - number))
    return number * number


async def main():
    tasks = []

    async with asyncio.TaskGroup() as group:
        for number in range(1, 4):
            tasks.append(group.create_task(square(number)))

    # Preserve input order by reading the stored task list
    results = [task.result() for task in tasks]
    print(results)  # [1, 4, 9]


asyncio.run(main())

Understand Failure and Cancellation

If one child task raises an exception other than asyncio.CancelledError, the task group cancels its unfinished sibling tasks. It then waits for those tasks to finish their cancellation and cleanup.

After all tasks finish, non-cancellation errors are raised together as an ExceptionGroup or BaseExceptionGroup.

Example

import asyncio


async def worker(name, delay, fail=False):
    try:
        print(f"{name} started")
        await asyncio.sleep(delay)

        if fail:
            raise ValueError(f"{name} failed")

        print(f"{name} completed")
    finally:
        # This cleanup also runs during cancellation
        print(f"{name} cleaned up")


async def main():
    try:
        async with asyncio.TaskGroup() as group:
            group.create_task(worker("first", 2))
            group.create_task(worker("second", 0.2, fail=True))
            group.create_task(worker("third", 3))
    except* ValueError as errors:
        for error in errors.exceptions:
            print(error)


asyncio.run(main())

When second fails, the unfinished tasks receive cancellation requests. Their finally blocks run before the exception group leaves the task-group context.

Handle Exception Groups with except*

The except* syntax handles matching exceptions inside an exception group. Different error types can be processed independently.

Example

import asyncio


async def validate(value):
    # Raise different errors for demonstration
    await asyncio.sleep(0)

    if value < 0:
        raise ValueError(f"Negative value: {value}")

    if value == 0:
        raise ZeroDivisionError("Value cannot be zero")

    return value


async def main():
    try:
        async with asyncio.TaskGroup() as group:
            group.create_task(validate(-1))
            group.create_task(validate(0))
    except* ValueError as errors:
        print("Validation errors:", errors.exceptions)
    except* ZeroDivisionError as errors:
        print("Zero errors:", errors.exceptions)


asyncio.run(main())

KeyboardInterrupt and SystemExit receive special treatment. The group cancels and waits for remaining tasks, then re-raises the original base exception instead of wrapping it in a group.

Do Not Swallow Cancellation

Task groups use cancellation internally. A coroutine should normally perform cleanup in finally and allow CancelledError to propagate.

Example

import asyncio


async def save_changes():
    try:
        # Perform cancellable asynchronous work
        await asyncio.sleep(10)
    except asyncio.CancelledError:
        print("Save operation was cancelled")

        # Complete necessary cleanup, then preserve cancellation
        raise
    finally:
        print("Closing save resources")

Tip: Suppressing CancelledError can interfere with TaskGroup and timeout behavior. Catch it only when cleanup requires it, and normally re-raise it afterward.

Add Tasks from a Child Coroutine

A child coroutine can receive the active task group and add more related work. New tasks may be added until group shutdown begins and the last task finishes.

Example

import asyncio


async def download(name):
    # Simulate downloading one file
    await asyncio.sleep(0.1)
    print(f"Downloaded {name}")


async def add_downloads(group, names):
    for name in names:
        # Add related tasks to the same active group
        group.create_task(download(name))


async def main():
    async with asyncio.TaskGroup() as group:
        group.create_task(
            add_downloads(group, ["one.txt", "two.txt"])
        )


asyncio.run(main())

Nest Task Groups

Task groups can be nested to represent related layers of work. Each inner group processes its own child failures before an outer group handles the resulting failure.

Example

import asyncio


async def run_batch(batch_id):
    async with asyncio.TaskGroup() as group:
        for item_id in range(3):
            # Each batch owns its item tasks
            group.create_task(process_item(batch_id, item_id))


async def process_item(batch_id, item_id):
    await asyncio.sleep(0.1)
    print(f"Batch {batch_id}, item {item_id}")


async def main():
    async with asyncio.TaskGroup() as group:
        # The outer group owns complete batches
        group.create_task(run_batch("A"))
        group.create_task(run_batch("B"))


asyncio.run(main())

Use a Timeout Around a Task Group

Place asyncio.timeout() around a task group when all its work must complete within one deadline.

Example

import asyncio


async def slow_operation(number):
    # Simulate variable-duration I/O
    await asyncio.sleep(number)
    return number


async def main():
    try:
        async with asyncio.timeout(1.0):
            async with asyncio.TaskGroup() as group:
                group.create_task(slow_operation(0.5))
                group.create_task(slow_operation(2.0))
    except TimeoutError:
        print("The group exceeded its deadline")


asyncio.run(main())

When the timeout expires, cancellation propagates through the task group, which waits for its children to perform cleanup before the timeout is reported.

TaskGroup Compared with gather()

Use a task group when tasks form one operation and should share a failure boundary. Use gather() when its ordered aggregate-result behavior and failure semantics are specifically appropriate.

Terminating a Group

TaskGroup has no direct termination method. The official documentation demonstrates termination by adding a task that raises a custom exception and then handling that exception with except*.

Example

import asyncio


class StopGroup(Exception):
    """Request early termination of the task group."""


async def stop_group():
    # Raising causes sibling tasks to be cancelled
    raise StopGroup()


async def work():
    await asyncio.sleep(10)


async def main():
    try:
        async with asyncio.TaskGroup() as group:
            group.create_task(work())
            group.create_task(stop_group())
    except* StopGroup:
        # Ignore only the intentional termination signal
        pass


asyncio.run(main())

Best Practices

• Use one task group for tasks that belong to the same operation.
• Read task results only after successful group exit.
• Place resource cleanup in finally blocks.
• Do not normally suppress CancelledError.
• Handle expected child failures with focused except* clauses.
• Nest groups when work has meaningful subgroups with their own lifetimes.

Conclusion

asyncio.TaskGroup gives related asynchronous tasks a shared lifetime and failure boundary. It waits for every child, cancels unfinished siblings after failure, groups concurrent exceptions, and integrates with cancellation and timeouts. These guarantees make it the preferred foundation for structured concurrent work in modern Python.