Implementing Streamable HTTP with SSE for Real‑Time Chat in FastAPI
This article walks through adding a POST /v1/chat/stream endpoint that uses Server‑Sent Events (SSE) to stream structured events—including status, tool calls, token deltas, and completion—into a static web page, enabling a multi‑turn, typewriter‑style chat UI while preserving the original synchronous /v1/chat API for backward compatibility.
Goal and Acceptance Criteria
The objective is to expose a POST /v1/chat/stream endpoint that returns text/event-stream frames, while keeping the existing POST /v1/chat JSON API unchanged. The front‑end must display a multi‑turn conversation with a typewriter effect, show tool‑execution status, persist session_id in localStorage, and remain functional when the page is refreshed.
Architecture Overview
The system consists of three layers:
Orchestration layer : TurnEventEmitter emits structured events (run_start, status, tool_start, tool_end, token, done, error) without being tied to SSE.
Service layer : chat_stream() creates a session, yields a run_start frame, runs the ReAct loop, and drains events from the emitter, finally yielding a done frame.
Transport layer : FastAPI route /v1/chat/stream returns a StreamingResponse with media_type="text/event-stream" and headers Cache-Control: no-cache, Connection: keep-alive, and X-Accel-Buffering: no to prevent Nginx buffering.
Server‑Side Implementation
Event framing utilities (api/stream.py) :
def sse_frame(event: str, data: dict) -> str:
payload = json.dumps(data, ensure_ascii=False)
return f"event: {event}
data: {payload}
"
def sse_comment(text: str = "keep-alive") -> str:
return f": {text}
"FastAPI route (api/routes.py) :
@router.post("/v1/chat/stream", tags=["chat"])
async def chat_stream(body: ChatRequest, service: AgentService = Depends(...)):
session_id = body.session_id or uuid.uuid4().hex[:12]
async def event_generator():
try:
async for frame in service.chat_stream(body.message.strip(), session_id=session_id, user_id=body.user_id or service.settings.user_id):
yield frame
except AgentError as exc:
yield sse_frame("error", {"message": str(exc)})
return StreamingResponse(event_generator(), media_type="text/event-stream", headers={
"Cache-Control": "no-cache",
"Connection": "keep-alive",
"X-Accel-Buffering": "no",
})Service logic (api/service.py) demonstrates the event flow:
def chat_stream(self, message, *, session_id, user_id):
record = self.session_store.get_or_create(session_id, user_id)
yield sse_frame("run_start", {"session_id": session_id, "trace_id": app.pending_trace_id, "agent_version": self.settings.agent_version})
emitter = TurnEventEmitter()
result = app.turn_stream(message, emitter=emitter)
for ev in emitter.drain():
yield sse_frame(ev.type, ev.data)
yield sse_frame("done", {"reply": result.reply, "trace_id": result.trace_id, "usage": result.usage, "steps": result.steps, "execution_mode": result.execution_mode})
self._persist_session(app, record)Static Front‑End
The web/ directory contains a plain HTML page, CSS, and a single JavaScript bundle. No build tools are required.
index.html provides the chat container, a status bar, and a composer textarea.
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>mini‑agent</title>
<link rel="stylesheet" href="/css/app.css">
</head>
<body>
<div class="app">
<header class="app-header"><h1>mini‑agent</h1><span id="session-label" class="muted"></span></header>
<main id="messages" class="messages" aria-live="polite"></main>
<footer class="composer">
<textarea id="input" rows="2" placeholder="Enter message, press Enter…"></textarea>
<button id="send" type="button">Send</button>
</footer>
</div>
<script src="/js/app.js"></script>
</body>
</html>app.js implements the fetch‑SSE loop, parses SSE blocks, updates the UI, and drives the typewriter queue.
const SESSION_KEY = "mini_agent_session_id";
let sessionId = localStorage.getItem(SESSION_KEY) || null;
let typingQueue = [];
let typingTimer = null;
let abortController = null;
async function sendMessage(text) {
appendUserBubble(text);
setComposerEnabled(false);
const agentEl = appendAgentBubble("");
showStatus("Connecting…");
abortController = new AbortController();
const headers = {"Content-Type": "application/json"};
const apiKey = localStorage.getItem("mini_agent_api_key");
if (apiKey) headers["X-API-Key"] = apiKey;
const res = await fetch("/v1/chat/stream", {
method: "POST",
headers,
body: JSON.stringify({message: text, session_id: sessionId}),
signal: abortController.signal,
});
if (!res.ok) { showError(await res.text()); setComposerEnabled(true); return; }
const reader = res.body.getReader();
const decoder = new TextDecoder();
let buffer = "";
while (true) {
const {done, value} = await reader.read();
if (done) break;
buffer += decoder.decode(value, {stream: true});
const parts = buffer.split("
");
buffer = parts.pop() || "";
for (const block of parts) {
const ev = parseSseBlock(block);
if (!ev) continue;
handleEvent(ev, agentEl, delta => { fullReply += delta; enqueueTyping(agentEl, delta); });
}
}
hideStatus();
setComposerEnabled(true);
}
function parseSseBlock(block) {
let event = "message";
let data = "";
for (const line of block.split("
")) {
if (line.startsWith("event:")) event = line.slice(6).trim();
if (line.startsWith("data:")) data += line.slice(5).trim();
}
if (!data) return null;
return {event, data: JSON.parse(data)};
}
function handleEvent({event, data}, agentEl, onToken) {
switch (event) {
case "run_start":
sessionId = data.session_id;
localStorage.setItem(SESSION_KEY, sessionId);
updateSessionLabel(sessionId, data.trace_id);
break;
case "status":
showStatus(data.message);
break;
case "tool_start":
showStatus(`Calling tool: ${data.name}…`);
break;
case "tool_end":
showStatus(data.ok ? `Tool ${data.name} completed` : `Tool ${data.name} failed`);
break;
case "token":
onToken(data.delta);
break;
case "done":
hideStatus();
break;
case "error":
showError(data.message);
break;
}
}
function enqueueTyping(el, text) {
for (const ch of text) typingQueue.push(ch);
if (!typingTimer) typingTimer = setInterval(flushTyping, 28);
}
function flushTyping() {
const el = document.querySelector(".msg.agent:last-child .content");
if (!el || typingQueue.length === 0) { clearInterval(typingTimer); typingTimer = null; return; }
el.textContent += typingQueue.shift();
scrollToBottom();
}The CSS defines message bubbles, a status bar, disabled composer state, and a mobile‑first flex layout.
Configuration
config/agent.yamladds a stream section with enabled, typing_fallback, pseudo_chunk_chars, and keepalive_seconds.
Environment variable STREAM_ENABLED toggles registration of the streaming route. HTTP_API_KEY (optional) is injected as X-API-Key header when present.
Running the Demo
Clone the repository and activate the virtual environment.
cd mini-agent
source .venv/bin/activate
pip install -r requirements.txt
python main.py serveOpen http://127.0.0.1:8000/ in a browser.
Ask "What time is it?" – see a status flash followed by a typewriter‑style answer.
Ask "Shanghai weather?" – see a tool‑start/status line, then the answer.
Refresh the page and ask "Which city did you just check?" – the conversation continues using the persisted session_id.
Verify the SSE frames with
curl -N -X POST http://127.0.0.1:8000/v1/chat/stream -H "Content-Type: application/json" -d '{"message": "Introduce yourself in one sentence"}'. The output should contain sequential event: and data: blocks for run_start, status, token, and done.
Run the evaluation script to ensure the original synchronous API still passes all golden tests:
python main.py evalCommon Pitfalls and Fixes
Static page 404 : Ensure WEB_DIR points to the repository root web/ folder and that API routes are mounted before the static StaticFiles mount.
SSE frames buffered by reverse proxy : Add X-Accel-Buffering: no header and configure Nginx with proxy_buffering off or bypass the proxy during development.
Typewriter glitches : Cancel any previous request with abortController.abort() before sending a new one and ensure only a single setInterval timer runs.
CORS errors when opening the HTML file directly : Always access the page via http://127.0.0.1:8000/ so the browser and FastAPI share the same origin.
Tool calls appearing as partial token streams : The brain layer accumulates tool_calls and only yields token events for the final natural‑language answer, matching the design in Module 12.
Next Steps
The upcoming article (Advance 01) will show how to mount an MCP server on the same FastAPI instance, allowing the agent to invoke external tools such as a filesystem service via the MCP protocol.
Conclusion
The new web/ chat page plus the POST /v1/chat/stream endpoint turns the agent from a raw API into an interactive demo that can be opened directly in a browser.
Structured SSE events (status, tool, token, done) implement the module‑12 contract, while the front‑end typewriter queue provides a smooth UX and supports a pseudo‑stream fallback when the LLM SDK cannot stream.
Only the streaming path is new; the original synchronous /v1/chat endpoint remains unchanged, preserving existing CLI and evaluation workflows.
Signed-in readers can open the original source through BestHub's protected redirect.
This article has been distilled and summarized from source material, then republished for learning and reference. If you believe it infringes your rights, please contactand we will review it promptly.
How this landed with the community
Was this worth your time?
0 Comments
Thoughtful readers leave field notes, pushback, and hard-won operational detail here.