Careeroza

Streaming Backend with FastAPI

Server-Sent Events

Browsers receive a streamed reply most simply as Server-Sent Events (SSE): a long HTTP response with Content-Type: text/event-stream where each event is a line data: ... followed by a blank line. Your FastAPI backend reads the model stream and forwards each text piece as an event. The API key stays on the server.

The Endpoint

# main.py
import json
import anthropic
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from pydantic import BaseModel

app = FastAPI()
client = anthropic.AsyncAnthropic()
SYSTEM = "You are the ShopEase support assistant. Keep answers under 80 words."

class ChatRequest(BaseModel):
    conversation_id: str
    message: str

@app.post("/api/chat")
async def chat(req: ChatRequest):
    history = await load_history(req.conversation_id)          # e.g. from Redis
    history.append({"role": "user", "content": req.message})

    async def events():
        try:
            async with client.messages.stream(model="claude-sonnet-5-5", max_tokens=800,
                                              system=SYSTEM, messages=history) as stream:
                async for text in stream.text_stream:
                    yield f"data: {json.dumps({'type': 'text', 'text': text})}\n\n"
                final = await stream.get_final_message()
            history.append({"role": "assistant", "content": final.content})
            await save_history(req.conversation_id, history)
            yield f"data: {json.dumps({'type': 'done'})}\n\n"
        except anthropic.APIError:
            yield f"data: {json.dumps({'type': 'error', 'message': 'The assistant is busy. Please try again.'})}\n\n"

    return StreamingResponse(events(), media_type="text/event-stream")

Example

Test it with curl (-N turns off buffering so events print as they arrive):

curl -N -X POST http://localhost:8000/api/chat \
  -H "Content-Type: application/json" \
  -d '{"conversation_id": "c-77", "message": "Do you deliver on Sundays?"}'

Example output

data: {"type": "text", "text": "Yes"}

data: {"type": "text", "text": ", ShopEase delivers on Sundays in most metro"}

data: {"type": "text", "text": " cities. Delivery slots show at checkout."}

data: {"type": "done"}

The SSE Wire Format

Example

Formatting and parsing SSE is plain string work. The browser does the parsing part; here both halves run in Python:

import json

def sse(event):
    return f"data: {json.dumps(event)}\n\n"

stream_body = "".join(sse(e) for e in [
    {"type": "text", "text": "Yes, "},
    {"type": "text", "text": "we deliver on Sundays."},
    {"type": "done"},
])
print(repr(stream_body[:60]), "...")

def parse_sse(body):
    for block in body.split("\n\n"):
        if block.startswith("data: "):
            yield json.loads(block[len("data: "):])

reply = ""
for event in parse_sse(stream_body):
    if event["type"] == "text":
        reply += event["text"]
    print(event["type"], "->", repr(reply))

Output

'data: {"type": "text", "text": "Yes, "}\n\ndata: {"type": "tex' ...
text -> 'Yes, '
text -> 'Yes, we deliver on Sundays.'
done -> 'Yes, we deliver on Sundays.'