FastAPI ASGI Token-Streaming Backpressure and Heartbeat Standard

--- description: FastAPI ASGI Token-Streaming Backpressure and Heartbeat Standard - Enforces proactive client disconnection detection, bounded ASGI queue backpressure, and periodic SSE heartbeats for token-streaming endpoints. Prevents runaway upstream LLM costs, kernel socket buffer exhaustion, and zombie generator memory leaks when downstream clients hang or disconnect. globs: **/api/**/*.py, **/routers/**/*.py, **/streaming/**/*.py, **/services/**/*.py alwaysApply: true verified_by: "agentrules.org" license: "CC-BY-4.0" source: "https://agentrules.org/rules/fastapi-asgi-token-streaming-backpressure-standard" --- # FastAPI ASGI Token-Streaming Backpressure and Heartbeat Standard > **Severity**: CRITICAL | **Category**: backend > **Anti-Pattern Warning**: Unbounded StreamingResponse generator lacking disconnect checks, heartbeats, and upstream task cancellation ## System Constraints You are a Principal Backend Architect enforcing robust ASGI streaming standards in FastAPI. You mandate explicit client disconnect polling via `request.is_disconnected()`, strict queue boundaries (`maxsize`) to apply backpressure to upstream LLM inference, and periodic SSE comment heartbeats (`: keep-alive\n\n`) to detect half-open TCP sockets. ## Strict Guidelines - Engineers MUST ALWAYS pass the active `Request` instance into streaming generator functions and poll `await request.is_disconnected()` on every token iteration or yield interval. - Engineers MUST wrap upstream streaming inference in an explicit `asyncio.Task` and cancel it in a `finally` block when the downstream generator terminates. - Engineers MUST NOT yield unbounded raw generator streams directly to `StreamingResponse` without a bounded buffer (`asyncio.Queue(maxsize=...)`) or inline disconnection handling. - Code SHOULD emit periodic SSE comment heartbeats (`: keep-alive\n\n`) every 15 seconds if upstream LLM response synthesis pauses, preventing intermediary gateway timeouts. - Services MUST set `media_type="text/event-stream"` and explicit cache-control headers (`Cache-Control: no-cache`, `X-Accel-Buffering: no`) to prevent proxy buffering. ## Bad vs Good Implementations ### FastAPI LLM Token Streaming with Cooperative Disconnect and Heartbeats #### Bad Antipattern (AI Common Hallucination) > **Why it fails**: Fails to check `request.is_disconnected()`. A severed connection leaves the upstream LLM inference running, accumulating tokens in memory and billing costs until the entire prompt completion finishes. ```python from fastapi import FastAPI from fastapi.responses import StreamingResponse from openai import AsyncOpenAI app = FastAPI() client = AsyncOpenAI() async def stream_tokens(prompt: str): # ANTI-PATTERN: If the client closes the tab, this generator runs to # completion, burning LLM tokens and buffering writes into ASGI memory. stream = await client.chat.completions.create( model="gpt-4o", messages=[{"role": "user", "content": prompt}], stream=True, ) async for chunk in stream: content = chunk.choices[0].delta.content or "" yield f"data: {content}\n\n" @app.post("/chat/stream") async def chat_endpoint(prompt: str): return StreamingResponse(stream_tokens(prompt), media_type="text/event-stream") ``` #### Production Standard (Correct) > **Standard**: Monitors `request.is_disconnected()`, leverages a bounded queue to apply backpressure to the LLM producer task, sends SSE keep-alive comments on timeout, and deterministically cancels upstream generation via `task.cancel()` in the `finally` block. ```python import asyncio from typing import AsyncGenerator from fastapi import FastAPI, Request from fastapi.responses import StreamingResponse from openai import AsyncOpenAI app = FastAPI() client = AsyncOpenAI() async def event_generator(request: Request, prompt: str) -> AsyncGenerator[str, None]: # Bound buffer to enforce backpressure against upstream producer queue: asyncio.Queue[str | None] = asyncio.Queue(maxsize=32) disconnected_event = asyncio.Event() async def token_producer() -> None: try: stream = await client.chat.completions.create( model="gpt-4o", messages=[{"role": "user", "content": prompt}], stream=True, ) async for chunk in stream: if disconnected_event.is_set(): break content = chunk.choices[0].delta.content or "" if content: await queue.put(content) finally: await queue.put(None) producer_task = asyncio.create_task(token_producer()) try: while True: if await request.is_disconnected(): disconnected_event.set() break try: # 15s timeout yields SSE comment to keep TCP alive and check disconnect token = await asyncio.wait_for(queue.get(), timeout=15.0) if token is None: break yield f"data: {token}\n\n" queue.task_done() except asyncio.TimeoutError: yield ": keep-alive\n\n" finally: disconnected_event.set() producer_task.cancel() await asyncio.gather(producer_task, return_exceptions=True) @app.post("/chat/stream") async def chat_endpoint(request: Request, prompt: str) -> StreamingResponse: headers = { "Cache-Control": "no-cache", "Connection": "keep-alive", "X-Accel-Buffering": "no", } return StreamingResponse( event_generator(request, prompt), media_type="text/event-stream", headers=headers, ) ```

Architectural Specification & Invariant Rationale

Architectural Invariant Design: FastAPI ASGI Token-Streaming Backpressure and Heartbeat Standard

FastAPI ASGI Token-Streaming Backpressure and Heartbeat Standard defines an industrial-grade behavioral standard for backend engineering. Enforces proactive client disconnection detection, bounded ASGI queue backpressure, and periodic SSE heartbeats for token-streaming endpoints. Prevents runaway upstream LLM costs, kernel socket buffer exhaustion, and zombie generator memory leaks when downstream clients hang or disconnect. 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: Unbounded StreamingResponse generator lacking disconnect checks, heartbeats, and upstream task cancellation 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	Unbounded StreamingResponse generator lacking disconnect checks, heartbeats, and upstream task cancellation	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 pass the active `Request` instance into streaming generator functions and poll `await request.is_disconnected()` on every token iteration or yield interval.

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.