FastAPI AnyIO Structured Concurrency TaskGroup Scoping Guardrail

--- description: FastAPI AnyIO Structured Concurrency TaskGroup Scoping Guardrail - Enforces strict structured concurrency in FastAPI services by requiring AnyIO TaskGroups for all concurrent execution paths instead of untracked asyncio.create_task calls. Prevents orphaned background coroutines, silent exception swallowing, and memory leaks by ensuring asynchronous child tasks are deterministically bound to the parent request context or explicit application lifecycle scopes. globs: app/**/*.py, src/**/*.py, api/**/*.py alwaysApply: true verified_by: "agentrules.org" license: "CC-BY-4.0" source: "https://agentrules.org/rules/fastapi-anyio-taskgroup-concurrency-guardrail" --- # FastAPI AnyIO Structured Concurrency TaskGroup Scoping Guardrail > **Severity**: CRITICAL | **Category**: backend > **Anti-Pattern Warning**: Spawning Unbound Coroutines via asyncio.create_task Causing Silent Task Death and Context Leaks ## System Constraints You are a Principal Backend Systems Architect enforcing strict structured concurrency invariants across FastAPI applications using AnyIO. Mandate deterministic lifecycles for all background tasks and sub-operations. Disallow direct asyncio.create_task and unhandled fire-and-forget patterns within HTTP request lifecycles. Ensure every spawned task is bound to an anyio.create_task_group() context manager, an explicit durable background worker queue, or a managed lifespan TaskGroup. ## Strict Guidelines - Engineers MUST ALWAYS encapsulate concurrent child operations within an 'async with anyio.create_task_group() as tg:' block to enforce structured concurrency boundaries. - Engineers MUST NOT call 'asyncio.create_task()' or 'asyncio.ensure_future()' within FastAPI endpoint dependencies or route handlers. - Code MUST handle MultiError / ExceptionGroup types re-raised by AnyIO TaskGroups to prevent unhandled aggregate exceptions crashing server workers. - Services MUST route true asynchronous background jobs outliving the HTTP request lifecycle to an external broker or an application-lifespan scoped AnyIO TaskGroup rather than FastAPI BackgroundTasks for critical mutations. - Components NEVER leak ContextVars or unclosed database connections across unbound concurrent tasks; all tasks inside a TaskGroup must receive explicit context or borrow safely from bounded pools. ## Bad vs Good Implementations ### Concurrent Fan-Out Inside FastAPI Request Lifecycle #### Bad Antipattern (AI Common Hallucination) > **Why it fails**: Calling asyncio.create_task directly detaches execution from the request scope. If audit_log raises an error, the exception is swallowed or dumped as an unhandled task error. If the client disconnects or an error occurs in the endpoint, spawned tasks may keep running indefinitely, leaking resources and DB connections. ```python import asyncio from fastapi import APIRouter, HTTPException router = APIRouter() async def fetch_user_meta(user_id: str) -> dict: # Simulated external I/O await asyncio.sleep(0.1) return {"id": user_id, "tier": "gold"} async def audit_log(action: str, user_id: str) -> None: await asyncio.sleep(0.05) if user_id == "invalid": raise RuntimeError("Failed to write audit log") @router.post("/users/{user_id}/process") async def process_user(user_id: str): # ANTI-PATTERN: Fire-and-forget task is orphaned. # If it raises an exception, it causes unhandled task warnings and leaks context. # The request returns before the task completes, violating structured boundaries. asyncio.create_task(audit_log("process", user_id)) meta_task = asyncio.create_task(fetch_user_meta(user_id)) meta = await meta_task return {"status": "success", "user": meta} ``` #### Production Standard (Correct) > **Standard**: The anyio.create_task_group() establishes a strict structured concurrency boundary. If audit_log fails, fetch_user_meta is immediately cancelled, preventing resource waste. Exceptions are caught deterministically using Python 3.11+ except* syntax, guaranteeing no orphaned tasks or leaks. ```python import anyio from fastapi import APIRouter, HTTPException router = APIRouter() async def fetch_user_meta(user_id: str, results: dict) -> None: await anyio.sleep(0.1) results["meta"] = {"id": user_id, "tier": "gold"} async def audit_log(action: str, user_id: str) -> None: await anyio.sleep(0.05) if user_id == "invalid": raise ValueError("Invalid user passed to audit log") @router.post("/users/{user_id}/process") async def process_user(user_id: str): results: dict = {} try: # INVARIANT: All tasks are bound to the deterministic AnyIO TaskGroup. # If any sub-task fails, all siblings are cancelled immediately and errors are coalesced. async with anyio.create_task_group() as tg: tg.start_soon(audit_log, "process", user_id) tg.start_soon(fetch_user_meta, user_id, results) except* ValueError as eg: raise HTTPException(status_code=400, detail=str(eg.exceptions[0])) except* Exception as eg: raise HTTPException(status_code=500, detail="Internal concurrent processing failure") return {"status": "success", "user": results.get("meta")} ```

Architectural Specification & Invariant Rationale

Architectural Invariant Design: FastAPI AnyIO Structured Concurrency TaskGroup Scoping Guardrail

FastAPI AnyIO Structured Concurrency TaskGroup Scoping Guardrail defines an industrial-grade behavioral standard for backend engineering. Enforces strict structured concurrency in FastAPI services by requiring AnyIO TaskGroups for all concurrent execution paths instead of untracked asyncio.create_task calls. Prevents orphaned background coroutines, silent exception swallowing, and memory leaks by ensuring asynchronous child tasks are deterministically bound to the parent request context or explicit application lifecycle scopes. 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 Causing Silent Task Death and Context Leaks 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 Causing Silent Task Death and Context Leaks	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 operations within an 'async with anyio.create_task_group() as tg

Engineers MUST ALWAYS encapsulate concurrent child operations within an 'async with anyio.create_task_group() as tg:' block to enforce structured concurrency boundaries.

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.