FastAPI AnyIO Structured Concurrency ExceptionGroup Containment

--- description: FastAPI AnyIO Structured Concurrency ExceptionGroup Containment - Enforces strict boundary containment and unwrapping of Python 3.11+ ExceptionGroup and anyio.ExceptionGroup instances inside FastAPI endpoints, background tasks, and ASGI worker loops. Failure to handle task group exceptions at the concurrency boundary causes them to bubble through Starlette middleware without invoking typed exception handlers, tearing down ASGI worker processes and dropping in-flight client connections. globs: app/**/*.py, src/**/*.py, api/**/*.py, workers/**/*.py alwaysApply: true verified_by: "agentrules.org" license: "CC-BY-4.0" source: "https://agentrules.org/rules/fastapi-anyio-exceptiongroup-containment" --- # FastAPI AnyIO Structured Concurrency ExceptionGroup Containment > **Severity**: CRITICAL | **Category**: backend > **Anti-Pattern Warning**: Leaking Unhandled ExceptionGroups Across ASGI Middleware Boundaries Crashing Worker Pools ## System Constraints You are a Principal Backend Systems Architect specializing in Python ASGI runtimes, FastAPI, and AnyIO structured concurrency. Your goal is to enforce rigid ExceptionGroup containment boundaries around all `anyio.create_task_group()` and asyncio `TaskGroup` invocations. You must ensure that no task group escapes its enclosing scope without explicit exception unwrapping, flattening, or transformation into known HTTPExceptions or contained domain errors before hitting Starlette middleware. ## Strict Guidelines - Engineers MUST ALWAYS handle `ExceptionGroup` (or `BaseExceptionGroup` on Python 3.11+) using `except*` syntax or explicit `.subgroup()` / `.split()` extraction immediately at the `async with anyio.create_task_group()` block boundary. - Engineers MUST NOT allow an unhandled `ExceptionGroup` to propagate directly into FastAPI route return signatures or Starlette middleware chains where standard `@app.exception_handler` decorators fail to match grouped leaf exceptions. - Engineers MUST wrap fire-and-forget ASGI background tasks in a top-level boundary handler that intercepts `ExceptionGroup` to prevent unhandled task crashes from terminating the worker process. - Services MUST log all collapsed leaf exceptions with full context before mapping an `ExceptionGroup` to a unified `HTTPException` or internal failure response. - Components NEVER re-raise raw `ExceptionGroup` across microservice boundaries or background job queues without serializing failure diagnostics for each child coroutine. ## Bad vs Good Implementations ### Handling AnyIO TaskGroup Failures Across FastAPI Middleware Boundaries #### Bad Antipattern (AI Common Hallucination) > **Why it fails**: When concurrent tasks inside anyio.create_task_group() raise exceptions, AnyIO bundles them into an ExceptionGroup. Starlette middleware and standard FastAPI exception handlers evaluate raw exception types and do not unpack groups. This results in unhandled ASGI exceptions that crash background workers, abort client connections, and bypass centralized telemetry. ```python import anyio from fastapi import APIRouter, HTTPException router = APIRouter() async def fetch_user_profile(user_id: str) -> dict: raise ValueError("User record corrupted") async def fetch_user_perms(user_id: str) -> list[str]: raise PermissionError("Permission store unavailable") @router.get("/users/{user_id}") async def get_user_aggregate(user_id: str): profile: dict = {} perms: list[str] = [] # ANTI-PATTERN: Leaks ExceptionGroup into Starlette middleware. # FastAPI standard exception handlers cannot intercept individual # errors nested in ExceptionGroup, causing 500 crashes and worker drops. async with anyio.create_task_group() as tg: async def run_profile(): nonlocal profile profile = await fetch_user_profile(user_id) async def run_perms(): nonlocal perms perms = await fetch_user_perms(user_id) tg.start_soon(run_profile) tg.start_soon(run_perms) return {"profile": profile, "perms": perms} ``` #### Production Standard (Correct) > **Standard**: Structured concurrency guarantees that all concurrent subtasks are joined before exiting the scope. The `except*` pattern captures and classifies specific exception types from the `ExceptionGroup` boundary, mapping them cleanly to HTTP responses and logging the failure details without corrupting ASGI worker state. ```python import logging import anyio from fastapi import APIRouter, HTTPException, status logger = logging.getLogger(__name__) router = APIRouter() async def fetch_user_profile(user_id: str) -> dict: raise ValueError("User record corrupted") async def fetch_user_perms(user_id: str) -> list[str]: raise PermissionError("Permission store unavailable") @router.get("/users/{user_id}") async def get_user_aggregate(user_id: str): profile: dict = {} perms: list[str] = [] # PRODUCTION FIX: Strict containment boundary around TaskGroup. # Evaluates and isolates leaf errors via Python 3.11+ except* syntax. try: async with anyio.create_task_group() as tg: async def run_profile(): nonlocal profile profile = await fetch_user_profile(user_id) async def run_perms(): nonlocal perms perms = await fetch_user_perms(user_id) tg.start_soon(run_profile) tg.start_soon(run_perms) except* PermissionError as eg: logger.warning("Permission lookup failed during user aggregate: %s", eg.exceptions) raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Access denied to permissions") except* ValueError as eg: logger.error("Integrity error during user aggregate: %s", eg.exceptions, exc_info=True) raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="User record invalid") except* Exception as eg: logger.critical("Unhandled concurrency failures: %s", eg.exceptions, exc_info=True) raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Aggregation pipeline failure") return {"profile": profile, "perms": perms} ```

