FastAPI Streaming Response Disconnect Handling and Resource Cancellation Contract

--- description: FastAPI Streaming Response Disconnect Handling and Resource Cancellation Contract - Enforces deterministic cancellation of upstream compute, external LLM streams, and database server-side cursors upon downstream HTTP/TCP client disconnection in FastAPI. Prevents catastrophic resource exhaustion and dangling background tasks by mandating periodic ASGI disconnect polling or structured async generator cleanup semantics. globs: **/api/**/*.py, **/routers/**/*.py, **/endpoints/**/*.py, **/streams/**/*.py alwaysApply: true verified_by: "agentrules.org" license: "CC-BY-4.0" source: "https://agentrules.org/rules/fastapi-stream-backpressure-cancellation-contract" --- # FastAPI Streaming Response Disconnect Handling and Resource Cancellation Contract > **Severity**: CRITICAL | **Category**: backend > **Anti-Pattern Warning**: Unchecked Async Streaming Yields Leading to Dangling LLM Inferences and Database Cursor Leaks ## System Constraints You are a Principal Software Architect enforcing production resilience for streaming APIs in FastAPI and Starlette. You must mandate that every async generator backing a StreamingResponse strictly accounts for client TCP disconnects. Never permit unbounded loops yielding data without checking `await request.is_disconnected()` or relying on structured task cancellation using `asyncio.TaskGroup` or `try...finally` resource cleanups. ## Strict Guidelines - Engineers MUST ALWAYS accept the raw `fastapi.Request` instance in streaming endpoints and actively verify `await request.is_disconnected()` on each iteration if the upstream source does not inherently raise `asyncio.CancelledError` on client drop. - Engineers MUST wrap all upstream async generator iterations (such as Anthropic/OpenAI SDK streams, asyncpg cursors, or Redis pub/sub) in `try...finally` blocks to explicitly invoke `.aclose()` or abort remote connections upon disconnection. - Code MUST NEVER use fire-and-forget background tasks to consume upstream streams into an in-memory queue without binding the producer lifecycle to the downstream consumer's cancellation context. - Services MUST configure Starlette's `StreamingResponse` with appropriate media types (`text/event-stream` or `application/x-ndjson`) and avoid buffering proxies by disabling response buffering headers (e.g., `X-Accel-Buffering: no`). - Components NEVER execute write mutations inside a read-oriented streaming generator without an idempotent, transactional boundary that rolls back when `asyncio.CancelledError` is injected. ## Bad vs Good Implementations ### FastAPI SSE Streaming with Deterministic Disconnect and Cancellation #### Bad Antipattern (AI Common Hallucination) > **Why it fails**: The generator continues to exhaust tokens and compute from the upstream LLM provider even after the downstream HTTP client aborts or closes their browser tab. The FastAPI StreamingResponse does not automatically terminate upstream iteration without cooperative checks or cancellation trapping. ```python from fastapi import APIRouter from fastapi.responses import StreamingResponse import openai router = APIRouter() client = openai.AsyncOpenAI() @router.post("/generate") async def generate_completion(prompt: str): async def event_publisher(): # BUG: Ignores client disconnection. If client disconnects (TCP FIN/RST), # this loop continues consuming costly upstream LLM tokens until completion. response = await client.chat.completions.create( model="gpt-4o", messages=[{"role": "user", "content": prompt}], stream=True, ) async for chunk in response: content = chunk.choices[0].delta.content or "" yield f"data: {content}\n\n" return StreamingResponse(event_publisher(), media_type="text/event-stream") ``` #### Production Standard (Correct) > **Standard**: Injects `request: Request` to inspect client state with `await request.is_disconnected()`. Utilizes a `try...finally` block to strictly close the upstream client response via `.aclose()`, re-raises `asyncio.CancelledError` for standard ASGI teardown, and adds `X-Accel-Buffering: no` to avoid reverse-proxy buffering. ```python from fastapi import APIRouter, Request from fastapi.responses import StreamingResponse import openai import asyncio import structlog logger = structlog.get_logger() router = APIRouter() client = openai.AsyncOpenAI() @router.post("/generate") async def generate_completion(request: Request, prompt: str): async def event_publisher(): response = await client.chat.completions.create( model="gpt-4o", messages=[{"role": "user", "content": prompt}], stream=True, ) try: async for chunk in response: # Deterministically check ASGI client connection state if await request.is_disconnected(): logger.info("client_disconnected_early", prompt_prefix=prompt[:20]) break content = chunk.choices[0].delta.content or "" if content: yield f"data: {content}\n\n" except asyncio.CancelledError: logger.info("stream_cancelled_by_runtime") raise finally: # Ensure upstream network socket and client response are closed await response.response.aclose() logger.info("upstream_llm_stream_closed") return StreamingResponse( event_publisher(), media_type="text/event-stream", headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"} ) ```

Architectural Specification & Invariant Rationale

Architectural Invariant Design: FastAPI Streaming Response Disconnect Handling and Resource Cancellation Contract

FastAPI Streaming Response Disconnect Handling and Resource Cancellation Contract defines an industrial-grade behavioral standard for backend engineering. Enforces deterministic cancellation of upstream compute, external LLM streams, and database server-side cursors upon downstream HTTP/TCP client disconnection in FastAPI. Prevents catastrophic resource exhaustion and dangling background tasks by mandating periodic ASGI disconnect polling or structured async generator cleanup semantics. 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: Unchecked Async Streaming Yields Leading to Dangling LLM Inferences and Database Cursor 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	Unchecked Async Streaming Yields Leading to Dangling LLM Inferences and Database Cursor 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 accept the raw `fastapi.Request` instance in streaming endpoints and actively verify `await request.is_disconnected()` on each iteration if the upstream source does not inherently raise `asyncio.CancelledError` on client drop.

Engineers MUST ALWAYS accept the raw `fastapi.Request` instance in streaming endpoints and actively verify `await request.is_disconnected()` on each iteration if the upstream source does not inherently raise `asyncio.CancelledError` on client drop.

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.