Streaming, Server-Sent Events, and WebSockets — Python Mastery

Why this matters

Real-time matters: LLM token streaming, live dashboards, collaborative editing, chat. The two browser-friendly options are Server-Sent Events (SSE — simple, unidirectional server→client) and WebSockets (bidirectional, lower-level). This lesson covers both with FastAPI on the server and httpx / browsers on the client.

Learning objectives

Implement SSE endpoints in FastAPI.
Consume SSE from httpx and from the browser.
Implement WebSocket endpoints in FastAPI.
Apply heartbeat, reconnection, and backpressure patterns.
Pick SSE vs WebSocket vs polling.

1. SSE vs WebSocket vs polling

Property	Polling	SSE	WebSocket
Protocol	HTTP	HTTP (text/event-stream)	WebSocket (HTTP upgrade)
Direction	Both (request/response)	Server → Client	Bidirectional
Reconnect	Manual	Auto in browser	Manual
Proxy / CDN friendly	Yes	Yes (it's just HTTP)	Sometimes painful
Browser support	Native	Native (`EventSource`)	Native
Use when	Infrequent updates	Push from server (LLMs, news feeds)	Bi-directional chat / games

Default for LLM streaming: SSE. Default for chat / collaborative editing: WebSocket.

2. SSE basics

SSE is a one-way text stream over HTTP with a specific format:

data: hello

data: this is a longer message
data: spread over two "data:" lines

event: ping
data: {"t": 1234567890}

retry: 5000

Rules:

Lines starting with data: carry payload.
Multiple data: lines in one event are joined with \n.
event: sets an event name (default "message").
id: sets the event ID (browser sends it back on reconnect).
retry: tells the browser the reconnect delay (ms).
Events are separated by blank lines.

Content-Type: text/event-stream.

3. FastAPI SSE endpoint

python

import asyncio
import json
from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()

async def event_generator():
    for i in range(10):
        await asyncio.sleep(0.5)
        data = json.dumps({"i": i, "msg": f"chunk-{i}"})
        yield f"data: {data}\n\n"

@app.get("/stream")
async def stream():
    return StreamingResponse(event_generator(), media_type="text/event-stream")

Test in browser console:

javascript

const es = new EventSource("/stream");
es.onmessage = (e) => console.log(JSON.parse(e.data));

Or curl:

powershell

curl -N http://127.0.0.1:8000/stream

4. LLM token streaming (the most common SSE use)

python

from openai import AsyncOpenAI
client = AsyncOpenAI()

async def llm_stream(prompt: str):
    response = await client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        stream=True,
    )
    async for chunk in response:
        token = chunk.choices[0].delta.content
        if token:
            yield f"data: {json.dumps({'token': token})}\n\n"
    yield "data: [DONE]\n\n"

@app.get("/chat")
async def chat(prompt: str):
    return StreamingResponse(llm_stream(prompt), media_type="text/event-stream")

The frontend opens an EventSource, displays tokens as they arrive. This is the ChatGPT typing effect.

Handling client disconnect

python

from starlette.requests import Request

async def llm_stream(prompt: str, request: Request):
    response = await client.chat.completions.create(...)
    async for chunk in response:
        if await request.is_disconnected():
            break
        token = chunk.choices[0].delta.content
        if token:
            yield f"data: {json.dumps({'token': token})}\n\n"

@app.get("/chat")
async def chat(prompt: str, request: Request):
    return StreamingResponse(llm_stream(prompt, request), media_type="text/event-stream")

Without this check you'll keep generating expensive tokens for a closed connection.

5. Heartbeats / keep-alive

Some proxies close idle connections after 30-60s. Send periodic comments (: ) or a keepalive event:

python

async def event_generator_with_heartbeat():
    while True:
        try:
            data = await asyncio.wait_for(queue.get(), timeout=15)
            yield f"data: {data}\n\n"
        except asyncio.TimeoutError:
            yield ":keepalive\n\n"        # SSE comment; client ignores

Lines starting with : are SSE comments. Browsers ignore them; proxies see traffic.

6. Consume SSE from Python with httpx

python

