FastAPI Streaming LLM SSE: Server-Sent Events, Backpressure, and Nginx for Production 2026

TL;DR — FastAPI streaming LLM SSE: Server-Sent Events, backpressure, Nginx. EliteDev: "SSE is the standard for LLM token streaming. StreamingResponse with async generator. Backpressure: generator paused at await, no unbounded buffer. asyncio.Queue with max size for strict control." CallSphere: "StreamingResponse with async generator, SSE format, X-Accel-Buffering: no header critical for nginx. EventSource for GET, fetch for POST." DEV Community: "6 Nginx directives: proxy_buffering off, proxy_read_timeout 300s, proxy_http_version 1.1. Without them, tokens arrive all at once in production." Token Streaming Proxy: "BackpressureController high/low watermark, SSE heartbeats every 15s, client cancellation in finally block." Learn more with FastAPI AI tutorial, Ollama REST API, function calling, and Docker AI API.

EliteDev frames the transport: "There are three common ways to send data from server to client in real time: chunked transfer encoding, Server-Sent Events, and WebSockets. For LLM token streaming, SSE is the standard. It's simpler than WebSockets, works natively in browsers, and has built-in reconnection logic."

DEV Community adds the production warning: "In development without Nginx everything works perfectly. In production, tokens arrive all at once. The first time it happened, I spent 2 hours thinking the problem was in the backend. It was 6 lines of Nginx."

SSE Streaming Architecture

Transport Comparison

Implementation

from dataclasses import dataclass
from typing import Optional
from enum import Enum

class SSEFeature(Enum):
    STREAMING = "streaming"
    BACKPRESSURE = "backpressure"
    HEARTBEAT = "heartbeat"
    DISCONNECT = "disconnect"

