FastAPI: Streaming, SSE, WebSockets & files

Not every response is a small JSON document produced all at once. Exports have millions of rows, LLM answers arrive token by token, dashboards want live updates, chats need two-way messages, and users upload files. FastAPI covers each with a specific tool: StreamingResponse for arbitrary streams, JSON Lines (0.134+) and Server-Sent Events (0.135+) by simply yielding from your endpoint, WebSockets for two-way traffic, and UploadFile/FileResponse for files. All of them hold a connection open, which changes the operational rules too.

1. Which tool?

2. JSON Lines: just yield (FastAPI 0.134+)

an endpoint that yields typed items, consumed as a stream

import asyncio
from collections.abc import AsyncIterable

from fastapi import FastAPI
from fastapi.testclient import TestClient
from pydantic import BaseModel

class OrderRow(BaseModel):
    id: int
    total: float

app = FastAPI()

@app.get("/exports/orders")
async def export_orders() -> AsyncIterable[OrderRow]:      # the return type validates + documents items
    for i in range(1, 6):
        await asyncio.sleep(0.01)                             # e.g. fetching the next page from the DB
        yield OrderRow(id=i, total=i * 99.5)

client = TestClient(app)
with client.stream("GET", "/exports/orders") as r:
    print(r.status_code, r.headers["content-type"])
    for line in r.iter_lines():                               # the client processes rows as they arrive
        print("  row:", line)

Output

200 application/jsonl
  row: {"id":1,"total":99.5}
  row: {"id":2,"total":199.0}
  row: {"id":3,"total":298.5}
  row: {"id":4,"total":398.0}
  row: {"id":5,"total":497.5}

With a declared AsyncIterable[Item] (or Iterable[Item] for a def endpoint), each item is validated and serialized by Pydantic’s Rust core. For binary streams (video, audio, generated files), return a StreamingResponse wrapping a generator of bytes.

3. Server-Sent Events (FastAPI 0.135+)

progress events with ids, names, and a raw [DONE] sentinel

import asyncio
from collections.abc import AsyncIterable
from typing import Annotated

from fastapi import FastAPI, Header
from fastapi.sse import EventSourceResponse, ServerSentEvent
from fastapi.testclient import TestClient

app = FastAPI()
STEPS = ["validating", "charging card", "reserving stock", "confirmed"]

@app.get("/orders/{order_id}/events", response_class=EventSourceResponse)
async def order_events(order_id: int,
                       last_event_id: Annotated[int | None, Header()] = None) -> AsyncIterable[ServerSentEvent]:
    start = (last_event_id or 0) + 1                         # a reconnecting browser sends Last-Event-ID
    for n in range(start, len(STEPS) + 1):
        await asyncio.sleep(0.01)
        yield ServerSentEvent(data={"order": order_id, "step": STEPS[n - 1]}, event="progress", id=str(n))
    yield ServerSentEvent(raw_data="[DONE]", event="done")

client = TestClient(app)
r = client.get("/orders/1042/events")
print(r.headers["content-type"], "| cache-control:", r.headers["cache-control"],
      "| x-accel-buffering:", r.headers["x-accel-buffering"])
print(r.text.split("\n\n")[0], "\n…")
resumed = client.get("/orders/1042/events", headers={"Last-Event-ID": "2"}).text
print("resumed after id 2:", [line for line in resumed.splitlines() if line.startswith("data")])

Output

text/event-stream; charset=utf-8 | cache-control: no-cache | x-accel-buffering: no
event: progress
data: {"order": 1042, "step": "validating"}
id: 1 
…
resumed after id 2: ['data: {"order": 1042, "step": "reserving stock"}', 'data: {"order": 1042, "step": "confirmed"}', 'data: [DONE]']

4. WebSockets

authenticated rooms with broadcast and clean disconnects

from collections import defaultdict

from fastapi import FastAPI, WebSocket, WebSocketDisconnect, status
from fastapi.testclient import TestClient

TOKENS = {"t-asha": "asha", "t-ravi": "ravi"}

class Rooms:
    def __init__(self):
        self.members: dict[str, set[WebSocket]] = defaultdict(set)
    async def join(self, room, ws):
        await ws.accept()
        self.members[room].add(ws)
    def leave(self, room, ws):
        self.members[room].discard(ws)
    async def broadcast(self, room, message):
        for ws in list(self.members[room]):
            await ws.send_json(message)

rooms = Rooms()
app = FastAPI()

