FastAPI + LLM Streaming: Production Backend Template

A production-ready FastAPI backend pattern for streaming LLM responses to a frontend — server-sent events, error handling, and cost tracking in one template.

Author

AICredits Team

Published

8 Sept 2026

Reading time

7 min read

Why a dedicated backend, not just client-side calls

Calling an LLM API directly from a frontend exposes your API key to anyone who opens dev tools. A FastAPI backend sits between your frontend and AICredits, holding the key server-side and giving you a place to add auth, logging, and rate limiting specific to your application — on top of what AICredits already enforces at the key level.

The streaming endpoint

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from openai import OpenAI
 
app = FastAPI()
client = OpenAI(base_url="https://api.aicredits.in/v1", api_key="sk-your-aicredits-key")
 
class ChatRequest(BaseModel):
    message: str
    model: str = "openai/gpt-4o-mini"
 
async def stream_completion(message: str, model: str):
    stream = client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": message}],
        stream=True,
    )
    for chunk in stream:
        delta = chunk.choices[0].delta.content
        if delta:
            yield f"data: {delta}\n\n"
    yield "data: [DONE]\n\n"
 
@app.post("/chat")
async def chat(request: ChatRequest):
    return StreamingResponse(
        stream_completion(request.message, request.model),
        media_type="text/event-stream",
    )

Consuming the stream from a frontend

const response = await fetch("/chat", {
  method: "POST",
  headers: { "Content-Type": "application/json" },
  body: JSON.stringify({ message: "Explain server-sent events." }),
});
 
const reader = response.body.getReader();
const decoder = new TextDecoder();
let result = "";
 
while (true) {
  const { done, value } = await reader.read();
  if (done) break;
  const chunk = decoder.decode(value);
  const lines = chunk.split("\n\n").filter(Boolean);
  for (const line of lines) {
    const data = line.replace("data: ", "");
    if (data === "[DONE]") break;
    result += data;
    // update UI with `result` as tokens arrive
  }
}

Adding cost tracking per request

Streaming responses don't return usage data mid-stream on every provider, but most send a final usage payload in the last chunk. Capture it to log actual cost per request rather than estimating:

async def stream_completion(message: str, model: str):
    usage_data = {}
    stream = client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": message}],
        stream=True,
        stream_options={"include_usage": True},
    )
    for chunk in stream:
        if chunk.choices and chunk.choices[0].delta.content:
            yield f"data: {chunk.choices[0].delta.content}\n\n"
        if chunk.usage:
            usage_data = chunk.usage.model_dump()
 
    # Log usage_data (prompt_tokens, completion_tokens) to your own analytics/DB here
    yield "data: [DONE]\n\n"

Error handling for production traffic

Provider timeouts and rate limits happen. Wrap the stream generator so a mid-stream failure degrades gracefully instead of leaving the frontend hanging:

async def stream_completion(message: str, model: str):
    try:
        stream = client.chat.completions.create(
            model=model,
            messages=[{"role": "user", "content": message}],
            stream=True,
        )
        for chunk in stream:
            delta = chunk.choices[0].delta.content
            if delta:
                yield f"data: {delta}\n\n"
    except Exception as e:
        yield f"data: [ERROR] {str(e)}\n\n"
    finally:
        yield "data: [DONE]\n\n"

For a more robust retry strategy on top of this — exponential backoff on 429s and 5xxs before the stream even starts — see the dedicated retry strategy guide.

Applying a per-request budget check before streaming starts

If your application has its own user-level quotas on top of AICredits' key-level budget, check them before opening the stream — cancelling mid-stream is a worse user experience than rejecting the request upfront:

@app.post("/chat")
async def chat(request: ChatRequest, user=Depends(get_current_user)):
    if user.remaining_quota <= 0:
        return {"error": "Daily quota exceeded"}, 429
    return StreamingResponse(stream_completion(request.message, request.model), media_type="text/event-stream")

Frequently Asked Questions

Should I use server-sent events or WebSockets for streaming?

Server-sent events (as shown here) are simpler and sufficient for one-directional LLM token streaming. Reach for WebSockets only if you need bidirectional communication — e.g., letting the user interrupt generation mid-stream.

Does this pattern work with async frameworks other than FastAPI?

The core idea — an async generator yielding tokens as they arrive — applies to any ASGI framework (Starlette, Django async views). FastAPI is used here because StreamingResponse makes the wiring minimal.

How do I handle concurrent streaming requests without blocking the server?

Use the async OpenAI client (AsyncOpenAI) instead of the sync client shown above, and run the FastAPI app with an ASGI server like uvicorn with multiple workers for real concurrency under load.

Get started

Create a free account and build this template against a real key.
Related: Streaming LLM Responses in Python: The Complete Guide and How to Build a Retry Strategy for LLM API Calls.

AICredits

Build with any AI model. Pay in INR.

One OpenAI-compatible API for every major model, with a prepaid rupee wallet. Top up via UPI — no international credit card, no forex surprises.

Top up via UPI · Prepaid INR wallet · No international card needed

Guardrails in Production: PII Masking and Keyword Blocking

6 min read

Rate Limits, Budgets, and Per-Key Controls: Taming Team AI Spend