SSE event contract

Users wait. A pipeline that takes 10 seconds feels broken without intermediate feedback. Server-Sent Events let the backend emit pipeline stages as they complete: connected, analyzing, rewrites generated, sources retrieved, answer content, eval scores, done. The client renders progress immediately and updates as each stage arrives.

SSE event sequence

One HTTP connection, many events. Each event is a JSON payload the client updates against.

One quick introduction before the streaming code. Every JSON payload you have been sending to /ask deserializes into the AskRequest model: question plus the rewrite_n, anonymize, with_evaluation, and top_k fields with their defaults. The router hands that model to a small service function named ask, which forwards it to run_graph and returns the answer, rewrites, sources, and eval_score shape you have seen in every curl response. The streaming endpoint reuses exactly this function.

service.py
python
async def ask(request: AskRequest) -> Dict[str, Any]:
    """Run the full parent graph and return a structured response."""
    return await rag_graph.run_graph(
        question=request.question,
        rewrite_n=request.rewrite_n,
        top_k=request.top_k,
        anonymize=request.anonymize,
        with_evaluation=request.with_evaluation,
    )

The non-streaming path is a one-liner: unpack the AskRequest fields and delegate to run_graph. The streaming wrapper below calls this same function and slices its return dict into individual SSE events.

service.py
python
async def ask_stream(request: AskRequest) -> AsyncGenerator[str, None]:
    """SSE wrapper that reports pipeline stages as they complete."""
    yield f"data: {json.dumps({'status': 'connected', 'message': 'Starting advanced RAG pipeline'})}\n\n"
    yield f"data: {json.dumps({'thinking': {'category': 'analysis', 'content': f'Analyzing question: {request.question}', 'timestamp': datetime.now().isoformat()}})}\n\n"

    try:
        result = await ask(request)
    except Exception as e:
        yield f"data: {json.dumps({'error': str(e), 'status': 'error'})}\n\n"
        return

    n_rewrites = len(result.get("rewrites", []))
    yield f"data: {json.dumps({'thinking': {'category': 'planning', 'content': f'Generated {n_rewrites} query rewrites', 'timestamp': datetime.now().isoformat()}})}\n\n"
    yield f"data: {json.dumps({'rewrites': result.get('rewrites', [])})}\n\n"
    yield f"data: {json.dumps({'sources': result.get('sources', [])})}\n\n"
    yield f"data: {json.dumps({'content': result.get('answer', '')})}\n\n"

    if result.get("eval_score"):
        yield f"data: {json.dumps({'eval_score': result['eval_score']})}\n\n"

    yield f"data: {json.dumps({'done': True, 'status': 'completed'})}\n\n"

Each yield is one SSE event. Each payload is a JSON object with a recognizable key. The client switches on the key to decide which UI region to update.

router.py
python
@router.post("/ask/stream")
async def ask_stream_endpoint(request: AskRequest):
    """SSE endpoint streaming pipeline stages + final answer + eval score."""
    return StreamingResponse(
        ask_stream(request),
        media_type="text/event-stream",
        headers={"Cache-Control": "no-cache", "Connection": "keep-alive"},
    )

FastAPI StreamingResponse with media_type text/event-stream is all you need. The Cache-Control and Connection headers prevent proxies from buffering the stream and cutting off events.