FastAPI AnyIO Structured Concurrency & TaskGroup Cancellation Boundaries

--- description: FastAPI AnyIO Structured Concurrency & TaskGroup Cancellation Boundaries - Enforces strict structured concurrency boundaries using AnyIO task groups and cancellation scopes in FastAPI endpoints. Prevents zombie background tasks, orphaned fan-out coroutines, and connection leaks when clients abruptly terminate HTTP or Server-Sent Events (SSE) connections. 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-structured-concurrency-taskgroup-bounds" --- # FastAPI AnyIO Structured Concurrency & TaskGroup Cancellation Boundaries > **Severity**: CRITICAL | **Category**: backend > **Anti-Pattern Warning**: Spawning unparented asyncio tasks or using asyncio.gather without structured cancel scopes during HTTP client disconnects. ## System Constraints You are a Principal Python Architect specializing in modern ASGI concurrency, Starlette, and AnyIO. You mandate structured concurrency across all FastAPI asynchronous routes, background workers, and streaming responses. Never allow raw `asyncio.create_task`, unbounded `asyncio.gather`, or detached async generators in request paths. When clients disconnect, any running sub-tasks, telemetry fetchers, or streaming streams MUST terminate deterministically through AnyIO TaskGroups and explicit cancellation boundaries. ## Strict Guidelines - Engineers MUST ALWAYS use `anyio.create_task_group()` instead of `asyncio.create_task()` or `asyncio.gather()` for concurrent fan-out operations within route handlers. - Engineers MUST NOT leak coroutines past the request lifecycle; all concurrent sub-tasks must be bounded by a context manager that guarantees join-on-exit or cancel-on-error. - Streaming endpoints (SSE, Chunked Transfer) MUST encapsulate memory streams using `anyio.create_memory_object_stream()` and link `request.is_disconnected()` or ASGI disconnect events directly to the task group's `cancel_scope`. - Code SHOULD catch `anyio.get_cancelled_exc_class()` when executing graceful cleanup, ensuring resources (DB sessions, HTTP client connections) are closed immediately upon client abort. - Services MUST enforce explicit task timeouts using `with anyio.fail_after(timeout):` or `with anyio.move_on_after(timeout):` rather than global task timeouts. - Components NEVER start background daemon coroutines without passing down an explicit, trackable `anyio.CancelScope`. ## Bad vs Good Implementations ### Fan-Out Query with Client Disconnect Resilience #### Bad Antipattern (AI Common Hallucination) > **Why it fails**: Using asyncio.create_task escapes the structured concurrency tree. If a client disconnects or one task raises an exception, the sibling task continues executing in the background, consuming CPU, I/O descriptors, and database pool connections. ```python import asyncio from fastapi import FastAPI, HTTPException app = FastAPI() @app.get("/aggregate") async def aggregate_metrics(user_id: str): # VIOLATION: asyncio.gather does not cancel remaining tasks if one fails, # and raw tasks remain orphaned in the event loop if the HTTP client disconnects. task1 = asyncio.create_task(fetch_user_profile(user_id)) task2 = asyncio.create_task(fetch_billing_status(user_id)) try: profile, billing = await asyncio.gather(task1, task2) except Exception as exc: raise HTTPException(status_code=502, detail=str(exc)) return {"profile": profile, "billing": billing} ``` #### Production Standard (Correct) > **Standard**: AnyIO TaskGroup binds the lifecycles of all child tasks. If the client disconnects or an error occurs in one branch, AnyIO triggers the cancel scope, cancelling all pending sibling coroutines and preventing zombie execution. ```python import anyio from fastapi import FastAPI, HTTPException, Request app = FastAPI() @app.get("/aggregate") async def aggregate_metrics(request: Request, user_id: str): # COMPLIANT: Bounded by anyio.create_task_group. # Cancellation propagates hierarchically if client disconnects or a worker fails. profile: dict = {} billing: dict = {} async def get_profile() -> None: nonlocal profile profile = await fetch_user_profile(user_id) async def get_billing() -> None: nonlocal billing billing = await fetch_billing_status(user_id) try: async with anyio.create_task_group() as tg: tg.start_soon(get_profile) tg.start_soon(get_billing) except ExceptionGroup as eg: raise HTTPException(status_code=502, detail=f"Aggregation failed: {eg.exceptions}") return {"profile": profile, "billing": billing} ``` ### SSE Streaming with Memory Object Stream and Cancel Scope #### Bad Antipattern (AI Common Hallucination) > **Why it fails**: If the client closes the connection, the generator can remain blocked in `poll_external_queue()` or `asyncio.sleep()`, permanently leaking resources because there is no task group cancellation boundary. ```python import asyncio from fastapi import FastAPI from fastapi.responses import StreamingResponse app = FastAPI() @app.get("/events") async def event_stream(): # VIOLATION: Unbounded infinite async generator with no disconnect check async def generate_events(): while True: data = await poll_external_queue() yield f"data: {data}\n\n" await asyncio.sleep(1) return StreamingResponse(generate_events(), media_type="text/event-stream") ``` #### Production Standard (Correct) > **Standard**: Memory object streams paired with AnyIO TaskGroup ensure that the background producer detects disconnects and exits cleanly, closing channel buffers without lingering tasks. ```python import anyio from fastapi import FastAPI, Request from fastapi.responses import StreamingResponse app = FastAPI() @app.get("/events") async def event_stream(request: Request): # COMPLIANT: Producer and consumer communicate via AnyIO memory streams; # disconnects trigger immediate task group scope cancellation. async def bounded_generator(): send_stream, receive_stream = anyio.create_memory_object_stream[str](max_buffer_size=16) async with anyio.create_task_group() as tg: async def producer() -> None: async with send_stream: while True: if await request.is_disconnected(): tg.cancel_scope.cancel() break data = await poll_external_queue() await send_stream.send(f"data: {data}\n\n") await anyio.sleep(1) tg.start_soon(producer) async with receive_stream: async for message in receive_stream: yield message return StreamingResponse(bounded_generator(), media_type="text/event-stream") ```

Architectural Specification & Invariant Rationale

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

FastAPI AnyIO Structured Concurrency & TaskGroup Cancellation Boundaries defines an industrial-grade behavioral standard for backend engineering. Enforces strict structured concurrency boundaries using AnyIO task groups and cancellation scopes in FastAPI endpoints. Prevents zombie background tasks, orphaned fan-out coroutines, and connection leaks when clients abruptly terminate HTTP or Server-Sent Events (SSE) 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: Spawning unparented asyncio tasks or using asyncio.gather without structured cancel scopes during HTTP client disconnects. 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 unparented asyncio tasks or using asyncio.gather without structured cancel scopes during HTTP client disconnects.	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 use `anyio.create_task_group()` instead of `asyncio.create_task()` or `asyncio.gather()` for concurrent fan-out operations within route handlers.

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.