Waiting 8 to 15 seconds for an LLM to generate an exhaustive JSON object creates terrible user experience. In modern web applications, users demand instant progressive feedback—seeing sections, bullet points, and tables render smoothly line-by-line.
Here is the battle-tested blueprint for streaming partially validated structured objects from a FastAPI backend to a React frontend using Server-Sent Events (SSE) and PydanticAI.
1. Architectural Pipeline
┌──────────────┐ HTTP POST /stream ┌──────────────┐
│ React Client │ ──────────────────────────────────> │ FastAPI App │
│ │ <────────────────────────────────── │ │
└──────────────┘ text/event-stream (SSE) └──────┬───────┘
│ run_stream()
┌──────▼───────┐
│ PydanticAI │
│ Agent Loop │
└──────────────┘
2. FastAPI Backend Implementation
import asyncio
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from pydantic import BaseModel, Field
from pydantic_ai import Agent
app = FastAPI(title="PydanticAI SSE Streamer")
class MarketAnalysis(BaseModel):
summary: str = Field(description="Executive summary")
bull_points: list[str] = Field(description="Bullish catalysts")
bear_points: list[str] = Field(description="Downside risks")
sentiment_score: int = Field(description="0 to 100 sentiment rating")
agent = Agent("google-gla:gemini-2.5-flash", result_type=MarketAnalysis)
@app.get("/api/analyze-stream")
async def stream_analysis(ticker: str):
async def event_generator():
prompt = f"Analyze current technical and fundamental setup for {ticker}."
async with agent.run_stream(prompt) as result:
async for partial_model in result.stream_structured():
# Stream partial JSON snapshot as SSE payload
yield f"data: {partial_model.model_dump_json()}
"
await asyncio.sleep(0.02)
yield "event: done
data: {}
"
return StreamingResponse(event_generator(), media_type="text/event-stream")
3. React Frontend Consumption Hook
import React, { useState, useEffect } from 'react';
export function useStreamingAnalysis(ticker: string) {
const [data, setData] = useState<any>(null);
const [isStreaming, setIsStreaming] = useState(false);
const startStream = () => {
setIsStreaming(true);
const eventSource = new EventSource(`/api/analyze-stream?ticker=${ticker}`);
eventSource.onmessage = (event) => {
try {
const parsed = JSON.parse(event.data);
setData(parsed);
} catch (err) {
console.error("JSON parse error on chunk", err);
}
};
eventSource.addEventListener("done", () => {
eventSource.close();
setIsStreaming(false);
});
eventSource.onerror = () => {
eventSource.close();
setIsStreaming(false);
};
};
return { data, isStreaming, startStream };
}
4. Key Takeaways
- Zero UI Blocking: Users see the headline and first bullet point within 400ms rather than waiting for the entire response to finish.
- Type-Safe Contract: Frontend and backend share exact TypeScript/Pydantic schemas, eliminating runtime parsing exceptions.
- HTTP/2 Efficiency: Multiple concurrent streams multiplex seamlessly across a single TCP connection.
FREE CODE TEMPLATE
Download the Complete PydanticAI Document Parser Blueprint
Get the complete, type-safe invoice and ID card parsing codebase in Python + a ready-to-run Docker environment. 100% free.
