Streaming LLM Responses Through FastAPI

You already know how to call a model and wait for the whole answer. This page resolves a narrower and more expensive question: when a paying customer hits your endpoint and you are relaying a model's output token by token, what wire format do you use, and what happens to the meter when they close the tab halfway through? Part of the Automating AI Workflows with Python APIs guide, which covers the background-worker shape; this page covers the one case where holding an HTTP connection open for ninety seconds is the correct design.

The decision is short, so here it is up front: ship server-sent events. Chunked plain text is easier to write and it will cost you a support ticket per week the moment a customer puts a proxy, a CDN, or a browser between themselves and your API. SSE gives you framed events, a documented reconnect story, and — the part that actually pays for itself — somewhere to put usage numbers that is not the response body your customer is parsing as prose.

SSE or a raw chunked body

Both options are the same ASGI mechanism underneath: FastAPI's StreamingResponse wraps an async generator, and the server flushes each yielded chunk with Transfer-Encoding: chunked. The difference is entirely in what the bytes mean.

A raw chunked body streams naked text. The client concatenates whatever arrives. There is no way to signal "this chunk is a usage record, not prose", no way to mark the end of a logical message without inventing a sentinel, and no standard for what a client should do after a dropped connection. Server-sent events add a two-line framing (event: and data:, terminated by a blank line) that costs you roughly twelve bytes per chunk and buys you typed events, JSON payloads, and a client contract every HTTP library on earth already understands.

There is one deployment gotcha worth internalising before you write a line of code: intermediaries buffer. Nginx buffers proxied responses by default, several PaaS edge layers buffer anything they think is a document, and a customer's corporate proxy will happily hold your stream until it has 8 KB or the connection closes. You defeat this with X-Accel-Buffering: no and Cache-Control: no-cache, which most edges honour for text/event-stream and ignore for text/plain. That alone settles the format argument.

The proxy endpoint

Here is the whole thing. It reads config from the environment, streams from the model with the async SDK, re-frames each delta as an SSE event, and emits a final usage event before closing.

import asyncio
import json
import os
from collections.abc import AsyncIterator

from anthropic import AsyncAnthropic
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
from pydantic import BaseModel

app = FastAPI()
client = AsyncAnthropic(timeout=float(os.getenv("LLM_REQUEST_TIMEOUT_S", "90")))

MODEL = os.getenv("LLM_MODEL", "claude-opus-4-8")
MAX_TOKENS = int(os.getenv("LLM_MAX_TOKENS", "2048"))
FIRST_TOKEN_DEADLINE_S = float(os.getenv("LLM_FIRST_TOKEN_DEADLINE_S", "20"))


class Prompt(BaseModel):
    text: str


def sse(event: str, payload: dict) -> bytes:
    return f"event: {event}\ndata: {json.dumps(payload)}\n\n".encode()


async def relay(prompt: str, request: Request) -> AsyncIterator[bytes]:
    async with client.messages.stream(
        model=MODEL,
        max_tokens=MAX_TOKENS,
        messages=[{"role": "user", "content": prompt}],
    ) as stream:
        async for text in stream.text_stream:
            if await request.is_disconnected():
                break
            yield sse("token", {"t": text})

        final = await stream.get_final_message()
        yield sse(
            "usage",
            {
                "in": final.usage.input_tokens,
                "out": final.usage.output_tokens,
                "stop": final.stop_reason,
            },
        )
    yield sse("done", {})


@app.post("/v1/complete")
async def complete(body: Prompt, request: Request) -> StreamingResponse:
    return StreamingResponse(
        relay(body.text, request),
        media_type="text/event-stream",
        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
    )

Three details carry the design. client.messages.stream is an async context manager, so the upstream HTTP connection closes deterministically when the generator exits — including on an exception. stream.text_stream hands you only the text deltas, which is what you want to relay; the SDK is still tracking the underlying message_start and message_delta events for you. And get_final_message() returns the assembled message with populated usage, which is the number you bill on.

Backpressure and disconnects

Backpressure in an ASGI stream is mostly free and mostly invisible. When you yield a chunk, Uvicorn awaits the socket write. If the client is reading slowly, that await simply takes longer, your generator is suspended, and no memory piles up. The failure mode is not a buffer explosion — it is a coroutine parked for minutes on a customer whose laptop went to sleep, holding an upstream model connection you are being billed for.