@dataclass
class FastAPIStreamingSSEGuide:
    """FastAPI streaming LLM SSE implementation guide."""

    def get_sse_endpoint(self) -> str:
        """SSE streaming endpoint with disconnect handling."""
        return (
            "# === SSE STREAMING ENDPOINT ===\n"
            "# pip install fastapi uvicorn httpx\n"
            "#   sse-starlette\n"
            "\n"
            "from fastapi import FastAPI, Request\n"
            "from fastapi.responses import\n"
            "    StreamingResponse\n"
            "from sse_starlette.sse import\n"
            "    EventSourceResponse\n"
            "import httpx, json, asyncio\n"
            "\n"
            "app = FastAPI()\n"
            "OLLAMA_URL =\n"
            "    'http://localhost:11434'\n"
            "\n"
            "@app.post('/v1/chat/stream')\n"
            "async def chat_stream(\n"
            "    request: Request):\n"
            "    '''SSE streaming endpoint.''' \n"
            "    body = await request.json()\n"
            "    messages = body.get('messages', [])\n"
            "    \n"
            "    async def event_generator():\n"
            "        async with httpx.AsyncClient(\n"
            "            timeout=httpx.Timeout(300.0)\n"
            "        ) as client:\n"
            "            async with client.stream(\n"
            "                'POST',\n"
            "                f'{OLLAMA_URL}/api/chat',\n"
            "                json={\n"
            "                    'model': 'llama3.2',\n"
            "                    'messages': messages,\n"
            "                    'stream': True,\n"
            "                }\n"
            "            ) as response:\n"
            "                async for line in (\n"
            "                    response\n"
            "                    .aiter_lines()):\n"
            "                    \n"
            "                    # Check disconnect\n"
            "                    if await (\n"
            "                      request\n"
            "                      .is_disconnected()):\n"
            "                        break\n"
            "                    \n"
            "                    if not line:\n"
            "                        continue\n"
            "                    \n"
            "                    data = json.loads(line)\n"
            "                    content = (\n"
            "                        data.get('message',{})\n"
            "                        .get('content',''))\n"
            "                    \n"
            "                    if content:\n"
            "                        # SSE format\n"
            "                        yield ({\n"
            "                          'event': 'token',\n"
            "                          'data': json.dumps({\n"
            "                            'content': content\n"
            "                          })\n"
            "                        })\n"
            "                    \n"
            "                    if data.get('done'):\n"
            "                        yield ({\n"
            "                          'event': 'done',\n"
            "                          'data': json.dumps({\n"
            "                            'content': ''\n"
            "                          })\n"
            "                        })\n"
            "                        break\n"
            "    \n"
            "    return EventSourceResponse(\n"
            "        event_generator())"
        )

    def get_manual_sse(self) -> str:
        """Manual SSE format without sse-starlette."""
        return (
            "# === MANUAL SSE FORMAT ===\n"
            "# (without sse-starlette)\n"
            "\n"
            "@app.post('/v1/chat/stream')\n"
            "async def chat_stream(request: Request):\n"
            "    body = await request.json()\n"
            "    \n"
            "    async def generate():\n"
            "        async with httpx.AsyncClient(\n"
            "            timeout=300.0\n"
            "        ) as client:\n"
            "            async with client.stream(\n"
            "                'POST',\n"
            "                f'{OLLAMA_URL}/api/chat',\n"
            "                json={\n"
            "                    'model': 'llama3.2',\n"
            "                    'messages': body[\n"
            "                      'messages'],\n"
            "                    'stream': True,\n"
            "                }\n"
            "            ) as response:\n"
            "                async for line in (\n"
            "                    response\n"
            "                    .aiter_lines()):\n"
            "                    \n"
            "                    if await (\n"
            "                      request\n"
            "                      .is_disconnected()):\n"
            "                        break\n"
            "                    \n"
            "                    if not line:\n"
            "                        continue\n"
            "                    \n"
            "                    data = json.loads(line)\n"
            "                    content = (\n"
            "                        data.get('message',{})\n"
            "                        .get('content',''))\n"
            "                    \n"
            "                    if content:\n"
            "                        # SSE format\n"
            "                        sse_data = (\n"
            "                          json.dumps({\n"
            "                            'content': content\n"
            "                          }))\n"
            "                        yield (\n"
            "                          f'data: {sse_data}'\n"
            "                          '\\n\\n')\n"
            "                    \n"
            "                    if data.get('done'):\n"
            "                        yield 'data: [DONE]\\n\\n'\n"
            "                        break\n"
            "    \n"
            "    return StreamingResponse(\n"
            "        generate(),\n"
            "        media_type=\n"
            "          'text/event-stream',\n"
            "        headers={\n"
            "            'Cache-Control': 'no-cache',\n"
            "            'Connection': 'keep-alive',\n"
            "            'X-Accel-Buffering': 'no',\n"
            "        })"
        )

    def get_backpressure(self) -> str:
        """Backpressure with asyncio.Queue."""
        return (
            "# === BACKPRESSURE CONTROL ===\n"
            "\n"
            "async def stream_with_backpressure(\n"
            "    request: Request,\n"
            "    messages: list):\n"
            "    '''Stream with bounded buffer.''' \n"
            "    \n"
            "    # Bounded queue for backpressure\n"
            "    queue = asyncio.Queue(\n"
            "        maxsize=100)\n"
            "    \n"
            "    # Producer: read from LLM\n"
            "    async def producer():\n"
            "        try:\n"
            "            async with httpx.AsyncClient(\n"
            "                timeout=300.0\n"
            "            ) as client:\n"
            "                async with client.stream(\n"
            "                    'POST',\n"
            "                    f'{OLLAMA_URL}/api/chat',\n"
            "                    json={\n"
            "                        'model': 'llama3.2',\n"
            "                        'messages': messages,\n"
            "                        'stream': True,\n"
            "                    }\n"
            "                ) as response:\n"
            "                    async for line in (\n"
            "                        response\n"
            "                        .aiter_lines()):\n"
            "                        if line:\n"
            "                            data = (\n"
            "                                json.loads(line))\n"
            "                            content = (\n"
            "                                data.get(\n"
            "                                'message',{})\n"
            "                                .get(\n"
            "                                  'content',''))\n"
            "                            if content:\n"
            "                                await queue.put({\n"
            "                                  'content':\n"
            "                                    content})\n"
            "                            if data.get('done'):\n"
            "                                await queue.put(\n"
            "                                  {'done': True})\n"
            "                                break\n"
            "        except Exception as e:\n"
            "            await queue.put(\n"
            "              {'error': str(e)})\n"
            "        finally:\n"
            "            await queue.put(None)\n"
            "    \n"
            "    # Consumer: yield to client\n"
            "    async def consumer():\n"
            "        task = asyncio.create_task(\n"
            "            producer())\n"
            "        try:\n"
            "            while True:\n"
            "                if await (\n"
            "                  request\n"
            "                  .is_disconnected()):\n"
            "                    task.cancel()\n"
            "                    break\n"
            "                \n"
            "                item = await queue.get()\n"
            "                if item is None:\n"
            "                    break\n"
            "                if item.get('done'):\n"
            "                    yield 'data: [DONE]\\n\\n'\n"
            "                    break\n"
            "                if item.get('error'):\n"
            "                    yield (\n"
            "                      f'data: {json.dumps'\n"
            "                      '(item)}\\n\\n')\n"
            "                    break\n"
            "                \n"
            "                sse = json.dumps(\n"
            "                    {'content':\n"
            "                      item['content']})\n"
            "                yield f'data: {sse}\\n\\n'\n"
            "        finally:\n"
            "            task.cancel()\n"
            "    \n"
            "    return consumer()"
        )

    def get_heartbeats(self) -> str:
        """SSE heartbeats for idle connections."""
        return (
            "# === SSE HEARTBEATS ===\n"
            "\n"
            "async def stream_with_heartbeats(\n"
            "    request: Request,\n"
            "    messages: list):\n"
            "    '''Stream with keepalive\n"
            "    heartbeats during idle.''' \n"
            "    \n"
            "    last_token = asyncio.get_event_loop(\n"
            "        ).time()\n"
            "    \n"
            "    async with httpx.AsyncClient(\n"
            "        timeout=300.0\n"
            "    ) as client:\n"
            "        async with client.stream(\n"
            "            'POST',\n"
            "            f'{OLLAMA_URL}/api/chat',\n"
            "            json={\n"
            "                'model': 'llama3.2',\n"
            "                'messages': messages,\n"
            "                'stream': True,\n"
            "            }\n"
            "        ) as response:\n"
            "            async for line in (\n"
            "                response\n"
            "                .aiter_lines()):\n"
            "                \n"
            "                if await (\n"
            "                  request\n"
            "                  .is_disconnected()):\n"
            "                    break\n"
            "                \n"
            "                now = asyncio.get_event_loop(\n"
            "                    ).time()\n"
            "                \n"
            "                # Heartbeat if idle >15s\n"
            "                if (now - last_token\n"
            "                    > 15):\n"
            "                    yield ':keepalive\\n\\n'\n"
            "                \n"
            "                if not line:\n"
            "                    continue\n"
            "                \n"
            "                data = json.loads(line)\n"
            "                content = (\n"
            "                    data.get('message',{})\n"
            "                    .get('content',''))\n"
            "                \n"
            "                if content:\n"
            "                    last_token = now\n"
            "                    sse = json.dumps({\n"
            "                      'content': content})\n"
            "                    yield f'data: {sse}\\n\\n'\n"
            "                \n"
            "                if data.get('done'):\n"
            "                    yield 'data: [DONE]\\n\\n'\n"
            "                    break"
        )

    def get_nginx_config(self) -> str:
        """Nginx configuration for SSE."""
        return (
            "# === NGINX SSE CONFIGURATION ===\n"
            "# 6 critical SSE directives\n"
            "\n"
            "location /v1/chat/stream {\n"
            "    proxy_pass http://api:8000;\n"
            "    \n"
            "    # 1. Don't buffer response\n"
            "    proxy_buffering off;\n"
            "    \n"
            "    # 2. Don't cache\n"
            "    proxy_cache off;\n"
            "    \n"
            "    # 3. Long timeout\n"
            "    #    (streams last minutes)\n"
            "    proxy_read_timeout 300s;\n"
            "    \n"
            "    # 4. Disable upstream\n"
            "    #    keep-alive\n"
            "    proxy_set_header\n"
            "      Connection '';\n"
            "    \n"
            "    # 5. HTTP/1.1 required\n"
            "    #    for chunked\n"
            "    proxy_http_version 1.1;\n"
            "    \n"
            "    # 6. Disable Nginx\n"
            "    #    chunked encoding\n"
            "    chunked_transfer_encoding off;\n"
            "}\n"
            "\n"
            "# Without these 6 directives:\n"
            "# - proxy_buffering: tokens\n"
            "#   arrive all at once\n"
            "# - proxy_read_timeout:\n"
            "#   connection closes\n"
            "#   mid-response (60s\n"
            "#   default)\n"
            "# - HTTP/1.0: chunked\n"
            "#   encoding fails"
        )

    def get_client_js(self) -> str:
        """JavaScript client for SSE."""
        return (
            "// === JAVASCRIPT CLIENT ===\n"
            "\n"
            "// Option 1: EventSource (GET only)\n"
            "const source = new EventSource(\n"
            "  '/v1/chat/stream?message=hi');\n"
            "\n"
            "source.addEventListener('token',\n"
            "  (e) => {\n"
            "    const data = JSON.parse(e.data);\n"
            "    console.log(data.content);\n"
            "    // Append to UI\n"
            "  });\n"
            "\n"
            "source.addEventListener('done',\n"
            "  (e) => {\n"
            "    source.close();\n"
            "  });\n"
            "\n"
            "// Option 2: fetch (POST)\n"
            "const response = await fetch(\n"
            "  '/v1/chat/stream', {\n"
            "    method: 'POST',\n"
            "    headers: {\n"
            "      'Content-Type':\n"
            "        'application/json'},\n"
            "    body: JSON.stringify({\n"
            "      messages: [\n"
            "        {role: 'user',\n"
            "         content: 'Hello'}\n"
            "      ]\n"
            "    })\n"
            "  });\n"
            "\n"
            "const reader = response.body\n"
            "  .getReader();\n"
            "const decoder = new TextDecoder();\n"
            "\n"
            "while (true) {\n"
            "  const {done, value} = await\n"
            "    reader.read();\n"
            "  if (done) break;\n"
            "  \n"
            "  const text = decoder\n"
            "    .decode(value);\n"
            "  // Parse SSE lines\n"
            "  const lines = text\n"
            "    .split('\\n');\n"
            "  for (const line of lines) {\n"
            "    if (line.startsWith(\n"
            "      'data: ')) {\n"
            "      const data = JSON.parse(\n"
            "        line.slice(6));\n"
            "      if (data.content)\n"
            "        console.log(\n"
            "          data.content);\n"
            "    }\n"
            "  }\n"
            "}"
        )

    def get_production_tips(self) -> dict:
        """Production tips."""
        return {
            "sse_vs_websocket": "SSE for LLM token streaming (simpler, native browser, reconnection). WebSocket for bidirectional chat with typing indicators.",
            "async_client": "Always use AsyncOpenAI/AsyncAnthropic in FastAPI async endpoints. Sync client blocks event loop and kills concurrency.",
            "x_accel_buffering": "X-Accel-Buffering: no header in StreamingResponse — without it Nginx buffers entire response, breaking streaming.",
            "disconnect_check": "request.is_disconnected() periodically in generator — stop LLM generation when client gone to save costs.",
            "timeout": "Set proxy_read_timeout 300s (default 60s too short for LLM responses with agentic search).",
            "heartbeats": "SSE comments (:keepalive) every 15s during idle — prevents proxy/LB from closing idle connections.",
            "backpressure": "asyncio.Queue(maxsize=100) for bounded buffer. Async generator natural backpressure via await pause.",
            "error_handling": "yield error events as SSE data — client can display error message gracefully.",
            "reconnection": "SSE id: field for reconnection. Last-Event-ID header on reconnect — skip already-delivered tokens.",
            "testing": "Always test through Nginx. What works on localhost without proxy is not representative of production.",
        }

