asyncio TaskGroup Patterns: The Complete 2026 Guide

You have a list of 50 API endpoints to fetch. You write a loop, create 50 tasks with `asyncio.create_task()`, and fire them all off. One of them times out after 30 seconds. But the other 49 keep running — some against a rate limit, some filling up your memory. You did not ask for that chaos. Python 3.11 shipped a solution: `asyncio.TaskGroup` — and by 2026 it is the standard for every serious async codebase.

In this tutorial, you will learn to:

• Understand why TaskGroup implements structured concurrency and what that means for your async code
• Migrate confidently from asyncio.gather() to TaskGroup for related tasks
• Apply three production-ready patterns: parallel HTTP requests, fan-out/fan-in pipelines, and timeout scopes
• Integrate TaskGroup patterns inside FastAPI route handlers and AIOHTTP client sessions

What Is asyncio.TaskGroup?

asyncio.TaskGroup is a context manager introduced in Python 3.11 that groups related coroutines into a shared execution scope. When you enter the scope via async with TaskGroup(), any task created inside it via create_task() is tied to the group’s lifetime. When the scope exits — whether normally, via an exception, or through cancellation — every unfinished task inside the group is cancelled automatically.

That cancellation propagation is the key insight. It is what the concurrency literature calls structured concurrency: the lifetime of a task is bounded by its lexical scope, not some global event loop that outlives the reason you started the work. You can read the full details in the official asyncio task documentation.

import asyncio

async def fetch_user(user_id: int) -> dict:
    await asyncio.sleep(0.1)  # Simulate I/O
    return {"id": user_id, "name": f"User_{user_id}"}

async def main() -> None:
    async with asyncio.TaskGroup() as tg:
        tg.create_task(fetch_user(1))
        tg.create_task(fetch_user(2))
        tg.create_task(fetch_user(3))
    # All three tasks are done or cancelled by the time we reach here
    print("TaskGroup exited")

asyncio.run(main())

The syntax looks simple, but the guarantees underneath are deep. You can also explore PEP 654 — which introduced ExceptionGroup to Python 3.11 alongside TaskGroup — to understand how structured concurrency and exception propagation fit together.

Prerequisites

This article assumes you are comfortable with async def, await, and the basic idea of an event loop. If you need a refresher, the Introduction to Python Programming (2026 Edition) covers these fundamentals. You will also need Python 3.11 or later — check your version with python --version.

---

TaskGroup vs asyncio.gather() — Why the Migration Matters

For years, `asyncio.gather()` was the default tool for running coroutines concurrently. You pass it a list of awaitables and it runs them all, returning results in order. It works — but it has a structural flaw.

The gather() Problem

When you call gather(task_a(), task_b(), task_c()), those three coroutines are fire-and-forget at the call site. If task_a() raises an exception, task_b and task_c keep running until they complete — the event loop has no concept that they belong to a logical group that should share fate.

# Problem: gather() has no structured cancellation
async def bad_pattern():
    results = await asyncio.gather(
        fetch_slow(),      # Times out after 30s
        fetch_fast(),      # Finishes in 2s but keeps running
        fetch_medium(),    # Finishes in 10s
    )
    return results
# If fetch_slow() times out, the other two keep hammering your I/O

In practice, this means you need manual cleanup: try/except blocks, explicit task.cancel() calls, and tracking which tasks are still running. Every forgotten cancellation is a resource leak — open connections, wasted API quota, memory pressure.

How TaskGroup Fixes It

TaskGroup implements structured concurrency. When any task inside the group raises an exception or the context manager exits for any reason, all remaining tasks receive a CancelledError. You do not need to track or cancel them manually. The scope is the lifecycle.

# Solution: TaskGroup cancels sibling tasks on the first failure
async def good_pattern():
    async with asyncio.TaskGroup() as tg:
        tg.create_task(fetch_slow())    # Times out
        tg.create_task(fetch_fast())    # Cancelled automatically when scope exits
        tg.create_task(fetch_medium())  # Cancelled automatically when scope exits
    return "all done or cancelled together"

Note: When the first task in a TaskGroup raises an exception, Python cancels the remaining tasks and re-raises the original exception. You can catch it with except* ExceptionGroup (Python 3.11+) to handle partial failures without losing information.

