FastAPI AnyIO Structured Concurrency & TaskGroup Cancellation Standard

--- description: FastAPI AnyIO Structured Concurrency & TaskGroup Cancellation Standard - Enforces strict structured concurrency in FastAPI using AnyIO task groups and explicit cancellation scopes bound to client disconnect lifecycles. Prevents orphaned background coroutines, CPU starvation, and connection pool exhaustion by guaranteeing that unhandled client terminations immediately propagate cancellation through hierarchical task trees while shielding critical cleanup routines. globs: app/**/*.py, src/**/*.py alwaysApply: true verified_by: "agentrules.org" license: "CC-BY-4.0" source: "https://agentrules.org/rules/fastapi-anyio-structured-concurrency-cancellation" --- # FastAPI AnyIO Structured Concurrency & TaskGroup Cancellation Standard > **Severity**: CRITICAL | **Category**: backend > **Anti-Pattern Warning**: Spawning unbound coroutines via asyncio.create_task and ignoring client disconnect events ## System Constraints You are a Principal Software Architect enforcing zero-orphan structured concurrency in FastAPI services. You prohibit untracked background execution, raw asyncio.create_task without parent scopes, and unhandled client disconnects. You mandate anyio.create_task_group for fan-out operations, explicit CancelScope usage for timeouts and aborts, and protected shields for transactional cleanups. ## Strict Guidelines - Engineers MUST ALWAYS manage concurrent sub-tasks using `anyio.create_task_group()` instead of `asyncio.create_task` or `asyncio.gather` to enforce bounded coroutine lifetimes. - Engineers MUST NOT initiate untracked asynchronous operations directly within request handlers; all background work must either be bounded by the request lifecycle or dispatched to a durable distributed task queue (e.g., Celery, ARQ). - Services MUST continuously monitor incoming request disconnect states (via `request.is_disconnected()`) during long-polling, streaming, or multi-step operations to trigger immediate cancellation of worker tasks. - Engineers MUST ALWAYS isolate essential rollback or cleanup operations inside `anyio.CancelScope(shield=True)` within `finally` blocks to guarantee execution even after cancellation triggers. - Handlers MUST NEVER swallow `anyio.get_cancelled_exc_class()` or `asyncio.CancelledError`; caught cancellation exceptions must always be re-raised after executing cleanup routines. ## Bad vs Good Implementations ### Unbounded Background Tasks vs Structured AnyIO TaskGroup with Disconnect Watcher #### Bad Antipattern (AI Common Hallucination) > **Why it fails**: Using asyncio.create_task detaches coroutine execution from the request lifecycle. When an HTTP client drops connection, the task continues running, consuming thread/network pool capacity, retaining database locks, and potentially writing corrupted partial states. ```python import asyncio from fastapi import FastAPI, Request, HTTPException app = FastAPI() async def fetch_upstream(client_id: str) -> dict: await asyncio.sleep(10) # Simulates long external API call return {"status": "ok", "client_id": client_id} @app.post("/process") async def process_data(request: Request): # ANTI-PATTERN: Fire-and-forget asyncio.create_task leaks memory and connection pools. # If the HTTP client disconnects, fetch_upstream continues execution indefinitely. task = asyncio.create_task(fetch_upstream("client_123")) result = await task return result ``` #### Production Standard (Correct) > **Standard**: Uses an AnyIO TaskGroup to bind the lifecycle of concurrent workers and a client disconnect monitor. If the client disconnects, `tg.cancel_scope.cancel()` tears down all running tasks immediately. Critical cleanup is shielded from cancellation in the `finally` block. ```python import anyio from fastapi import FastAPI, Request, HTTPException app = FastAPI() async def fetch_upstream(client_id: str) -> dict: await anyio.sleep(10) return {"status": "ok", "client_id": client_id} async def monitor_disconnect(request: Request, cancel_scope: anyio.CancelScope) -> None: while True: if await request.is_disconnected(): cancel_scope.cancel() break await anyio.sleep(0.5) @app.post("/process") async def process_data(request: Request): result_container: dict = {} try: async with anyio.create_task_group() as tg: # Cancel the entire task group if client disconnects tg.start_soon(monitor_disconnect, request, tg.cancel_scope) async def worker(): res = await fetch_upstream("client_123") result_container.update(res) tg.cancel_scope.cancel() # Exit disconnect loop when worker finishes tg.start_soon(worker) except anyio.get_cancelled_exc_class(): raise HTTPException(status_code=499, detail="Client Closed Request") finally: with anyio.CancelScope(shield=True): # Guaranteed cleanup even under cancellation pass return result_container ```

Architectural Specification & Invariant Rationale

Architectural Invariant Design: FastAPI AnyIO Structured Concurrency & TaskGroup Cancellation Standard

FastAPI AnyIO Structured Concurrency & TaskGroup Cancellation Standard defines an industrial-grade behavioral standard for backend engineering. Enforces strict structured concurrency in FastAPI using AnyIO task groups and explicit cancellation scopes bound to client disconnect lifecycles. Prevents orphaned background coroutines, CPU starvation, and connection pool exhaustion by guaranteeing that unhandled client terminations immediately propagate cancellation through hierarchical task trees while shielding critical cleanup routines. 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: Spawning unbound coroutines via asyncio.create_task and ignoring client disconnect events 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	Spawning unbound coroutines via asyncio.create_task and ignoring client disconnect events	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 manage concurrent sub-tasks using `anyio.create_task_group()` instead of `asyncio.create_task` or `asyncio.gather` to enforce bounded coroutine lifetimes.

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.