Streaming LLM responses from FastAPI with StreamingResponse

Wiring a fastapi streamingresponse llm endpoint correctly means your users see tokens arrive as they’re generated instead of staring at a spinner. Most copied snippets either buffer the whole response or crash when the client hangs up. This guide builds a production-shaped streaming route from scratch, with runnable code and a concrete way to verify it works.

Step 1: Scaffold the app and install dependencies

Create a virtual environment and install the minimal stack. You need FastAPI, an ASGI server, and an HTTP client that supports async streaming. The OpenAI Python SDK is the fastest path to an OpenAI-compatible stream, but plain httpx works if you want zero vendor lock.

python -m venv .venv
source .venv/bin/activate
pip install fastapi uvicorn openai

Create main.py with a bare FastAPI instance. Keep configuration in environment variables, not hardcoded strings.

import os
from fastapi import FastAPI

app = FastAPI(title="llm-stream-proxy")

OPENAI_API_KEY = os.environ["OPENAI_API_KEY"]
BASE_URL = os.environ.get("LLM_BASE_URL", "https://api.openai.com/v1")

Step 2: Write the async token generator

The core of any fastapi streamingresponse llm route is an async generator that yields string chunks. The OpenAI SDK exposes stream=True, which returns an async iterator of delta objects. Extract .delta.content and yield only non-empty strings.

from openai import AsyncOpenAI

client = AsyncOpenAI(api_key=OPENAI_API_KEY, base_url=BASE_URL)

async def llm_token_stream(prompt: str, model: str = "gpt-4o-mini"):
    completion = await client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": prompt}],
        stream=True,
        temperature=0.7,
    )
    async for chunk in completion:
        delta = chunk.choices[0].delta.content
        if delta:
            yield delta

If you call a raw endpoint with httpx, you would async for line in response.aiter_lines() and parse SSE. The SDK hides that, but the generator contract is identical: yield text fragments.

Step 3: Expose the generator via StreamingResponse

FastAPI’s StreamingResponse accepts a generator and a media type. For plain text tokens, text/plain is honest. If you need browser EventSource compatibility, use text/event-stream and format each yield as data: {chunk}\n\n.

from fastapi import Request
from fastapi.responses import StreamingResponse

@app.post("/v1/chat")
async def chat(request: Request):
    body = await request.json()
    prompt = body.get("prompt", "")
    if not prompt:
        return {"error": "prompt required"}, 400
    return StreamingResponse(
        llm_token_stream(prompt),
        media_type="text/plain",
    )

This is the minimal fastapi streamingresponse llm wiring. It will stream, but it lacks validation and cleanup.

Choosing the media type

If you serve a browser via fetch + ReadableStream, text/plain is fine. For Server-Sent Events, set media_type="text/event-stream" and yield formatted frames:

async def sse_format(stream):
    async for token in stream:
        yield f"data: {token}\n\n"

# inside route:
return StreamingResponse(
    sse_format(llm_token_stream(prompt)),
    media_type="text/event-stream",
)

The client then reads event.data on each message event. Pick one and stay consistent; mixing them breaks clients.

Step 4: Validate input with Pydantic

Accepting raw request.json() invites malformed payloads. Define a body model so FastAPI handles parsing and docs.

from pydantic import BaseModel

class ChatRequest(BaseModel):
    prompt: str
    model: str = "gpt-4o-mini"

@app.post("/v1/chat")
async def chat(req: ChatRequest):
    return StreamingResponse(
        llm_token_stream(req.prompt, req.model),
        media_type="text/plain",
    )

Step 5: Handle client disconnects

When a user closes the tab, the generator keeps running until the LLM finishes unless you cancel it. Wrap the stream consumption in a try/except for asyncio.CancelledError and ensure the SDK stream is closed. The OpenAI SDK’s async context manager does this if you use async with, but with await create() you must close manually.

import asyncio