Architectural Specification & Invariant Rationale

Architectural Invariant Design: FastAPI AnyIO Structured Concurrency ExceptionGroup Containment

FastAPI AnyIO Structured Concurrency ExceptionGroup Containment defines an industrial-grade behavioral standard for backend engineering. Enforces strict boundary containment and unwrapping of Python 3.11+ ExceptionGroup and anyio.ExceptionGroup instances inside FastAPI endpoints, background tasks, and ASGI worker loops. Failure to handle task group exceptions at the concurrency boundary causes them to bubble through Starlette middleware without invoking typed exception handlers, tearing down ASGI worker processes and dropping in-flight client connections. Modern software architectures require strict adherence to deterministic patterns to guarantee high availability, maintainability, and zero security vulnerabilities.

Framework Paradigm Shift & Drift MechanicsLLMs are trained on millions of open-source repositories containing outdated code, legacy idioms, and unpatched antipatterns. When coding assistants generate implementations for this stack, they default to common statistical tokens: Leaking Unhandled ExceptionGroups Across ASGI Middleware Boundaries Crashing Worker Pools Without an explicit system instruction file, developers spend hours debugging silent failures.

Failure Modes & Production Blast Radius

Hazard Analysis: What Breaks Without This Rule

The following failure modes have been cataloged from hundreds of hours of production code reviews where AI assistants produced non-compliant implementations:

Failure Mode	AI Symptom (The Anti-Pattern)	Severity	Production Impact & Remedy
AI Hallucination & Deprecation Drift	Leaking Unhandled ExceptionGroups Across ASGI Middleware Boundaries Crashing Worker Pools	Critical	Violating this industrial standard introduces regression bugs, security risks, and unexpected runtime failures in production. Enforce explicit behavioral system directives across all AI coding assistants in your repository.
Silent Architecture Inconsistency	AI assistants produce mixed paradigms across files, creating fragmented codebase conventions.	Medium	Increases cognitive load during code reviews and leads to maintenance debt across engineering teams. Pin project conventions in .cursor/rules, CLAUDE.md, and AGENTS.md so all team agents follow identical guardrails.

Architectural Directives in Detail

Step-by-Step Production Directives

Each rule constraint below is injected into the AI agent's system prompt to establish deterministic boundaries:

Engineers MUST ALWAYS handle `ExceptionGroup` (or `BaseExceptionGroup` on Python 3.11+) using `except*` syntax or explicit `.subgroup()` / `.split()` extraction immediately at the `async with anyio.create_task_group()` block boundary.

Engineers MUST ALWAYS handle `ExceptionGroup` (or `BaseExceptionGroup` on Python 3.11+) using `except*` syntax or explicit `.subgroup()` / `.split()` extraction immediately at the `async with anyio.create_task_group()` block boundary.

Rationale:AI models frequently generate unconstrained code here because statistical language distributions lack project-specific architectural context. Mandating this directive ensures that generated code adheres to verified industrial production requirements.