That is why the disconnect check matters, and why the naive version is wrong. await request.is_disconnected() only reports what the ASGI server already knows, and the server frequently does not know until it tries to write. In practice the check fires one or two chunks late, which is fine. What is not fine is omitting it entirely: without it, your generator keeps pulling deltas from the model until the model finishes, and you pay full output tokens for a response nobody read.

There is a second, sharper failure mode. If the ASGI server cancels your task outright — which happens on some deploy targets and on shutdown — the async for raises a cancellation exception inside your generator. Because the SDK stream is an async with block, the upstream connection still closes correctly. Do not wrap the whole thing in a bare except Exception that swallows cancellation; catch asyncio.CancelledError explicitly if you need to log, then re-raise it.

Token accounting mid-stream

This is where streaming quietly breaks billing. In a non-streaming call you get one response object with usage attached, you write a row, and you are done. In a stream the accounting arrives in pieces, and the piece you most need shows up last — or, on a disconnect, never.

The underlying event sequence is fixed: message_start carries input tokens (you know your cost floor before a single output token exists), a run of content_block_delta events carry the text, and message_delta carries the cumulative output token count. The SDK folds these into get_final_message(), but the shape matters because it tells you exactly what you can salvage from an aborted stream.

The commercially correct pattern is a finally block that always writes a usage row, flagged with whether the numbers are exact or estimated. Count characters as you relay, divide by roughly four for an English-text estimate, and reconcile against the exact figure whenever the stream completes.

import os
import time

CHARS_PER_TOKEN = float(os.getenv("LLM_CHARS_PER_TOKEN", "4.0"))


async def metered_relay(prompt: str, request: Request, api_key_id: str):
    started = time.monotonic()
    chars = 0
    usage = None
    try:
        async with client.messages.stream(
            model=MODEL,
            max_tokens=MAX_TOKENS,
            messages=[{"role": "user", "content": prompt}],
        ) as stream:
            async for text in stream.text_stream:
                if await request.is_disconnected():
                    break
                chars += len(text)
                yield sse("token", {"t": text})
            else:
                usage = (await stream.get_final_message()).usage
    finally:
        match usage:
            case None:
                out, exact = int(chars / CHARS_PER_TOKEN), False
            case _:
                out, exact = usage.output_tokens, True
        await record_usage(
            api_key_id=api_key_id,
            model=MODEL,
            output_tokens=out,
            exact=exact,
            seconds=time.monotonic() - started,
        )

The for ... else is load-bearing: get_final_message() runs only when the loop was never broken, so an abandoned stream skips it rather than blocking on a response the customer no longer wants. Write the row from finally regardless, and feed those rows into the same pipeline you use for logging API usage events to Postgres — a stripe.billing.MeterEvent posted per stream is the cleanest bridge to metered billing configuration.

Timeouts stack, and you need all three

A single timeout number does not describe a stream. You need three, and they measure different things.

The first is time to first token, the one customers judge you on. Twenty seconds is a generous ceiling; past that, fail loudly rather than hold a socket. The second is inter-token idle time — a stream that emits nothing for thirty seconds is dead even though its total elapsed time looks healthy. The third is total wall clock, which protects you from a pathological generation that trickles tokens for ten minutes. Neither httpx nor the SDK gives you a wall-clock cap by default: their timeouts are per-read and reset on every byte.

import asyncio
import os

IDLE_S = float(os.getenv("LLM_IDLE_TIMEOUT_S", "30"))
TOTAL_S = float(os.getenv("LLM_TOTAL_TIMEOUT_S", "180"))


async def guarded(source, request: Request):
    async with asyncio.timeout(TOTAL_S):
        iterator = source.__aiter__()
        while True:
            try:
                chunk = await asyncio.wait_for(iterator.__anext__(), IDLE_S)
            except StopAsyncIteration:
                return
            except TimeoutError:
                yield sse("error", {"code": "upstream_stalled"})
                return
            if await request.is_disconnected():
                return
            yield chunk

