FastAPI AnyIO ExceptionGroup Structured Concurrency Guardrail

--- description: FastAPI AnyIO ExceptionGroup Structured Concurrency Guardrail - Enforces strict task-group boundaries and robust unwrapping of unhandled ExceptionGroups across FastAPI endpoints and AnyIO background workers. Prevents orphaned background coroutines, threadpool leakages, and silent worker loop crashes caused by unhandled multi-exception propagation across asynchronous boundaries. Mandates Python 3.11+ 'except*' or 'exceptiongroup' polyfill matching with deterministic cancellation cleanup. globs: app/**/*.py, src/**/*.py, services/**/*.py, workers/**/*.py alwaysApply: true verified_by: "agentrules.org" license: "CC-BY-4.0" source: "https://agentrules.org/rules/fastapi-anyio-exceptiongroup-structured-concurrency" --- # FastAPI AnyIO ExceptionGroup Structured Concurrency Guardrail > **Severity**: CRITICAL | **Category**: backend > **Anti-Pattern Warning**: Bare asyncio.create_task and Naked AnyIO TaskGroups Crashing Worker Loops via Silently Swallowed or Unhandled ExceptionGroups ## System Constraints You are a Principal Backend Architect enforcing strict AnyIO structured concurrency and deterministic ExceptionGroup handling in FastAPI applications. You must prevent orphaned fire-and-forget tasks, unhandled ExceptionGroup bubbles that bypass FastAPI standard exception handlers, and task cancellation leaks. Mandate the use of anyio.create_task_group context managers, Python 3.11+ except* syntax (or ExceptionGroup unwrapping), and proper Starlette middleware exception mapping. ## Strict Guidelines - Engineers MUST ALWAYS encapsulate concurrent child tasks inside an 'async with anyio.create_task_group() as tg:' context block; raw 'asyncio.create_task' or unbounded fire-and-forget coroutines are strictly forbidden. - Engineers MUST handle concurrent errors using 'except* SubExceptionType:' (Python 3.11+) or explicit 'eg.subgroup()' / 'eg.split()' unwrapping to ensure all nested exceptions are inspected and logged. - FastAPI exception handlers MUST explicitly register handlers for 'ExceptionGroup' and 'BaseExceptionGroup' because Starlette's standard HTTPExceptionHandler does not unwrap nested HTTPExceptions automatically. - Engineers MUST NEVER allow task-group worker loops to terminate silently without propagating critical root failures or acknowledging cancellation via 'anyio.get_cancelled_exc_class()'. - Services launching long-lived background workers MUST manage their lifecycle via FastAPI lifespan context managers using an explicit task group that cancels deterministically upon server shutdown. - Engineers MUST NOT shield tasks without documented justification and an explicit timeout boundary via 'anyio.fail_after()' or 'anyio.move_on_after()'. ## Bad vs Good Implementations ### Handling Concurrent Tasks and Unwrapping ExceptionGroups in FastAPI Lifespan / Endpoints #### Bad Antipattern (AI Common Hallucination) > **Why it fails**: Using asyncio.create_task and asyncio.gather allows task1 to continue executing unmonitored if task2 fails, leaking network handles and thread resources. In AnyIO and modern Python runtimes, concurrent errors form an ExceptionGroup which bypasses standard FastAPI exception handlers if not handled via structured concurrency. ```python import asyncio from fastapi import FastAPI, HTTPException app = FastAPI() async def worker_fetch(item_id: int): if item_id == 13: raise ValueError("Corrupt item payload") await asyncio.sleep(0.01) return {"id": item_id} @app.get("/batch-fetch") async def batch_fetch(): # ANTI-PATTERN: Fire-and-forget raw asyncio.create_task creates orphaned coroutines. # If an exception occurs, it is not tied to request lifecycle and fails silently or corrupts the event loop. task1 = asyncio.create_task(worker_fetch(1)) task2 = asyncio.create_task(worker_fetch(13)) results = await asyncio.gather(task1, task2) # Raw gather does not guarantee safe cleanup on task2 failure return results ``` #### Production Standard (Correct) > **Standard**: Enforces strict structured concurrency using AnyIO's TaskGroup. If any subtask raises an exception, the TaskGroup automatically cancels remaining subtasks, awaits their cleanup, and wraps failures in an ExceptionGroup. The Python 3.11+ 'except*' clause isolates specific ValueError instances, translating them to deterministic HTTP 422 errors without leaking unhandled group exceptions to the server engine. ```python import anyio from fastapi import FastAPI, HTTPException, status from contextlib import asynccontextmanager app = FastAPI() async def fetch_item(item_id: int, results: list[dict]): if item_id == 13: raise ValueError(f"Item {item_id} is corrupt") await anyio.sleep(0.01) results.append({"id": item_id}) @app.get("/batch-fetch") async def batch_fetch(): collected: list[dict] = [] try: # GUARDRAIL: Strict TaskGroup boundary guarantees all child tasks are cancelled and awaited if one fails async with anyio.create_task_group() as tg: tg.start_soon(fetch_item, 1, collected) tg.start_soon(fetch_item, 13, collected) except* ValueError as eg: # Python 3.11+ structured unwrapping of ExceptionGroup errors = [str(exc) for exc in eg.exceptions] raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail={"error": "Validation failure during batch processing", "causes": errors} ) from eg return collected ```

Architectural Specification & Invariant Rationale

Architectural Invariant Design: FastAPI AnyIO ExceptionGroup Structured Concurrency Guardrail

FastAPI AnyIO ExceptionGroup Structured Concurrency Guardrail defines an industrial-grade behavioral standard for backend engineering. Enforces strict task-group boundaries and robust unwrapping of unhandled ExceptionGroups across FastAPI endpoints and AnyIO background workers. Prevents orphaned background coroutines, threadpool leakages, and silent worker loop crashes caused by unhandled multi-exception propagation across asynchronous boundaries. Mandates Python 3.11+ 'except*' or 'exceptiongroup' polyfill matching with deterministic cancellation cleanup. 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: Bare asyncio.create_task and Naked AnyIO TaskGroups Crashing Worker Loops via Silently Swallowed or Unhandled ExceptionGroups 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	Bare asyncio.create_task and Naked AnyIO TaskGroups Crashing Worker Loops via Silently Swallowed or Unhandled ExceptionGroups	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 encapsulate concurrent child tasks inside an 'async with anyio.create_task_group() as tg

Engineers MUST ALWAYS encapsulate concurrent child tasks inside an 'async with anyio.create_task_group() as tg:' context block; raw 'asyncio.create_task' or unbounded fire-and-forget coroutines are strictly forbidden.

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.