Creating tasks with asyncio.create_task() is easy, but coordinating their lifetime can be difficult. One task may fail while siblings continue running, references may be lost, and shutdown may happen before cleanup completes. asyncio.TaskGroup places related tasks inside a structured scope: when the context exits, every child has finished, been cancelled, or contributed an error.
This guide covers task creation, result collection, cancellation, ExceptionGroup, comparison with asyncio.gather(), nested groups, timeouts, capacity limits, and reliable cleanup.
The problem with detached tasks
A task created with create_task() starts independently. The caller must keep a reference, await it, and handle failure.
import asyncio
async def work(name: str) -> str:
await asyncio.sleep(0.2)
return name.upper()
async def main():
tasks = [
asyncio.create_task(work("a")),
asyncio.create_task(work("b")),
]
results = await asyncio.gather(*tasks)
print(results)
asyncio.run(main())This works, but it is easy for a task to outlive the logical operation that created it. TaskGroup makes ownership explicit.
Your first TaskGroup
async def main():
async with asyncio.TaskGroup() as group:
task_a = group.create_task(work("a"))
task_b = group.create_task(work("b"))
print(task_a.result())
print(task_b.result())The context does not exit until all children have completed. After the block, results are available and no group-owned work remains pending.
Structured concurrency
Structured concurrency means child tasks live inside a clear lexical scope. Code that creates children is responsible for their completion. This improves reasoning about resources, cancellation, and errors.
The idea extends the guide to Python asyncio. Coroutines provide cooperative concurrency; TaskGroup provides a lifecycle structure for related coroutines.
What happens when one task fails?
When a child raises an exception other than cancellation, TaskGroup normally cancels unfinished siblings. After every child has stopped, failures are raised as an ExceptionGroup.
async def fast_failure():
await asyncio.sleep(0.1)
raise ValueError("invalid data")
async def slow_work():
try:
await asyncio.sleep(10)
finally:
print("slow cleanup")
async def main():
async with asyncio.TaskGroup() as group:
group.create_task(fast_failure())
group.create_task(slow_work())The slow task receives cancellation, executes its finally block, and only then does the group propagate errors.
Handling ExceptionGroup with except*
async def main():
try:
async with asyncio.TaskGroup() as group:
group.create_task(fast_failure())
group.create_task(slow_work())
except* ValueError as errors:
for error in errors.exceptions:
print("validation error:", error)except* selects matching exceptions inside a group. Unhandled types continue propagating.
Collecting results
TaskGroup.create_task() returns a Task. Keep references when individual results are needed.
async def fetch(item_id: int) -> dict:
await asyncio.sleep(0.1)
return {"id": item_id}
async def load_all(ids: list[int]) -> list[dict]:
tasks: list[asyncio.Task[dict]] = []
async with asyncio.TaskGroup() as group:
for item_id in ids:
tasks.append(group.create_task(fetch(item_id)))
return [task.result() for task in tasks]The list preserves creation order even when tasks finish in a different order.
TaskGroup versus gather
asyncio.gather() remains useful when you already have awaitables and want one ordered result. With return_exceptions=True, it can return errors as values. TaskGroup is designed around creating and owning tasks in a scope, and it cancels siblings when one fails.
Do not mechanically replace every gather call. Use TaskGroup when operations should live and die together; use gather when its exact aggregation behavior is appropriate.
Adding tasks dynamically
While the context remains active, a child may receive the group and schedule additional work.
async def discover(group: asyncio.TaskGroup, page: int) -> None:
await asyncio.sleep(0.1)
if page < 3:
group.create_task(discover(group, page + 1))
async def main():
async with asyncio.TaskGroup() as group:
group.create_task(discover(group, 1))New tasks cannot be added after the group closes. Always put limits on recursive discovery, queue growth, or fan-out.
External cancellation
If the task containing the group is cancelled, TaskGroup cancels its children and waits for them to finish. Coroutines should release resources in finally.
async def consumer():
resource = await open_resource()
try:
await process(resource)
finally:
await resource.close()Do not catch CancelledError and continue indefinitely. Cleanup should finish and cancellation should normally be re-raised.
Why swallowing CancelledError is dangerous
TaskGroup and other asyncio tools use cancellation internally. A coroutine that consumes CancelledError can break structured shutdown.
async def incorrect():
try:
await asyncio.sleep(10)
except asyncio.CancelledError:
return # hides cancellationThe safer pattern performs cleanup and then uses raise, unless consuming cancellation is a deliberate and documented design.
Applying a timeout to the group
Wrap the entire scope with asyncio.timeout().
async def main():
try:
async with asyncio.timeout(2.0):
async with asyncio.TaskGroup() as group:
group.create_task(operation_a())
group.create_task(operation_b())
except TimeoutError:
print("deadline exceeded")When the deadline expires, the containing task is cancelled. TaskGroup then cancels and waits for children.
Per-task deadlines
When each operation needs a separate limit, put a timeout inside the child coroutine.
async def with_deadline(coro, seconds: float):
async with asyncio.timeout(seconds):
return await coroAn individual timeout becomes a TimeoutError in that child and normally causes sibling cancellation. If timeout is expected and recoverable, catch it inside the child and return an explicit result.
Expected failures as data
Not every failure should terminate the group. In batch processing, an invalid record may be represented as a result while other records continue.
from dataclasses import dataclass
@dataclass
class Result:
item_id: int
value: str | None = None
error: str | None = None
async def safe_process(item_id: int) -> Result:
try:
value = await fetch_text(item_id)
return Result(item_id=item_id, value=value)
except ExpectedError as exc:
return Result(item_id=item_id, error=str(exc))Reserve unhandled exceptions for conditions that invalidate the whole concurrent unit.
Nested groups
TaskGroups can model sub-operations.
async def process_customer(customer_id: int):
async with asyncio.TaskGroup() as group:
group.create_task(load_profile(customer_id))
group.create_task(load_orders(customer_id))
async def main(ids: list[int]):
async with asyncio.TaskGroup() as group:
for customer_id in ids:
group.create_task(process_customer(customer_id))Each level owns its children. Failures may produce nested exception groups that preserve the work structure.
Limiting concurrency
TaskGroup does not limit the number of simultaneous tasks. Creating hundreds of thousands of tasks may consume substantial memory. Use a semaphore, worker queue, or batches.
limit = asyncio.Semaphore(20)
async def limited(item):
async with limit:
return await process(item)The group owns lifetime; the semaphore controls capacity.
Task names and context
create_task() supports task names and context-related options depending on the Python version.
group.create_task(
fetch(42),
name="customer-query-42",
)Names improve logs and debugging. Context variables are normally copied to child tasks, supporting request IDs and tracing.
Common mistakes
- Creating unrelated tasks outside the group: they may escape the lifecycle.
- Swallowing CancelledError: structured cancellation may stop working.
- Expecting TaskGroup to return a list: keep Task objects for results.
- Scheduling unlimited work: use semaphores, queues, or batching.
- Treating every expected failure as fatal: model recoverable outcomes as data.
- Skipping cleanup: use finally and async context managers.
Complete example: service aggregator
import asyncio
async def get_user(user_id: int) -> dict:
await asyncio.sleep(0.1)
return {"id": user_id, "name": "Ana"}
async def get_orders(user_id: int) -> list[dict]:
await asyncio.sleep(0.2)
return [{"id": 1, "total": 99.0}]
async def build_dashboard(user_id: int) -> dict:
async with asyncio.timeout(3.0):
async with asyncio.TaskGroup() as group:
user = group.create_task(get_user(user_id), name="user")
orders = group.create_task(get_orders(user_id), name="orders")
return {
"user": user.result(),
"orders": orders.result(),
}The dashboard returns only after both operations complete. A failure cancels the sibling operation and avoids an accidental partial result.
Conclusion
asyncio.TaskGroup organizes related tasks inside a scope with a clear beginning and end. It waits for children, coordinates cancellation, and groups failures, making asynchronous code easier to reason about.
The official Python TaskGroup documentation explains exception and cancellation behavior. Use groups for operations that belong together, control capacity separately, and handle cleanup and expected failures explicitly.
