Per-task event stream

A single request that runs five LLM calls can take twelve seconds. Holding the UI on a loading spinner for twelve seconds feels broken. Server-Sent Events let you push the sentiment result after two seconds, the emotion result after four, and the summary after eight. The user sees progress instead of a blank wait.

SSE event lifecycle

A single request emits one start, one task_start plus task_done or task_error per task, and one done.

agent/graph.py
python
from typing import AsyncGenerator


async def stream_agent(
    text: str, tasks: list[str]
) -> AsyncGenerator[dict, None]:
    """Yield per-task progress events."""
    tasks = _validate_tasks(tasks)

    yield {"event": "start", "tasks": tasks, "engine": "langgraph" if HAS_LANGGRAPH else "plain"}

    for name in tasks:
        yield {"event": "task_start", "task": name}
        fn = TASK_REGISTRY[name]
        try:
            result = await fn(text)
            yield {"event": "task_done", "task": name, "result": result}
        except Exception as e:
            yield {"event": "task_error", "task": name, "error": str(e)}

    yield {"event": "done"}

Five event types: start, task_start, task_done, task_error, done. The client knows exactly what is happening at every moment. task_error lets you surface a single task failure without tearing down the whole stream.

service.py
python
import json
from typing import AsyncGenerator


async def analyze_stream(request: AnalyzeRequest) -> AsyncGenerator[str, None]:
    async for event in stream_agent(request.text, request.tasks):
        yield f"data: {json.dumps(event)}\n\n"

SSE wire format is tiny: each event is data: <json>\n\n. Two newlines are the event boundary. FastAPI wraps this in a StreamingResponse with media_type text/event-stream.

router.py
python
from fastapi.responses import StreamingResponse


@router.post("/analyze/stream")
async def analyze_text_stream(request: AnalyzeRequest):
    return StreamingResponse(
        analyze_stream(request),
        media_type="text/event-stream",
        headers={"Cache-Control": "no-cache", "Connection": "keep-alive"},
    )

Cache-Control: no-cache prevents proxies from buffering the stream. Connection: keep-alive keeps the TCP connection open for the whole response. Without these headers, your events may arrive in batches instead of real time.

Ordering exercise: Arrange the events in the order the client sees them

Loading practice…