Feature	asyncio.gather()	asyncio.TaskGroup
Automatic cancellation on failure	No	Yes
Structured lifetime (lexical scope)	No	Yes
Exception grouping with ExceptionGroup	Basic	Full
Task handle access before completion	No (returns results directly)	Yes (.result() on task)
Readability for related tasks	Scattered	Self-contained block

Common Pattern 1 — Parallel HTTP Requests

The most common use case for TaskGroup is fan-out HTTP fetching. You have a list of URLs and you want to fetch all of them concurrently, stopping early if any one fails.

Here is a production-ready pattern using AIOHTTP 3.13+ — the current stable release as of Q1 2026 — inside a TaskGroup with a asyncio.Semaphore to bound concurrency and avoid hammering downstream services:

import asyncio
import aiohttp
from dataclasses import dataclass

@dataclass
class FetchResult:
    url: str
    status: int
    body: bytes

async def fetch_all(urls: list[str], max_concurrency: int = 10) -> list[FetchResult]:
    sem = asyncio.Semaphore(max_concurrency)

    async def fetch_one(session: aiohttp.ClientSession, url: str) -> FetchResult:
        async with sem:
            async with session.get(url) as response:
                return FetchResult(url=url, status=response.status, body=await response.read())

    connector = aiohttp.TCPConnector(limit=max_concurrency)
    async with aiohttp.ClientSession(connector=connector) as session:
        async with asyncio.TaskGroup() as tg:
            tasks = [tg.create_task(fetch_one(session, url)) for url in urls]
        # All tasks completed (or cancelled together) here
        return [task.result() for task in tasks]

A few things to notice in this pattern:

• The TaskGroup is the outer scope — when session exits, all tasks are guaranteed to be complete
• The Semaphore limits how many requests run simultaneously, protecting both your client and the server
• All task results are collected via task.result() after the group exits — no orphaned connections
• aiohttp.TCPConnector(limit=max_concurrency) matches the semaphore so the connector does not buffer more connections than you allow

AIOHTTP 3.13+ (current as of early 2026) removed the async-timeout dependency for Python 3.11+ and uses asyncio.timeout() natively. You can read the AIOHTTP changelog for the full list of changes.

Common Pattern 2 — Fan-Out/Fan-In Pipeline

The fan-out/fan-in pattern distributes work across multiple workers (fan-out) and collects their results (fan-in). TaskGroup is ideal here because the “fan-out” tasks are structurally related — they should all run for the duration of the job and be cancelled together if the job is cancelled.

import asyncio

async def process_item(item: int) -> int:
    await asyncio.sleep(0.1)  # Simulate CPU/IO work
    return item * 2

async def fan_out_fan_in(items: list[int], num_workers: int = 5) -> list[int]:
    """Distribute items across workers, collect results in order."""
    queue: asyncio.Queue[int] = asyncio.Queue()
    results: list[int] = []
    results_lock = asyncio.Lock()

    for item in items:
        await queue.put(item)

    async def worker(worker_id: int) -> None:
        try:
            while True:
                item = queue.get_nowait()
                try:
                    result = await process_item(item)
                    async with results_lock:
                        results.append(result)
                finally:
                    queue.task_done()
        except asyncio.QueueEmpty:
            return

    async with asyncio.TaskGroup() as tg:
        for i in range(num_workers):
            tg.create_task(worker(i))
        # Queue is filled, workers are running
    # TaskGroup ensures all workers finish or are cancelled together

    return results

# Example
async def main():
    items = list(range(20))
    results = await fan_out_fan_in(items, num_workers=5)
    print(sorted(results))  # [0, 2, 4, 6, ..., 38]

asyncio.run(main())

The Queue acts as the work distributor, and the Lock protects the shared results list from concurrent writes. The worker loop drains the queue with get_nowait() — when the queue is empty it raises QueueEmpty and the worker exits cleanly.

When the TaskGroup exits, all workers are guaranteed to have either processed all queue items or been cancelled mid-flight. This is the fan-in guarantee: you always get a complete result set or a clean cancellation — never a partial result with workers still running in the background.

Common Pattern 3 — Timeouts with asyncio.timeout()