FastAPI Streaming SSE Checklist

• [ ] SSE is the standard for LLM token streaming — simpler than WebSockets, native browser support, built-in reconnection
• [ ] Three transport options: SSE (standard), WebSockets (bidirectional), chunked transfer (simple)
• [ ] SSE: server-to-client only, EventSource API in browsers, built-in reconnection with Last-Event-ID
• [ ] WebSockets: bidirectional, manual reconnection, better for chat with typing indicators and presence
• [ ] FastAPI StreamingResponse with async generator for SSE streaming
• [ ] sse-starlette: EventSourceResponse for proper SSE format with event types
• [ ] Manual SSE: data: {json}\n\n format, event: token\ndata: {json}\n\n for typed events
• [ ] data: [DONE]\n\n terminator signal
• [ ] media_type='text/event-stream' for SSE
• [ ] X-Accel-Buffering: no header critical for Nginx — without it, entire response is buffered
• [ ] Cache-Control: no-cache, Connection: keep-alive headers
• [ ] Ollama streaming: stream=true, NDJSON over HTTP, async for line in response.aiter_lines()
• [ ] OpenAI streaming: AsyncOpenAI, stream.text_stream, sync generator wrapped async
• [ ] Always use async client (AsyncOpenAI, httpx.AsyncClient) in FastAPI async endpoints
• [ ] Sync client blocks event loop and kills server concurrency
• [ ] Backpressure: async generator paused at await — natural backpressure, no unbounded buffer
• [ ] Backpressure: asyncio.Queue(maxsize=100) for bounded buffer between producer and consumer
• [ ] Backpressure: high/low watermark with hysteresis to prevent oscillation
• [ ] Backpressure: TCP flow control propagates end-to-end — client slow → yield pauses → reader pauses → LLM slows
• [ ] Client disconnect: request.is_disconnected() check periodically in generator loop
• [ ] Client disconnect: break generator loop when client gone
• [ ] Client disconnect: finally block to close upstream connection and cancel reader task
• [ ] Client disconnect: stop LLM generation to save GPU compute and API costs
• [ ] Without disconnect handling: server keeps calling LLM API after user closed browser tab
• [ ] Heartbeats: SSE comments (:keepalive\n\n) every 15 seconds during idle periods
• [ ] Heartbeats: clients ignore comment lines per SSE spec — invisible to application
• [ ] Heartbeats: prevents intermediate proxies, load balancers, CDNs from closing idle connections
• [ ] Heartbeats: critical during prefill phase (long prompts can take seconds before first token)
• [ ] Nginx directive 1: proxy_buffering off — don't buffer response
• [ ] Nginx directive 2: proxy_cache off — don't cache
• [ ] Nginx directive 3: proxy_read_timeout 300s — default 60s too short, LLM can take 2-3 minutes
• [ ] Nginx directive 4: proxy_set_header Connection '' — disable upstream keep-alive
• [ ] Nginx directive 5: proxy_http_version 1.1 — required for chunked transfer
• [ ] Nginx directive 6: chunked_transfer_encoding off — disable Nginx's chunked encoding
• [ ] Without Nginx directives: tokens arrive all at once in production (works fine in dev)
• [ ] Always test through Nginx — localhost without proxy is not representative of production
• [ ] EventSource API: native browser SSE, GET requests only, built-in reconnection
• [ ] fetch + ReadableStream: POST requests, manual stream reading, total control
• [ ] SSE reconnection: id: field for event ID, Last-Event-ID header on reconnect, skip delivered tokens
• [ ] Error handling: yield error events as SSE data — client displays error gracefully
• [ ] Timeout: set streaming timeout in reverse proxy (10 minutes for long outputs)
• [ ] Read FastAPI AI tutorial for base setup
• [ ] Read Ollama REST API for Ollama streaming
• [ ] Read function calling for structured output
• [ ] Read Docker AI API for deployment
• [ ] Test: streaming yields tokens incrementally (not all at once)
• [ ] Test: client disconnect stops LLM generation
• [ ] Test: backpressure prevents memory spikes under slow client
• [ ] Test: heartbeats keep connection alive during prefill
• [ ] Test: Nginx configuration delivers token-by-token (not buffered)
• [ ] Test: error events are properly formatted as SSE
• [ ] Test: reconnection with Last-Event-ID skips delivered tokens
• [ ] Document SSE format, backpressure strategy, Nginx config, heartbeat interval, disconnect handling

