Streaming Responses with SSE

Why Stream

A 300-word answer can take several seconds. With streaming, the first words appear in well under a second and the user reads while the rest arrives. Browsers receive it as server-sent events (SSE): a long HTTP response made of data: ... lines separated by blank lines.

Example

The SSE format itself: how the server encodes chunks and how a client parses them back.

import json

def sse(data, event=None):
    lines = ([f"event: {event}"] if event else []) + [f"data: {json.dumps(data)}"]
    return "\n".join(lines) + "\n\n"

chunks = ["Shoes can ", "be returned ", "within 10 days."]
stream = "".join(sse({"text": c}) for c in chunks) + sse({"output_tokens": 9}, event="done")
print(stream)

# client side: split on blank lines, read the data field
text = ""
for message in stream.strip().split("\n\n"):
    fields = dict(line.split(": ", 1) for line in message.split("\n"))
    payload = json.loads(fields["data"])
    if fields.get("event") == "done":
        print("done, tokens:", payload["output_tokens"])
    else:
        text += payload["text"]
print("assembled:", text)

Output

data: {"text": "Shoes can "}

data: {"text": "be returned "}

data: {"text": "within 10 days."}

event: done
data: {"output_tokens": 9}


done, tokens: 9
assembled: Shoes can be returned within 10 days.

Streaming Endpoint

import json
from fastapi.responses import StreamingResponse

@app.post("/chat/stream")
async def chat_stream(body: ChatIn):
    async def events():
        async with client.messages.stream(
            model="claude-sonnet-5-5",
            max_tokens=1000,
            system=SYSTEM,
            messages=body.history + [{"role": "user", "content": body.message}],
        ) as stream:
            async for text in stream.text_stream:
                yield f"data: {json.dumps({'text': text})}\n\n"
            final = await stream.get_final_message()
            yield f"event: done\ndata: {json.dumps({'output_tokens': final.usage.output_tokens})}\n\n"
    return StreamingResponse(events(), media_type="text/event-stream",
                             headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"})

Reading the Stream in the Browser

const res = await fetch('/api/chat/stream', {
  method: 'POST',
  headers: { 'Content-Type': 'application/json' },
  body: JSON.stringify({ message }),
});
const reader = res.body!.pipeThrough(new TextDecoderStream()).getReader();
let buffer = '';
while (true) {
  const { value, done } = await reader.read();
  if (done) break;
  buffer += value;
  const events = buffer.split('\n\n');
  buffer = events.pop()!;                       // keep the incomplete part
  for (const ev of events) {
    const data = ev.split('\n').find((l) => l.startsWith('data: '));
    if (data && !ev.startsWith('event: done')) setReply((r) => r + JSON.parse(data.slice(6)).text);
  }
}

Note: Proxies like Nginx buffer responses by default, which breaks streaming. The X-Accel-Buffering: no header (or proxy_buffering off;) fixes it.