import httpx, json

async def consume_sse(url: str):
    async with httpx.AsyncClient(timeout=None) as client:
        async with client.stream("GET", url) as r:
            r.raise_for_status()
            event_data = []
            async for line in r.aiter_lines():
                if line == "":
                    if event_data:
                        yield "\n".join(event_data)
                        event_data = []
                elif line.startswith("data:"):
                    event_data.append(line.removeprefix("data: "))

async def main():
    async for event in consume_sse("https://api.example.com/stream"):
        print(json.loads(event))

For full SSE protocol (events, IDs, retry), use httpx-sse (uv add httpx-sse):

python

import httpx
from httpx_sse import aconnect_sse

async with httpx.AsyncClient() as client:
    async with aconnect_sse(client, "GET", url) as event_source:
        async for sse in event_source.aiter_sse():
            print(sse.event, sse.data, sse.id)

7. WebSockets in FastAPI

python

from fastapi import FastAPI, WebSocket, WebSocketDisconnect

app = FastAPI()

@app.websocket("/ws")
async def websocket_endpoint(ws: WebSocket):
    await ws.accept()
    try:
        while True:
            msg = await ws.receive_text()
            await ws.send_text(f"echo: {msg}")
    except WebSocketDisconnect:
        print("client disconnected")

Browser:

javascript

const ws = new WebSocket("ws://127.0.0.1:8000/ws");
ws.onmessage = (e) => console.log(e.data);
ws.onopen = () => ws.send("hello");

JSON messages

python

await ws.receive_json()
await ws.send_json({"type": "tick", "value": 42})

Binary

python

await ws.receive_bytes()
await ws.send_bytes(b"\x00\x01\x02")

8. Connection manager / broadcast

python

from collections import defaultdict
from fastapi import WebSocket

class Manager:
    def __init__(self):
        self.connections: dict[str, set[WebSocket]] = defaultdict(set)

    async def connect(self, room: str, ws: WebSocket):
        await ws.accept()
        self.connections[room].add(ws)

    def disconnect(self, room: str, ws: WebSocket):
        self.connections[room].discard(ws)

    async def broadcast(self, room: str, msg: dict):
        dead = []
        for ws in self.connections[room]:
            try:
                await ws.send_json(msg)
            except Exception:
                dead.append(ws)
        for ws in dead:
            self.connections[room].discard(ws)

mgr = Manager()

@app.websocket("/rooms/{name}")
async def room(name: str, ws: WebSocket):
    await mgr.connect(name, ws)
    try:
        while True:
            msg = await ws.receive_json()
            await mgr.broadcast(name, {"user": msg["user"], "text": msg["text"]})
    except WebSocketDisconnect:
        mgr.disconnect(name, ws)

For multi-process / multi-machine (uvicorn --workers 4), single-process broadcast doesn't reach all clients. Use Redis Pub/Sub or a message bus to distribute events across workers.

python

import redis.asyncio as redis
r = redis.from_url("redis://localhost")
async def publish(room, msg):
    await r.publish(f"room:{room}", json.dumps(msg))

async def subscribe(room):
    pubsub = r.pubsub()
    await pubsub.subscribe(f"room:{room}")
    async for msg in pubsub.listen():
        if msg["type"] == "message":
            yield json.loads(msg["data"])

9. Authentication

SSE / WebSocket auth

Browsers can't easily set headers on EventSource or new WebSocket(...). Options:

Cookie-based auth — works automatically with same-origin.
Token in query param — wss://api/ws?token=... (logged in URLs! Treat as bearer).
First WebSocket message — client sends {"token": "..."} immediately after connect.

For SSE, you can use fetch with EventSource polyfill (e.g., @microsoft/fetch-event-source) that supports custom headers.

Server side:

python

@app.websocket("/ws")
async def ws_endpoint(ws: WebSocket, token: str = Query(...)):
    user = await authenticate(token)
    if not user:
        await ws.close(code=1008)
        return
    await ws.accept()
    ...

10. Backpressure

If you produce events faster than the client consumes, memory grows. Patterns:

Drop oldest (financial tickers): keep a deque, drop on overflow.
Coalesce (cursor updates): keep only latest.
Apply backpressure (real work): block the producer if the queue is full (asyncio.Queue(maxsize=N)).

python

out = asyncio.Queue(maxsize=100)

async def producer():
    while True:
        item = compute()
        await out.put(item)        # blocks when full

async def consumer(ws):
    while True:
        item = await out.get()
        await ws.send_json(item)

11. Reconnect strategy

EventSource auto-reconnects on disconnect (after retry: ms). Use id: so the client can resume from where it left off:

python

async def gen():
    last_id = 0
    async for item in events_after(last_id):
        last_id = item.id
        yield f"id: {item.id}\n"
        yield f"data: {json.dumps(item.data)}\n\n"

Server reads Last-Event-ID header on reconnects:

python

@app.get("/stream")
async def stream(request: Request):
    last_id = int(request.headers.get("Last-Event-ID", 0))
    return StreamingResponse(gen_after(last_id), media_type="text/event-stream")

WebSockets have no auto-reconnect — implement client-side (browser libraries like reconnecting-websocket help).

12. Production checklist

Heartbeats / keepalives every 15-30s.
Detect client disconnect (request.is_disconnected() or WebSocketDisconnect).
Apply backpressure (bounded queues).
Authenticate before accept().
Set CORS headers if cross-origin.
For multi-worker deployments, use a message bus for broadcast.
Time-out idle connections (closeable from server).
Don't log message bodies (may contain PII).
Compress large messages (permessage-deflate for WebSocket; gzip middleware for SSE).
Monitor: open connections, dropped messages, queue depth.

13. Worked example: live LLM chat

python

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from openai import AsyncOpenAI
import json

app = FastAPI()
client = AsyncOpenAI()

@app.websocket("/chat")
async def chat(ws: WebSocket):
    await ws.accept()
    history = [{"role": "system", "content": "You are concise."}]
    try:
        while True:
            user_msg = await ws.receive_json()
            history.append({"role": "user", "content": user_msg["text"]})

            assistant_text = []
            response = await client.chat.completions.create(
                model="gpt-4o-mini",
                messages=history,
                stream=True,
            )
            async for chunk in response:
                token = chunk.choices[0].delta.content
                if token:
                    assistant_text.append(token)
                    await ws.send_json({"type": "token", "text": token})

            full = "".join(assistant_text)
            history.append({"role": "assistant", "content": full})
            await ws.send_json({"type": "done"})
    except WebSocketDisconnect:
        pass

Frontend opens the WebSocket, sends user messages, displays streaming tokens. ~30 lines on each side.

Hands-on lab (2 hours)

Build a FastAPI SSE endpoint that emits a counter every second. View in browser with EventSource.
Add heartbeats (:keepalive) every 15 seconds.
Consume the SSE from Python using httpx-sse.
Build a WebSocket echo endpoint; connect with a browser.
Build a room-based chat with a connection manager; broadcast messages.
Add token-in-query authentication.
Stream OpenAI / Anthropic responses through SSE; show client-side reconnect / Last-Event-ID handling.

Common pitfalls

Forgetting media_type="text/event-stream" → browser doesn't parse as SSE.
Forgetting the blank line between events.
EventSource with non-GET request — not supported (use fetch polyfill).
Keeping LLM streaming alive after client disconnects (burns money).
Unbounded broadcast lists; growing memory; never freeing dead connections.
WebSocket auth checked after accept() (too late; attacker is already in).
Forgetting that multi-worker uvicorn breaks single-process broadcast.

Self-check

When SSE vs WebSocket vs polling?
What does Retry-After style work look like with EventSource?
How do you detect a client disconnect?
How do you broadcast to all clients across workers?
What problem does backpressure solve?

References

HTML5 EventSource spec: https://html.spec.whatwg.org/multipage/server-sent-events.html.
FastAPI WebSocket docs: https://fastapi.tiangolo.com/advanced/websockets/.
httpx-sse: https://github.com/florimondmanca/httpx-sse.
"Server-Sent Events vs WebSockets" — MDN.
OpenAI streaming docs.