@app.websocket("/ws/rooms/{room}")
async def room_socket(ws: WebSocket, room: str, token: str | None = None):
    user = TOKENS.get(token or "")
    if user is None:                                       # authenticate BEFORE accepting
        await ws.close(code=status.WS_1008_POLICY_VIOLATION)
        return
    await rooms.join(room, ws)
    await rooms.broadcast(room, {"system": f"{user} joined"})
    try:
        while True:
            text = await ws.receive_text()
            await rooms.broadcast(room, {"from": user, "text": text[:500]})   # cap message size
    except WebSocketDisconnect:
        rooms.leave(room, ws)
        await rooms.broadcast(room, {"system": f"{user} left"})

client = TestClient(app)
with client.websocket_connect("/ws/rooms/support?token=t-asha") as asha:
    print("asha got:", asha.receive_json())
    with client.websocket_connect("/ws/rooms/support?token=t-ravi") as ravi:
        print("asha got:", asha.receive_json(), "| ravi got:", ravi.receive_json())
        ravi.send_text("Hi Asha, how can I help?")
        print("asha got:", asha.receive_json())
    print("asha got:", asha.receive_json())
try:
    with client.websocket_connect("/ws/rooms/support?token=bad") as ws:
        ws.receive_json()
except Exception as e:
    print("bad token:", type(e).__name__, getattr(e, "code", ""))

Output

asha got: {'system': 'asha joined'}
asha got: {'system': 'ravi joined'} | ravi got: {'system': 'ravi joined'}
asha got: {'from': 'ravi', 'text': 'Hi Asha, how can I help?'}
asha got: {'system': 'ravi left'}
bad token: WebSocketDisconnect 1008

5. Uploads with limits

stream the upload to disk, enforce size and type, never trust the filename

import hashlib
import uuid
from pathlib import Path
from typing import Annotated

from fastapi import FastAPI, File, Form, HTTPException, UploadFile
from fastapi.testclient import TestClient

MAX_BYTES = 2 * 1024 * 1024                    # 2 MiB
ALLOWED = {b"%PDF": "application/pdf", b"\x89PNG": "image/png"}
UPLOADS = Path("uploads")
UPLOADS.mkdir(exist_ok=True)

app = FastAPI()

@app.post("/documents", status_code=201)
async def upload(file: Annotated[UploadFile, File()], title: Annotated[str, Form(max_length=100)]):
    head = await file.read(4)
    if head not in ALLOWED:                                   # check the bytes, not the client's claim
        raise HTTPException(415, "only PDF and PNG files are accepted")
    digest, size = hashlib.sha256(head), len(head)
    target = UPLOADS / f"{uuid.uuid4().hex}.bin"              # never use the client's filename as a path
    with target.open("wb") as out:
        out.write(head)
        while chunk := await file.read(64 * 1024):            # stream in chunks: memory stays flat
            size += len(chunk)
            if size > MAX_BYTES:
                out.close()
                target.unlink()
                raise HTTPException(413, f"file larger than {MAX_BYTES // 1024} KiB")
            digest.update(chunk)
            out.write(chunk)
    return {"title": title, "type": ALLOWED[head], "bytes": size, "sha256": digest.hexdigest()[:12]}

client = TestClient(app)
pdf = b"%PDF-1.7\n" + b"x" * 5000
print(client.post("/documents", files={"file": ("../../etc/passwd", pdf, "application/pdf")},
                  data={"title": "Invoice"}).json())
print(client.post("/documents", files={"file": ("evil.pdf", b"MZ\x90\x00 exe", "application/pdf")},
                  data={"title": "x"}).status_code)
print(client.post("/documents", files={"file": ("big.pdf", b"%PDF" + b"0" * (3 * 1024 * 1024))},
                  data={"title": "x"}).status_code)

Output

{'title': 'Invoice', 'type': 'application/pdf', 'bytes': 5009, 'sha256': '901f8e1a2f67'}
415
413

6. Downloads

FileResponse: content type, filename, and range requests

from pathlib import Path

from fastapi import FastAPI
from fastapi.responses import FileResponse
from fastapi.testclient import TestClient

Path("report.csv").write_text("id,total\n" + "\n".join(f"{i},{i * 10}" for i in range(1, 1001)), encoding="utf-8")
app = FastAPI()

@app.get("/reports/latest")
def latest_report():
    return FileResponse("report.csv", media_type="text/csv", filename="orders-2026-09.csv")

client = TestClient(app)
r = client.get("/reports/latest")
print(r.status_code, r.headers["content-type"], "|", r.headers["content-disposition"], "|", len(r.content), "bytes")
r = client.get("/reports/latest", headers={"Range": "bytes=0-19"})       # resumable downloads, video seeking
print(r.status_code, r.headers["content-range"], repr(r.text))

Output

200 text/csv; charset=utf-8 | attachment; filename="orders-2026-09.csv" | 9794 bytes
206 bytes 0-19/9794 'id,total\r\n1,10\r\n2,20'

7. Operating long-lived connections

Recap