FAQ

How do you stream LLM tokens with FastAPI SSE?

Use StreamingResponse with async generator yielding SSE-formatted chunks. EliteDev: "For LLM token streaming, SSE is the standard. Simpler than WebSockets, works natively in browsers, built-in reconnection. StreamingResponse with async generator. sse-starlette for SSE format. yield data: content for each token, data: [DONE] at end. Under the hood, StreamingResponse uses chunked transfer encoding. Browser sees text/event-stream and treats as SSE." CallSphere: "StreamingResponse with async generator to produce SSE events. event: token, data: JSON content. X-Accel-Buffering: no header critical for nginx. EventSource API handles SSE natively but only supports GET. For POST, use fetch with stream reader." Streaming: (1) StreamingResponse with async generator. (2) SSE format: data: {json}. (3) event: token for typed events. (4) data: [DONE] terminator. (5) media_type='text/event-stream'. (6) X-Accel-Buffering: no header.

How do you handle backpressure in FastAPI SSE streaming?

Use async generator semantics and asyncio.Queue with max size for bounded buffer. EliteDev: "Backpressure: no buffer ever grows unbounded because generator is paused at await until data ready. For strict control, use asyncio.Queue with max size between provider and HTTP response. FastAPI built-in streaming handles moderate backpressure fine." Token Streaming Proxy: "BackpressureController with high/low watermark. When client slow, yield pauses, buffer fills to high watermark, reader pauses, httpx stops reading, TCP flow control propagates to LLM API. End-to-end backpressure using TCP flow control and Python async/await. Hysteresis prevents oscillation." Backpressure: (1) Async generator pauses at await — natural backpressure. (2) asyncio.Queue(maxsize=100) for bounded buffer. (3) High/low watermark for hysteresis. (4) TCP flow control propagates end-to-end. (5) No explicit flow control needed — standard async/await.