Python 3.11 also introduced asyncio.timeout() — a native context manager that cancels a block after a deadline. Combined with TaskGroup, you get deadline-scoped concurrency: an entire group of tasks gets cancelled together when the timeout fires.

import asyncio

async def fetch_with_deadline(urls: list[str], timeout_seconds: float = 5.0) -> dict:
    """
    Fetch all URLs with a shared deadline.
    If timeout triggers, all in-flight requests are cancelled together.
    """
    async def fetch_one(url: str, results: dict) -> None:
        await asyncio.sleep(0.5)  # Simulate I/O
        results[url] = "ok"

    results: dict[str, str] = {}

    try:
        async with asyncio.timeout(timeout_seconds):
            async with asyncio.TaskGroup() as tg:
                for url in urls:
                    tg.create_task(fetch_one(url, results))
    except asyncio.TimeoutError:
        print(f"Deadline of {timeout_seconds}s exceeded — all tasks cancelled")
        return results  # Return whatever was collected before timeout

    return results

async def main():
    urls = [f"https://api.example.com/item/{i}" for i in range(20)]
    results = await fetch_with_deadline(urls, timeout_seconds=2.0)
    print(f"Completed: {len(results)} items")

asyncio.run(main())

Note the key difference from asyncio.wait_for(): asyncio.timeout() does not raise if the timeout fires — it simply cancels the block and exits. The except asyncio.TimeoutError catches it, and you can inspect partial results. This is cleaner than wrapping each individual task in wait_for().

Note: asyncio.wait_for() wraps a single awaitable with a timeout. asyncio.timeout() wraps a block of code with a deadline. Use asyncio.timeout() when you want all tasks inside a scope to share the same deadline — exactly the fan-out pattern.

---

Integration with FastAPI and AIOHTTP in Q1 2026

FastAPI released 16+ versions across late 2025 and Q1 2026 — starting from 0.118.0 in September 2025 and running through 0.135.2 in early March 2026, with several breaking-change releases in late 2025 and February 2026. The framework has leaned harder into native asyncio patterns, and TaskGroup fits naturally inside FastAPI route handlers.

FastAPI Route Handler Pattern

Here is how you would use TaskGroup to fan out multiple service calls inside a FastAPI endpoint, using the asyncio native stack. Note that api.example.com is a placeholder — replace it with your actual service URLs to run this example:

from fastapi import FastAPI
from contextlib import asynccontextmanager
import asyncio
import aiohttp

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup: create a shared AIOHTTP session
    connector = aiohttp.TCPConnector(limit=20)
    app.state.session = aiohttp.ClientSession(connector=connector)
    yield
    # Shutdown: session is closed automatically
    await app.state.session.close()

app = FastAPI(lifespan=lifespan)

async def fetch_user_data(session: aiohttp.ClientSession, user_id: int) -> dict:
    async with session.get(f"https://api.example.com/users/{user_id}") as resp:
        if resp.status == 404:
            raise ValueError(f"User {user_id} not found")
        return await resp.json()

@app.get("/users/{user_id}/profile")
async def get_user_profile(user_id: int):
    session: aiohttp.ClientSession = app.state.session

    async with asyncio.timeout(3.0):  # 3-second deadline for the whole profile
        async with asyncio.TaskGroup() as tg:
            user_task = tg.create_task(fetch_user_data(session, user_id))
            perm_task = tg.create_task(fetch_user_data(session, f"{user_id}/permissions"))
            org_task = tg.create_task(fetch_user_data(session, f"{user_id}/org"))

    return {
        "user": user_task.result(),
        "permissions": perm_task.result(),
        "org": org_task.result(),
    }

This pattern has several advantages for FastAPI developers:

• The shared aiohttp.ClientSession via app.state.session is reused across requests — no connection overhead per call
• The asyncio.timeout(3.0) deadline means the entire profile fetch fails fast if any sub-request times out
• The TaskGroup ensures that if any of the three parallel fetches raises, the other two are cancelled and no resources are wasted
• No external dependencies beyond aiohttp — async-timeout was removed in aiohttp 3.13 for Python 3.11+

AIOHTTP Q1 2026: Key Changes

AIOHTTP 3.13+ (current stable) brought several changes relevant to TaskGroup-based code in early 2026:

• Removed async-timeout dependency for Python 3.11+ in favour of native asyncio.timeout()
• Added support for free-threading in Python 3.14+
• Security fixes for request smuggling vulnerabilities in versions prior to 3.10.11

If you are on an older aiohttp version, the migration to 3.13+ is straightforward and the removal of async-timeout actually simplifies your dependency tree. You can install the latest version with pip install aiohttp>=3.13.

Common Mistakes and Gotchas

• Creating tasks outside the TaskGroup scope — tg.create_task() must be called inside the async with tg: block. Calling it outside the context manager means the task is not part of the group’s cancellation scope.
• Forgetting that TaskGroup re-raises the first exception — if one of your tasks raises, the others are cancelled, but the original exception is re-raised. Wrap with except* ExceptionGroup to handle multiple exceptions gracefully.
• Blocking I/O inside a TaskGroup — if you have CPU-bound work or synchronous I/O, use asyncio.to_thread() to move it off the event loop. Blocking the loop inside a TaskGroup blocks the entire group.
• Using gather() by habit — if tasks are structurally related (they share a logical job), prefer TaskGroup. Reserve gather() for truly independent fire-and-forget operations.
• Not bounding concurrency with a Semaphore — a TaskGroup with no semaphore can create thousands of tasks at once. Always use asyncio.Semaphore when fetching from an external service to protect against rate limits and memory exhaustion.

Practical Example — Building a Parallel Data Fetcher

Let us put it all together with a realistic example: a DataFetcher class that downloads a list of user profiles from multiple API endpoints concurrently, applies a deadline, and returns a typed result. This is the kind of pattern you will find in real microservices querying multiple downstream services to assemble a single API response. Replace https://api.example.com with your actual service base URL to run this code:

import asyncio
import aiohttp
from dataclasses import dataclass
from typing import Optional

@dataclass
class UserProfile:
    user_id: int
    name: str
    email: str
    roles: list[str]
    org_name: str

class DataFetchError(Exception):
    """Raised when the data fetcher fails to collect all required fields."""
    pass

class DataFetcher:
    def __init__(self, base_url: str, timeout_seconds: float = 5.0, max_concurrency: int = 10):
        self.base_url = base_url
        self.timeout_seconds = timeout_seconds
        self.max_concurrency = max_concurrency
        self._session: Optional[aiohttp.ClientSession] = None

    async def _ensure_session(self) -> aiohttp.ClientSession:
        if self._session is None or self._session.closed:
            connector = aiohttp.TCPConnector(limit=self.max_concurrency)
            self._session = aiohttp.ClientSession(connector=connector)
        return self._session

    async def _fetch_json(self, path: str, sem: asyncio.Semaphore) -> dict:
        session = await self._ensure_session()
        async with sem:
            async with session.get(f"{self.base_url}{path}") as resp:
                resp.raise_for_status()
                return await resp.json()

    async def fetch_profile(self, user_id: int) -> UserProfile:
        sem = asyncio.Semaphore(self.max_concurrency)

        user_data: dict = {}
        perm_data: dict = {}
        org_data: dict = {}

        async def fetch_user():
            nonlocal user_data
            user_data = await self._fetch_json(f"/users/{user_id}", sem)

        async def fetch_permissions():
            nonlocal perm_data
            perm_data = await self._fetch_json(f"/users/{user_id}/permissions", sem)

        async def fetch_org():
            nonlocal org_data
            org_data = await self._fetch_json(f"/users/{user_id}/org", sem)

        try:
            async with asyncio.timeout(self.timeout_seconds):
                async with asyncio.TaskGroup() as tg:
                    tg.create_task(fetch_user())
                    tg.create_task(fetch_permissions())
                    tg.create_task(fetch_org())
        except* (aiohttp.ClientError, asyncio.TimeoutError) as eg:
            raise DataFetchError(f"Failed to fetch profile for user {user_id}") from eg

        return UserProfile(
            user_id=user_id,
            name=user_data.get("name", ""),
            email=user_data.get("email", ""),
            roles=perm_data.get("roles", []),
            org_name=org_data.get("name", ""),
        )

    async def fetch_profiles(self, user_ids: list[int]) -> dict[int, UserProfile]:
        results: dict[int, UserProfile] = {}

        async with asyncio.TaskGroup() as tg:
            async def fetch_and_store(uid: int) -> None:
                results[uid] = await self.fetch_profile(uid)

            for uid in user_ids:
                tg.create_task(fetch_and_store(uid))

        return results

    async def close(self) -> None:
        if self._session and not self._session.closed:
            await self._session.close()

    async def __aenter__(self):
        return self

    async def __aexit__(self, *args):
        await self.close()


