Real-Time AI Pipeline Progress with Server-Sent Events

I built a multi-agent AI pipeline that runs planning cycles of five to fifteen minutes. Multiple specialized agents run in sequence and parallel — analysis, review, synthesis, revision — each producing findings and status updates that are useful to see in real time.

The first version of the API returned the full result at the end. The client submitted a request, waited, and eventually got a complete response back. This worked. It also meant that for ten minutes, the user stared at a spinner with no idea whether the system was making progress, stuck, or broken.

A ten-minute API call with no intermediate feedback feels broken even when it is working perfectly. The absence of signal is indistinguishable from the absence of progress.

The fix was Server-Sent Events. Not WebSockets, not long polling, not a message queue. Just SSE — the unglamorous, one-way, HTTP-based protocol that has been in browsers since 2009 and still gets ignored in favor of more complex alternatives.

Why Not WebSockets

The instinct is to reach for WebSockets. They are bidirectional, well-supported, and feel like the "serious" choice for real-time communication. But this pipeline's progress stream is strictly one-way. The server sends progress updates. The client receives them. There is no client-to-server messaging during the cycle — the request parameters are sent once at the start, as a normal HTTP POST.

WebSockets add complexity that does not pay for itself in this scenario:

• Connection upgrade handshake — an extra round trip before any data flows.
• Custom protocol — you need a message format for errors, close events, and reconnection logic.
• No built-in reconnection — the client has to implement reconnection logic manually.
• No HTTP semantics — status codes, headers, and caching all disappear.

SSE gives you all of this for free because it is HTTP. The connection is a normal HTTP response with Content-Type: text/event-stream. The browser's EventSource API handles reconnection automatically, including sending Last-Event-ID so the server can resume from where it left off.

If your real-time stream is one-way, WebSockets are overkill. SSE does the job with less code, better HTTP semantics, and free reconnection.

The Server Side

FastAPI makes this straightforward. The endpoint is an async generator that yields events. FastAPI wraps it in a StreamingResponse with the right content type.

from fastapi import APIRouter
from fastapi.responses import StreamingResponse
import json

router = APIRouter()

@router.post("/sessions/{session_id}/stream")
async def stream_session(session_id: str):
    async def event_generator():
        while True:
            update = await progress_queue.get(session_id)
            if update is None:
                yield f"event: complete\ndata: {json.dumps({'status': 'done'})}\n\n"
                break

            event_type = update.get("event", "progress")
            yield f"event: {event_type}\ndata: {json.dumps(update)}\n\n"

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

The generator pulls progress updates from an internal queue — one per session — and formats them as SSE events. When the cycle completes, the queue returns None and the generator yields a final complete event before closing.

Each event has a type (cycle_started, step_started, step_completed, synthesis_ready, revision_complete, complete) and a JSON data payload. The client can listen for specific event types or treat them all as generic progress updates.

The X-Accel-Buffering: no header is important if you run behind nginx. By default, nginx buffers responses. For SSE, that means the browser receives events in batches every few seconds instead of immediately. That header disables buffering for this response, so events arrive in real time.

Handling Disconnections

Clients disconnect. They close laptops, lose Wi-Fi, navigate to other pages. The server needs to handle this gracefully.

FastAPI's StreamingResponse handles the happy path: when the client disconnects, the async generator is eventually cancelled. Python's asyncio raises a CancelledError in the generator, which cleans up the queue subscription.

But "eventually" is doing work in that sentence. The generator only notices the cancellation when it tries to yield. If the generator is blocked waiting for the next progress update from a long-running step — say, an analysis agent is still processing — the cancellation does not propagate until the generator wakes up.

The fix is a timeout on the queue read:

async def event_generator():
    while True:
        try:
            update = await asyncio.wait_for(
                progress_queue.get(session_id),
                timeout=15.0,
            )
        except asyncio.TimeoutError:
            # Client may still be connected. Send a keepalive.
            yield ": keepalive\n\n"
            continue

Every 15 seconds of silence, the server sends a comment line (starts with :). SSE clients ignore comments, but the write operation forces the connection to flush. If the client has disconnected, the write fails and the generator exits.

A keepalive comment is two bytes of traffic that prevents a silent disconnection from tying up server resources indefinitely. Without it, your server accumulates ghost connections.

The Client Side

The browser's EventSource API is refreshingly simple:

const source = new EventSource(`/sessions/${sessionId}/stream`);

source.addEventListener("review_completed", (event) => {
    const data = JSON.parse(event.data);
    updateReviewPanel(data.agent, data.findings);
});

source.addEventListener("complete", (event) => {
    source.close();
    showFinalPlan();
});

source.onerror = () => {
    // EventSource reconnects automatically.
    // Show a "reconnecting..." indicator.
};

No libraries. No WebSocket wrapper. No reconnection logic. The browser handles reconnection, sends Last-Event-ID, and parses the event stream. The application code is just event listeners.

When SSE Is Not Enough

SSE is one-way, server-to-client. If you need the client to send messages during the stream — typing into a chat, voting, sending commands — SSE cannot do it without a separate POST endpoint for the upstream channel. At that point, the question is whether maintaining two channels (SSE down, POST up) is simpler than one WebSocket channel.

For this pipeline, it is. The client sends one request to start the cycle. Everything after that is the server reporting progress. Two channels would be more complex than one, but the one channel does not need to be bidirectional.

The decision is not "SSE or WebSockets." It is "do I need upstream communication during this stream?" If no, SSE. If yes, weigh the two-channel approach against WebSockets.

The Principle

Real-time progress is not a feature. It is a trust mechanism. When a user submits a request that takes ten minutes, silence is anxiety. A progress feed — even a minimal one that says "analysis step in progress" — converts that anxiety into patience.

The cost of adding SSE to a FastAPI endpoint is about thirty lines of server code and zero client-side dependencies. The SSE protocol is just HTTP with a specific content type and a newline-delimited format. There is no protocol negotiation, no connection upgrade, no frame format to parse.

When you have a long-running operation and a user waiting for it, add an SSE endpoint. Do not build a WebSocket server. Do not set up a message broker. Do not write a reconnection library. The boring protocol from 2009 is the right answer.

The key insight is about who the customer is. When your API serves AI agents as consumers, the UX expectations flip. A human accepts a spinner for two seconds and starts refreshing. An agent accepts silence for two seconds and starts retrying. Neither is right for a fifteen-minute operation. SSE gives both — human and machine — the signal they need to trust that the work is still happening.