How do you handle client disconnects in FastAPI streaming?

Check request.is_disconnected() inside the generator and stop producing tokens. AI Learning Hub: "If client disconnects mid-stream, server should stop generating tokens to avoid wasting LLM API costs. FastAPI handles this with request.is_disconnected() — check periodically inside generator. Not handling client disconnects: server keeps calling LLM API and accumulating costs even after user closed browser tab." Token Streaming Proxy: "When client disconnects, ASGI lifecycle cancels response task. finally block closes backpressure controller and cancels upstream reader, closing upstream connection. Without this, upstream continues generating tokens wasting GPU compute." Disconnect handling: (1) if await request.is_disconnected(): break. (2) Check periodically in generator loop. (3) finally block to close upstream connection. (4) ASGI lifecycle cancels task on disconnect. (5) Stop LLM generation to save costs. (6) Clean up resources in finally.

What Nginx configuration is needed for SSE streaming?

Disable proxy buffering, set long timeout, HTTP/1.1, and disable chunked encoding. DEV Community: "6 SSE directives: proxy_buffering off, proxy_cache off, proxy_read_timeout 300s, proxy_set_header Connection empty, proxy_http_version 1.1, chunked_transfer_encoding off. Default proxy_read_timeout 60s — LLM response with agentic search can take 2-3 minutes. Without timeout, Nginx closes connection mid-response. In dev without Nginx everything works. In production, tokens arrive all at once. Those 6 Nginx directives are the difference between real token-by-token streaming and all text at once after 10 seconds." CallSphere: "X-Accel-Buffering: no header critical — without it proxy buffers entire response. Disable response buffering in reverse proxy. Set idle timeout higher than max response time." Nginx: (1) proxy_buffering off. (2) proxy_cache off. (3) proxy_read_timeout 300s. (4) proxy_set_header Connection ''. (5) proxy_http_version 1.1. (6) chunked_transfer_encoding off. (7) X-Accel-Buffering: no header in FastAPI.

How do you implement SSE heartbeats for idle connections?

Send SSE comments every 15 seconds during idle periods to keep connection alive. Token Streaming Proxy: "SSE connections can go idle between tokens, especially during prefill phase which can take seconds for long prompts. Intermediate proxies, load balancers, and CDNs may close idle connections after 30-60 seconds. Proxy sends SSE comments (:keepalive) every 15 seconds during idle periods. Per SSE spec, clients ignore comment lines, so heartbeats are invisible to application." EliteDev: "Set reasonable streaming timeout in reverse proxy (10 minutes for long outputs) and handle timeouts gracefully with StreamChunk of type ERROR." Heartbeats: (1) SSE comment: :keepalive every 15s. (2) Clients ignore comment lines per SSE spec. (3) Prevents proxy/LB from closing idle connections. (4) Critical during prefill phase (long prompts). (5) Set proxy_read_timeout higher than max response time. (6) Handle timeouts gracefully with error events.

Want a self-hosted AI company brain that does all of this out of the box?
Book a demo →