# Usage example — replace with your actual service base URL
async def main():
    fetcher = DataFetcher("https://api.example.com", timeout_seconds=3.0)

    try:
        async with fetcher:
            profiles = await fetcher.fetch_profiles([1, 2, 3, 4, 5])
            for uid, profile in profiles.items():
                print(f"User {uid}: {profile.name} at {profile.org_name}")
    except DataFetchError as e:
        print(f"Data fetch failed: {e}")

asyncio.run(main())

Walking through the key decisions in this implementation:

• Shared session — the DataFetcher reuses a single ClientSession across all requests via __aenter__/__aexit__, so connection pooling is efficient.
• Deadline at the top level — the asyncio.timeout() wraps the entire TaskGroup, so all three parallel fetches are cancelled together if any one exceeds 3 seconds.
• ExceptionGroup handling — except* (Python 3.11+) catches the ExceptionGroup that TaskGroup raises when multiple tasks fail, giving you a clean error without losing the original exception.
• Batch fetching with fetch_profiles — a top-level TaskGroup fans out all individual fetch_profile() calls, each of which has its own inner TaskGroup and timeout. This gives you both per-user deadlines and a global batch deadline.

Summary and Next Steps

By 2026, asyncio.TaskGroup has become the default tool for managing related concurrent tasks in Python. It brings structured concurrency — automatic cancellation, shared lifetimes, and clean exception propagation — to the standard library. The key patterns covered in this article are:

• Parallel HTTP requests — TaskGroup + AIOHTTP ClientSession + Semaphore for bounded fan-out fetching
• Fan-out/fan-in pipelines — TaskGroup + asyncio.Queue for distributing work across a fixed worker pool
• Timeout scopes — asyncio.timeout() wrapping a TaskGroup so all tasks share the same deadline
• FastAPI integration — TaskGroup inside route handlers, with a shared AIOHTTP session via app.state

Pair this with the uv package manager to quickly spin up projects with modern Python versions, and the Ruff formatter to keep your async code clean and consistent.

---

Frequently Asked Questions

Can TaskGroup replace asyncio.gather() entirely?

Mostly yes, but not always. Use TaskGroup when tasks are structurally related and should share a lifetime and cancellation fate — like parallel fetches for a single request. Use gather() for fire-and-forget tasks where you genuinely want all of them to run to completion regardless of what happens to the others.

How does TaskGroup handle multiple exceptions?

When multiple tasks raise exceptions inside a TaskGroup, Python wraps them in an ExceptionGroup and raises that. You can catch it with except* ExceptionGroup (the * syntax from Python 3.11+) to handle each sub-exception individually. If you use a plain except ExceptionGroup, it still works but you cannot easily inspect individual exceptions.

What is the difference between asyncio.timeout() and asyncio.wait_for()?

asyncio.wait_for() wraps a single awaitable and raises TimeoutError when the deadline passes. asyncio.timeout() wraps a block of code (not just one awaitable) and cancels everything inside the block when the deadline passes. For fan-out patterns where multiple tasks share one deadline, asyncio.timeout() is cleaner because you do not need to wrap each task individually.

Is TaskGroup safe to use with FastAPI background tasks?

Yes. FastAPI supports BackgroundTasks and you can also use TaskGroups inside dependency injection. The key constraint is that the TaskGroup context manager must stay alive for the duration of the work — so avoid returning a TaskGroup-scoped coroutine from a dependency that exits before the background work completes.

What happens if an exception is raised inside a TaskGroup?

When any task inside a TaskGroup raises an exception, Python immediately cancels all other tasks in the group and then re-raises the first exception. If multiple tasks raise before cancellation propagates, Python wraps them in an ExceptionGroup. The parent context receives the combined exception, and you can use except* to handle each sub-exception individually — useful for logging or retrying specific failures rather than treating all failures the same way.