Stream responses with Server-Sent Events

Approach

Return a StreamingResponse with media_type="text/event-stream" from an async generator that yields data: ...\n\n frames. Send heartbeats to keep proxies from closing idle connections, stop when the client disconnects, and disable proxy buffering (X-Accel-Buffering: no for Nginx).

Solution

import asyncio, json
from fastapi import Request
from fastapi.responses import StreamingResponse


def sse(event: str, data: dict) -> str:
    return f"event: {event}\ndata: {json.dumps(data)}\n\n"


@app.get("/jobs/{job_id}/events")
async def job_events(job_id: str, request: Request, user: CurrentUser):
    async def stream():
        last = None
        while not await request.is_disconnected():
            job = await jobs.get(job_id, owner=user.id)
            if job.progress != last:
                last = job.progress
                yield sse("progress", {"progress": job.progress})
            if job.status in {"done", "failed"}:
                yield sse(job.status, {"result_url": job.result_url})
                return
            yield ": heartbeat\n\n"
            await asyncio.sleep(1)

    return StreamingResponse(stream(), media_type="text/event-stream",
                             headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"})

# Browser: const es = new EventSource(`/jobs/${id}/events`); es.addEventListener("progress", e => ...)

Pitfalls

• Proxy buffering delaying events until the response ends.
• Not stopping work when the client disconnects.
• Expecting EventSource to send custom headers (use cookies or a fetch-based SSE client).

Variations

• sse-starlette adds EventSourceResponse with built-in pings.
• WebSockets when you need bidirectional messaging.

Further reading: MDN: Using server-sent events