Set the deadline on your reverse proxy higher than TOTAL_S, never lower — a proxy that kills the connection first strips your ability to emit a clean error event, and the customer sees a truncated response instead of a diagnosable failure. If you run Gunicorn in front of Uvicorn workers, raise its --timeout too; the defaults there are tuned for request-response endpoints, not for the streaming shape, which is one more reason to read the Uvicorn and Gunicorn worker configuration trade-offs before you deploy this.

When to stream, and when not to

Stream when the output is prose a human reads as it appears, when generations routinely exceed five seconds, or when your competitors stream and yours feels broken by comparison. Perceived latency is the entire benefit, and it is worth real money on a chat-shaped product.

Do not stream when the consumer is another program. If your customer parses your output as JSON, streaming buys them nothing and costs you a held connection per request, which is the worst possible concurrency profile on a small box. Do not stream when the work belongs in a queue either — anything over two minutes should be a job with a webhook callback, using the pattern from running background jobs with Celery. And be careful on serverless platforms billed by wall-clock duration: a ninety-second stream at low concurrency can cost more in compute than the model tokens did, a trap worth modelling with cost per API request before you commit.

Migrating from a blocking endpoint

If you already have a JSON endpoint that returns the whole completion, the switch is four steps and does not require breaking existing customers.

1. Add a new path — /v1/complete/stream — rather than changing the old one's content type. Existing integrations keep working.
2. Move the model call into an async generator, swap messages.create for messages.stream, and yield SSE frames.
3. Move usage recording into a finally block with the exact-versus-estimated flag, so both endpoints feed the same meter.
4. Once the streaming path carries the majority of traffic, mark the blocking endpoint deprecated with a sunset header rather than deleting it.

Builder verdict

Ship SSE, put the usage numbers in their own event, and write the meter row from a finally block. That combination costs you about forty lines over the blocking version and removes the two failure modes that actually cost money: silent buffering that makes your API look slow, and abandoned streams you either bill wrongly or bill not at all. The twelve bytes of framing per chunk is a rounding error against a single model call.

The one thing worth being strict about is the disconnect check. Every stream you keep pulling after the customer left is pure margin burned — at typical output pricing, one abandoned long generation per hundred requests is a measurable dent in gross margin on a cheap tier. Check on every chunk, break the loop, and let the context manager close the upstream connection. Get that right on day one and streaming becomes a feature you charge for instead of a leak you discover on the invoice.

FAQ

Does streaming cost more than a blocking call? The model tokens are identical. What changes is your compute bill: you hold a worker slot for the whole generation instead of a fraction of a second, so an async worker that handled 200 requests per second might handle 40 concurrent streams. On a fixed-price VPS that is free capacity you were not using; on per-second serverless it can double your cost per request. Model it before you price the tier.

How do I bill a customer who disconnects halfway? Bill the estimate and mark it as such. You are charged by the model for every token it generated before you closed the connection, so absorbing that is real margin loss. Record the row with an exact=false flag, and if disputes appear, credit them — the volume is small and the goodwill is cheaper than the argument.

Can I rotate an API key mid-stream? No, and you should not try. The upstream connection was authenticated when it opened. Rotate keys between requests using an overlap window where both the old and new key validate, then let in-flight streams drain naturally. Anything else drops paying customers mid-sentence.

What breaks when I put this behind a CDN? Buffering, almost always. A CDN that treats text/event-stream as a document will hold your chunks until the response completes, turning a streaming API into a slow blocking one with no error to debug. Send Cache-Control: no-cache and X-Accel-Buffering: no, then verify with curl -N against the public URL — not against localhost, where the problem never reproduces.

Is this migration risky for existing integrations? Not if you add a new path instead of changing the existing one's content type. Customers parsing your JSON endpoint keep parsing it; new customers get the stream. Deprecate the old path only after the traffic has moved, using a sunset header and a dated notice.

Related

Same track:

• Automating AI Workflows with Python APIs — the parent guide, covering the background-worker shape this page is the exception to.
• Controlling LLM API Costs in Production — budget caps and model routing, which is where the usage rows you record here get used.

Other tracks:

• Handling Large JSON Payloads with Streaming — the same incremental mindset applied to inbound data.
• Setting Up FastAPI — the baseline app structure the endpoint above assumes.
• Structured Logging with structlog — how to make disconnect and stall events searchable instead of anecdotal.