async def llm_token_stream(prompt: str, model: str = "gpt-4o-mini"):
    try:
        completion = await client.chat.completions.create(
            model=model,
            messages=[{"role": "user", "content": prompt}],
            stream=True,
        )
        async for chunk in completion:
            delta = chunk.choices[0].delta.content
            if delta:
                yield delta
    except asyncio.CancelledError:
        # Client disconnected; tear down upstream stream
        await completion.close()
        raise

FastAPI cancels the task on disconnect, so the CancelledError will fire. Without the close call, you leak a connection to the provider.

Step 6: Run and verify with curl

Start the server with Uvicorn. Use --reload only in dev.

uvicorn main:app --host 0.0.0.0 --port 8000

Streaming requires the client to disable buffering. curl -N does that.

curl -N -X POST http://localhost:8000/v1/chat \
  -H 'Content-Type: application/json' \
  -d '{"prompt":"Explain recursion in one sentence"}'

You should see words appear incrementally in the terminal, not all at once after a pause. If you get a single block, check for a proxy (like Nginx) that buffers upstream responses, or confirm you passed -N.

Step 7: Point at an OpenAI-compatible gateway

The same generator works against any OpenAI-compatible endpoint. If you set BASE_URL to a gateway that aggregates models, the fastapi streamingresponse llm code does not change. For example, an inference gateway such as n4n.ai exposes one endpoint for 240+ models and automatically falls back when a provider is rate-limited, while forwarding cache-control hints—so the streaming route stays unchanged and you avoid writing fallback branching yourself.

BASE_URL = "https://api.n4n.ai/v1"  # OpenAI-compatible, 240+ models

Set LLM_BASE_URL in the environment and restart. The curl test from Step 6 should behave identically.

Step 8: Write an automated streaming test

Manual curl is good for a smoke test, but you want a regression guard. Use httpx.AsyncClient with stream=True and count yielded chunks.

import httpx
import pytest

@pytest.mark.asyncio
async def test_stream_chunks():
    async with httpx.AsyncClient(app=app, base_url="http://test") as ac:
        async with ac.stream("POST", "/v1/chat", json={"prompt":"hi"}) as resp:
            assert resp.status_code == 200
            chunks = []
            async for piece in resp.aiter_text():
                chunks.append(piece)
            assert len(chunks) > 1  # confirms token-level streaming, not one blob

Run with pytest after installing pytest and pytest-asyncio. If len(chunks) is 1, your server or test client buffered.

Step 9: Production hardening

A few things separate a demo from a service:

• Worker model: Run Uvicorn with multiple workers behind a load balancer that supports long-lived connections (e.g., uvicorn main:app --workers 4). Streaming holds connections open, so size your file descriptors.
• Timeouts: Set a per-request upstream timeout. The OpenAI SDK accepts timeout on the client. FastAPI’s StreamingResponse does not enforce one for the generator; wrap the LLM call with asyncio.wait_for.
• Backpressure: If your downstream client is slow, the generator’s yield blocks until the ASGI server sends. That’s correct, but it means a slow browser can pin a worker. Monitor concurrency.
• Token metering: If you use a gateway, per-token usage metering usually arrives in a final usage chunk or response header. Capture it by inspecting the last chunk in the generator and logging.

async def llm_token_stream(prompt: str, model: str = "gpt-4o-mini"):
    completion = await client.chat.completions.create(
        model=model, messages=[{"role":"user","content":prompt}], stream=True,
    )
    async for chunk in completion:
        if chunk.usage:
            log_usage(chunk.usage)  # gateway or provider final accounting
        delta = chunk.choices[0].delta.content
        if delta:
            yield delta

Proxy buffering

If deploying behind Nginx, add:

proxy_buffering off;
proxy_cache off;

Otherwise Nginx will wait for the response to reach 4–8 KB before forwarding, defeating the stream. For Gunicorn-based deployments, use the Uvicorn worker: gunicorn -k uvicorn.workers.UvicornWorker main:app.

Verify success

Success means three things: (1) curl -N shows incremental output; (2) the pytest stream test asserts more than one chunk; (3) killing the client mid-stream does not raise unhandled exceptions in the server log—the CancelledError path closes the upstream. Once those hold, your fastapi streamingresponse llm route is safe to put behind a real frontend.