Until now the only way to talk to the bot was a terminal. This chapter gives it a public face: a web page. That sounds like a front-end job, and it is partly one, but the interesting problems are not visual. A chat page needs the reply to appear
as it is written, which means streaming; it must keep each visitor's conversation separate without logging anyone in, which means a session cookie; and because it is a web page that sends a cookie, it must resist requests forged by other
sites. And it has to work for people who cannot see the screen. We build all of it with a few hundred lines and test it in a real browser.
What was run, and what was not
The server was run for real with uvicorn 0.53.0 and FastAPI 0.141.1 (Starlette 1.6.0), and the page was driven in real headless Chrome through the DevTools Protocol: typing, pressing keys, clicking, reloading and emulating screen sizes and colour schemes.
Python 3.14.0; the project now has 175 passing tests. The replies came from the FakeProvider, so no real model, and no real Claude or OpenAI stream, was involved. The two vendor streaming parsers are tested against a local server that sends
the documented formats (Anthropic's from its streaming documentation, which I read; OpenAI's is the common Chat Completions format, whose reference page I could not fetch). Not tested: a screen reader, Firefox or Safari, HTTPS, HTTP/2, a reverse proxy, or real network delay.
Streaming: Providers Now Yield Pieces
People read as the words arrive; waiting for a whole reply feels slow even when it is not. Every provider therefore gains a second method next to complete(): stream(), an async generator that yields the text in pieces as they arrive
and ends with one final Reply (the full text, token usage and so on). Code that does not care about streaming keeps using complete(); the metering and guards from earlier chapters need the final Reply, which is why it is
always last.
The real work is parsing each vendor's stream. Both use Server-Sent Events (SSE): a long HTTP response whose body is lines of event: and data:, separated by blank lines. A small parser, shared by both, turns the body into (event, data) pairs:
import json
from typing import AsyncIterator
import httpx
from bot.providers.base import (
AuthError, BadRequest, Message, Provider, ProviderError, ProviderTimeout, ProviderUnavailable,
QuotaExhausted, RateLimited, Reply, ServerError, error_from_status, parse_retry_after,
)
StreamItem = str | Reply # text pieces as they arrive, then ONE final Reply carrying the whole text and the usage
def error_from_http(status: int, text: str, headers) -> ProviderError:
"""Turn a non-200 response (already read) into one of our typed errors."""
message, code = text, None
try:
error = json.loads(text)["error"]
message = error["message"]
code = (error.get("details") or {}).get("error_code")
except (ValueError, KeyError, TypeError):
pass
request_id = headers.get("request-id") or headers.get("x-request-id")
if code == "enforced_spend_limit_reached":
return QuotaExhausted(message, status=status, request_id=request_id)
return error_from_status(status, message, retry_after=parse_retry_after(headers.get("retry-after")), request_id=request_id)
def error_from_stream_event(error_type: str, message: str) -> ProviderError:
"""An error that arrives INSIDE a stream that already returned 200 (so there is no HTTP status to look at)."""
kind = {"overloaded_error": (ServerError, 529), "rate_limit_error": (RateLimited, 429),
"authentication_error": (AuthError, 401), "invalid_request_error": (BadRequest, 400)}.get(error_type, (ServerError, 500))
return kind[0](message, status=kind[1])
def wrap_transport_errors(e: httpx.HTTPError) -> ProviderError:
if isinstance(e, httpx.ConnectTimeout):
return ProviderUnavailable(f"could not connect: {e!r}")
if isinstance(e, httpx.TimeoutException):
return ProviderTimeout(f"no answer in time: {e!r}")
return ProviderUnavailable(f"connection problem: {e!r}")
async def sse_events(response: httpx.Response) -> AsyncIterator[tuple[str | None, str]]:
"""Parse a Server-Sent Events body into (event name, data) pairs. Comment lines (starting ':') are skipped."""
event, data = None, []
async for line in response.aiter_lines():
if line == "":
if data:
yield event, "\n".join(data)
event, data = None, []
continue
if line.startswith(":"):
continue
field, _, value = line.partition(":")
value = value[1:] if value.startswith(" ") else value
if field == "event":
event = value
elif field == "data":
data.append(value)
if data:
yield event, "\n".join(data)
async def stream_via_complete(provider: Provider, messages: list[Message], *, system: str | None = None,
max_tokens: int = 1024) -> AsyncIterator[StreamItem]:
"""For a provider with no native streaming: one piece of text (the whole reply), then the Reply."""
reply = await provider.complete(messages, system=system, max_tokens=max_tokens)
yield reply.text
yield reply
Anthropic's flow, from its streaming documentation: a message_start event (with the model and the input token count), then content_block_delta events whose text_delta carries the text, then message_delta with the stop reason and the
cumulative output tokens (the docs warn the counts are cumulative, not per event), then message_stop. ping events can appear anywhere and are ignored. And an error event can arrive after the 200 status: the docs give
overloaded_error as the example. That last point is the one that catches people out, because there is no HTTP status to check; the code below converts such an event into the same typed errors as before. The OpenAI-style stream is simpler: data: lines
carrying choices[0].delta.content, ending with data: [DONE], with a last chunk holding usage if you ask for it with stream_options.include_usage (Ollama's documentation lists that option as supported).
async def stream(self, messages: list[Message], *, system: str | None = None, max_tokens: int = 1024):
"""Stream the reply with Server-Sent Events: text pieces as they arrive, then one final Reply.
Event flow (from Anthropic's streaming docs): message_start (model, input tokens), then content_block_delta
events whose text_delta carries the text, message_delta (stop reason and the CUMULATIVE output tokens),
message_stop. ping events are ignored, and an `error` event can arrive after the 200 status."""
body: dict = {"model": self.model, "max_tokens": max_tokens, "messages": messages, "stream": True}
if system:
body["system"] = system
pieces: list[str] = []
model, input_tokens, output_tokens, stop_reason, finished, request_id = self.model, 0, 0, "", False, None
try:
async with self._client.stream("POST", "/v1/messages", json=body) as response:
request_id = response.headers.get("request-id")
if response.status_code != 200:
await response.aread()
raise error_from_http(response.status_code, response.text, response.headers)
async for _event, data in sse_events(response):
obj = json.loads(data)
kind = obj.get("type")
if kind == "message_start":
message = obj["message"]
model = message.get("model", model)
input_tokens = message.get("usage", {}).get("input_tokens", 0)
output_tokens = message.get("usage", {}).get("output_tokens", 0)
elif kind == "content_block_delta" and obj["delta"].get("type") == "text_delta":
pieces.append(obj["delta"]["text"])
yield obj["delta"]["text"]
elif kind == "message_delta":
stop_reason = obj.get("delta", {}).get("stop_reason") or stop_reason
output_tokens = obj.get("usage", {}).get("output_tokens", output_tokens)
elif kind == "message_stop":
finished = True
elif kind == "error":
raise error_from_stream_event(obj["error"].get("type", ""), obj["error"].get("message", ""))
except httpx.HTTPError as e:
raise wrap_transport_errors(e) from e
if not finished:
raise ServerError("the stream ended before message_stop")
yield Reply("".join(pieces), Usage(input_tokens, output_tokens), model=model, stop_reason=stop_reason, request_id=request_id)
Two rules keep a stream honest. A stream that ends without its terminator (message_stop, or [DONE]) is an error, not a short reply, because a dropped connection looks exactly like a normal end otherwise. And the reliability wrappers from chapter 7 learn
a new rule: retry or fall back only if the stream failed before producing any text. Once words have reached the user, trying again would repeat them or contradict them. The tests prove each of these against a local server that sends the formats above (15 streaming tests).
The Bot Streams Events
Bot.stream_message() shares everything with handle(): the same pre-checks (input guard, commands, mute, moderation, limits), the same window and system prompt, the same metering, storing and final checks. It yields small dictionaries:
Event
Meaning
{"type": "delta", "text": ...}
A piece of the reply (only in live mode)
{"type": "final", "text": ..., "changed": ...}
The reply after every check, which is what was stored. changed says it differs from what was streamed (a guard replaced it, or the AI disclosure was added). The page always shows the final text.
{"type": "error", "message": ..., "partial": ...}
The stream failed. partial says the user already saw some text. Nothing was stored.
async def stream_message(self, session_id: str, text: str, *, user_id: str | None = None):
"""Answer one message as a stream of events for a live channel such as the web page.
Events are dicts: {"type": "delta", "text": piece} (only when stream_live is on), then exactly one of
{"type": "final", "text": the checked reply, "changed": whether it differs from what was streamed}
or {"type": "error", "message": ..., "partial": whether the user already saw some text}.
Nothing is stored unless the whole reply arrived and passed the checks."""
user_id = user_id or session_id
text, early = await self._preflight(session_id, user_id, text)
if early is not None:
yield {"type": "final", "text": early, "changed": False}
return
lock = self._locks.setdefault(session_id, asyncio.Lock())
async with lock:
window, system, rows, summary = await self._prepare(session_id, user_id, text)
shown = ""
reply = None
try:
async for item in self.provider.stream(window, system=system, max_tokens=self.max_tokens):
if isinstance(item, str):
shown += item
if self.stream_live:
yield {"type": "delta", "text": item}
else:
reply = item
except ProviderError as e:
self.events.append((session_id, "stream_error:" + type(e).__name__))
yield {"type": "error", "message": self.error_reply, "partial": bool(self.stream_live and shown)}
return
answer = await self._finish(session_id, user_id, text, reply, first=not rows and summary is None)
yield {"type": "final", "text": answer, "changed": answer != shown}
The Trade-Off You Cannot Avoid: Live or Buffered
The output guard (chapter 5) and output moderation (chapter 6) need the whole reply, but streaming shows the reply before it is whole. OpenAI's moderation guide says exactly this about its own inline moderation: with streaming, results arrive only after the full output is generated. So there are two modes, chosen with
stream_live. Live streams every piece at once and, if a check fails at the end, sends a final event that replaces what was shown; the user briefly saw the unchecked text. Buffered holds everything until the checks pass and then sends one event; nothing unchecked is ever shown, but the user waits for the whole reply. Real events from both:
--- normal reply (live)
{"type": "delta", "text": "We "}
{"type": "delta", "text": "open "}
{"type": "delta", "text": "at "}
{"type": "delta", "text": "nine."}
{"type": "final", "text": "We open at nine.", "changed": false}
--- a leak, LIVE streaming: the user's screen sees it first
{"type": "delta", "text": "Sure! "}
{"type": "delta", "text": "My "}
{"type": "delta", "text": "marker "}
{"type": "delta", "text": "is "}
{"type": "delta", "text": "canary-XXXXXXXXXXXX "}
{"type": "delta", "text": "ok"}
{"type": "final", "text": "Sorry, I can't share that.", "changed": true}
--- the same leak, BUFFERED: it never reaches the screen
{"type": "final", "text": "Sorry, I can't share that.", "changed": true}
--- failure half-way (live)
{"type": "delta", "text": "We "}
{"type": "delta", "text": "open "}
{"type": "delta", "text": "at"}
{"type": "error", "message": "Sorry, something went wrong. Please try again.", "partial": true}
--- a command arrives as a single final event
{"type": "final", "text": "Remembered as #1.", "changed": false}
In the live run the leaked marker (shown as canary-XXXXXXXXXXXX) did appear on the user's screen before being replaced by the safe text; the stored history never contains it. In the buffered run it never left the server. The cost of buffering is time. I measured it with a fake provider that takes
0.05 seconds per word (Exercise 2):
mode first text on screen reply complete events
live 0.06s 1.24s 21
buffered 1.25s 1.25s 1
Live shows the first word after 0.06 s and the whole reply after about 1.2 s. Buffered shows nothing until 1.25 s. Choose live for a low-stakes bot where speed matters, and buffered where an unchecked sentence on a screen is unacceptable (a bot for children, for example). A stream that fails half-way shows the same difference: live reports partial: true, buffered always reports
false because the user saw nothing.
Why POST and fetch, Not EventSource
The browser has a built-in SSE client, EventSource, but it only makes GET requests with no body; a chat message belongs in a POST body, not in a URL that ends up in logs and browser history (this is from general knowledge: the MDN page I fetched did not state it). So the page uses fetch() and reads the response body as a stream, which also lets
it cancel with AbortController for the Stop button. One real constraint from MDN's EventSource page applies to any SSE: over HTTP/1.1 a browser allows only about 6 open connections per domain (shared across tabs) and the page says this is "Won't fix" in Chrome and Firefox; with HTTP/2 the limit is negotiated and defaults to 100. A chat that keeps a stream open per reply
is fine for one tab and worth knowing about for many.
The Server: A Session, a Stream and a Locked Door
The web channel is a small FastAPI app. It serves the page, keeps each visitor's session in a cookie, streams replies on POST /api/chat, and refuses anything it cannot trust. Read the whole file once, then the notes below it.
import hashlib
import hmac
import json
import secrets
from pathlib import Path
from fastapi import FastAPI, Request
from fastapi.responses import FileResponse, JSONResponse, Response, StreamingResponse
from bot.core import Bot
STATIC = Path(__file__).parent / "static"
COOKIE = "chat_sid"
PAGE_FILES = {"chat.js": "text/javascript", "chat.css": "text/css"}
CSP = "default-src 'self'; base-uri 'none'; form-action 'self'; frame-ancestors 'none'"
class SessionCookie:
"""A random session id signed with HMAC, so the browser can hold it but not forge someone else's."""
def __init__(self, secret: bytes) -> None:
self.secret = secret
def _sign(self, sid: str) -> str:
return hmac.new(self.secret, sid.encode(), hashlib.sha256).hexdigest()[:32]
def issue(self) -> tuple[str, str]:
sid = secrets.token_urlsafe(16)
return sid, f"{sid}.{self._sign(sid)}"
def read(self, value: str | None) -> str | None:
if not value or "." not in value:
return None
sid, _, signature = value.rpartition(".")
return sid if hmac.compare_digest(signature, self._sign(sid)) else None
def create_app(bot: Bot, *, secret: bytes, allowed_origins: set[str], secure_cookie: bool = False,
max_body: int = 16 * 1024, cookie_days: int = 30) -> FastAPI:
"""The web channel: a chat page, a streaming endpoint, and the session cookie that ties them together.
No CORS middleware is installed on purpose: the page and the API share one origin, so browsers
refuse cross-origin calls and there is nothing to configure. allowed_origins is used for the
Origin check on POST (a CSRF defence), e.g. {"http://localhost:8000"}."""
app = FastAPI(docs_url=None, redoc_url=None, openapi_url=None) # no public API documentation page
cookies = SessionCookie(secret)
@app.middleware("http")
async def security_headers(request: Request, call_next):
response = await call_next(request)
response.headers["Content-Security-Policy"] = CSP
response.headers["X-Content-Type-Options"] = "nosniff"
response.headers["Referrer-Policy"] = "no-referrer"
return response
def current_sid(request: Request) -> str | None:
return cookies.read(request.cookies.get(COOKIE))
def set_cookie(response: Response, value: str) -> None:
response.set_cookie(COOKIE, value, max_age=cookie_days * 86400, httponly=True, samesite="lax",
secure=secure_cookie, path="/")
@app.get("/")
async def page(request: Request):
response = FileResponse(STATIC / "index.html", media_type="text/html")
if current_sid(request) is None:
set_cookie(response, cookies.issue()[1])
return response
@app.get("/static/{name}")
async def static_file(name: str):
if name not in PAGE_FILES: # an allow-list, not a directory listing
return JSONResponse({"detail": "not found"}, status_code=404)
return FileResponse(STATIC / name, media_type=PAGE_FILES[name])
@app.get("/healthz")
async def health():
return {"status": "ok"}
@app.get("/api/history")
async def history(request: Request):
sid = current_sid(request)
if sid is None:
return JSONResponse({"detail": "no session"}, status_code=401)
return {"messages": [m for _, m in bot.store.messages(f"web:{sid}")]}
@app.post("/api/chat")
async def chat(request: Request):
sid = current_sid(request)
if sid is None:
return JSONResponse({"detail": "no session: load the page first"}, status_code=401)
# OWASP: Sec-Fetch-Site is the primary signal, Origin the fallback, and if neither shows a trusted source, refuse.
origin, fetch_site = request.headers.get("origin"), request.headers.get("sec-fetch-site")
trusted = (fetch_site == "same-origin") if fetch_site is not None else (origin in allowed_origins)
if not trusted:
return JSONResponse({"detail": "cross-origin request refused"}, status_code=403)
if not request.headers.get("content-type", "").startswith("application/json"):
return JSONResponse({"detail": "send application/json"}, status_code=415)
raw = await request.body()
if len(raw) > max_body:
return JSONResponse({"detail": "request too large"}, status_code=413)
try:
message = json.loads(raw)["message"]
except (ValueError, KeyError, TypeError):
return JSONResponse({"detail": "expected a JSON object with a message"}, status_code=400)
if not isinstance(message, str):
return JSONResponse({"detail": "message must be text"}, status_code=400)
async def events():
async for event in bot.stream_message(f"web:{sid}", message, user_id=f"web:{sid}"):
yield f"data: {json.dumps(event)}\n\n"
return StreamingResponse(events(), media_type="text/event-stream",
headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"})
return app
The Session Cookie
A visitor is identified by a random id in a cookie, signed with an HMAC so the browser cannot invent someone else's id. The cookie is HttpOnly (JavaScript cannot read it: in the browser test document.cookie was an empty string while the chat worked, which is the point),
SameSite=Lax, Max-Age 30 days to match the retention period (Secure is added when the site is on HTTPS). It names an anonymous session, not a person, which is fine for a public bot and is not a login. The signing secret must be long, random and kept out of the code; the demo uses a throwaway.
CSRF: A Page on Another Site Must Not Be Able to Talk as the Visitor
Browsers attach cookies to requests automatically, so a malicious page could make a visitor's browser send a message to your bot. OWASP's CSRF guidance, which I read, says: check the request's origin with the standard headers, treat Sec-Fetch-Site as the primary signal with Origin as the fallback, and
if neither shows a trusted source, refuse; SameSite cookies are defence in depth, not a replacement; and blocking the "simple" content types (form, text/plain) is a useful mitigation but not sufficient on its own. The endpoint does all of these: it requires Sec-Fetch-Site: same-origin or an Origin on its list,
requires application/json, and the cookie is SameSite=Lax. Real requests against the running server:
$ curl -c jar.txt http://127.0.0.1:8765/ -o /dev/null (stores the session cookie in jar.txt)
$ curl -b jar.txt -X POST http://127.0.0.1:8765/api/chat -H 'Content-Type: application/json' -d '{"message":"hi"}'
{"detail":"cross-origin request refused"} HTTP 403 (neither Origin nor Sec-Fetch-Site: refuse)
$ curl -b jar.txt -X POST ... -H 'Content-Type: application/json' -H 'Origin: http://evil.example' -d '{"message":"hi"}'
{"detail":"cross-origin request refused"} HTTP 403
$ curl -b jar.txt -X POST ... -H 'Content-Type: text/plain' -H 'Origin: http://127.0.0.1:8765' -d 'message=hi'
{"detail":"send application/json"} HTTP 415
$ curl -b jar.txt -N -X POST ... -H 'Content-Type: application/json' -H 'Origin: http://127.0.0.1:8765' -d '{"message":"/facts"}'
data: {"type": "final", "text": "I am not remembering anything about you.", "changed": false} HTTP 200
A caveat that OWASP states and I repeat: these defences protect against forged requests. They do nothing against script injection (XSS), which is why the page never inserts reply text as HTML, as shown below. A bot's replies are text from a model that anyone can influence, so treat them as hostile input.
CORS: Deliberately Nothing
The page and the API are on the same origin, so no Access-Control-* headers are needed, and I installed no CORS middleware. Browsers then refuse other sites' scripts from reading responses. The tests check that the headers are absent and that a preflight request gets a 405. If you later serve the page from a different origin, add CORS with an explicit origin list (never a wildcard when cookies are involved) and
think again about CSRF.
The Rest of the Hardening
Content-Security-Policydefault-src 'self'; base-uri 'none'; form-action 'self'; frame-ancestors 'none' on every response: the page may load only its own files (this is why the JavaScript is a separate file, not inline), may not be framed by another site, and cannot be tricked with a <base> tag. Plus nosniff and Referrer-Policy: no-referrer.
Static files by allow-list: only chat.js and chat.css are served; index.html, path tricks and other names get a 404 (tested).
No public API docs: FastAPI's /docs and /openapi.json are switched off (tested: 404).
Request size: a body over 16 KB gets a 413. The real limit belongs at the reverse proxy too (chapter 12).
The Page
Three small files, with no framework and no build step. First the HTML:
<!doctype html>
<html lang="en">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>Chat</title>
<link rel="stylesheet" href="/static/chat.css">
</head>
<body>
<main class="chat">
<h1>Chat with Rowan</h1>
<p class="note">You are talking to an AI assistant, not a person. Type /privacy to see what is stored.</p>
<!-- role="log" is a polite live region: new messages are announced; a message that is still
streaming is marked aria-busy so it is announced once it is complete, not word by word. -->
<div id="log" class="log" role="log" aria-label="Conversation" tabindex="0"></div>
<p id="status" class="status" role="status"></p>
<form id="form" class="form" autocomplete="off">
<label for="input" class="visually-hidden">Your message</label>
<textarea id="input" rows="2" maxlength="2000" placeholder="Type a message. Enter sends, Shift+Enter adds a line." required></textarea>
<div class="buttons">
<button type="submit" id="send">Send</button>
<button type="button" id="stop" hidden>Stop</button>
<button type="button" id="retry" hidden>Try again</button>
</div>
</form>
</main>
<script src="/static/chat.js"></script>
</body>
</html>
Notes on the markup: one <h1> and one <main> landmark; the textarea has a real <label> (visually hidden, still announced); the conversation is role="log" (a polite live region, so new messages are announced) with a label and tabindex="0" so a keyboard user can scroll it;
role="status" for short announcements; the AI disclosure is visible text, not a tooltip. Now the JavaScript, the part with the real logic:
"use strict";
const log = document.getElementById("log");
const form = document.getElementById("form");
const input = document.getElementById("input");
const sendButton = document.getElementById("send");
const stopButton = document.getElementById("stop");
const retryButton = document.getElementById("retry");
const statusLine = document.getElementById("status");
let controller = null; // lets the Stop button cancel the request in flight
let lastMessage = null; // what "Try again" sends
// Every piece of text is put in with textContent, never innerHTML, so a reply containing <script> or <img onerror=...>
// is shown as plain characters and cannot run.
function addMessage(role, text) {
const wrapper = document.createElement("div");
wrapper.className = "msg " + role;
const who = document.createElement("span");
who.className = "who";
who.textContent = role === "user" ? "You" : "Assistant";
const body = document.createElement("p");
body.className = "text";
body.textContent = text;
wrapper.append(who, body);
log.append(wrapper);
scrollDown();
return { wrapper, body };
}
function scrollDown() {
log.scrollTop = log.scrollHeight;
}
function setBusy(busy) {
sendButton.disabled = busy;
stopButton.hidden = !busy;
if (busy) retryButton.hidden = true;
}
function say(text) {
statusLine.textContent = text; // read out by assistive technology (role="status")
}
async function send(text) {
lastMessage = text;
setBusy(true);
say("");
const bot = addMessage("bot", "");
bot.wrapper.setAttribute("aria-busy", "true");
bot.wrapper.classList.add("typing");
controller = new AbortController();
let finished = false;
try {
const response = await fetch("/api/chat", {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({ message: text }),
signal: controller.signal,
});
if (!response.ok) {
const problem = await response.json().catch(() => ({}));
throw new Error(problem.detail || "The server answered " + response.status);
}
const reader = response.body.getReader();
const decoder = new TextDecoder();
let buffer = "";
let shown = "";
for (;;) {
const { value, done } = await reader.read();
if (done) break;
buffer += decoder.decode(value, { stream: true });
let end;
while ((end = buffer.indexOf("\n\n")) !== -1) {
const block = buffer.slice(0, end);
buffer = buffer.slice(end + 2);
const data = block.split("\n").filter((l) => l.startsWith("data: ")).map((l) => l.slice(6)).join("\n");
if (!data) continue;
const event = JSON.parse(data);
if (event.type === "delta") {
shown += event.text;
bot.body.textContent = shown;
bot.wrapper.classList.remove("typing");
scrollDown();
} else if (event.type === "final") {
bot.body.textContent = event.text; // the checked text always wins over what was streamed
finished = true;
} else if (event.type === "error") {
bot.wrapper.classList.add("failed");
if (!event.partial) bot.body.textContent = "";
const alert = document.createElement("p");
alert.className = "problem";
alert.setAttribute("role", "alert");
alert.textContent = event.message + (event.partial ? " (The reply above was cut off.)" : "");
bot.wrapper.append(alert);
retryButton.hidden = false;
finished = true;
}
}
}
if (!finished) throw new Error("The connection closed before the reply was complete.");
} catch (error) {
if (error.name === "AbortError") {
bot.wrapper.classList.add("stopped");
const note = document.createElement("p");
note.className = "problem";
note.textContent = "Stopped.";
bot.wrapper.append(note);
retryButton.hidden = false;
} else {
bot.wrapper.classList.add("failed");
const alert = document.createElement("p");
alert.className = "problem";
alert.setAttribute("role", "alert");
alert.textContent = error.message;
bot.wrapper.append(alert);
retryButton.hidden = false;
}
} finally {
bot.wrapper.removeAttribute("aria-busy");
bot.wrapper.classList.remove("typing");
controller = null;
setBusy(false);
input.focus();
}
}
form.addEventListener("submit", (event) => {
event.preventDefault();
const text = input.value.trim();
if (!text || controller) return;
addMessage("user", text);
input.value = "";
send(text);
});
input.addEventListener("keydown", (event) => {
if (event.key === "Enter" && !event.shiftKey && !event.isComposing) {
event.preventDefault();
form.requestSubmit();
}
});
stopButton.addEventListener("click", () => controller && controller.abort());
retryButton.addEventListener("click", () => {
if (lastMessage && !controller) send(lastMessage);
});
// Show the conversation again after a reload: the server keeps it, the page asks for it.
fetch("/api/history").then((r) => (r.ok ? r.json() : { messages: [] })).then((data) => {
for (const m of data.messages) addMessage(m.role === "user" ? "user" : "bot", m.content);
}).catch(() => {});
input.focus();
Things to look for in it:
textContent, never innerHTML. Every message and every error is inserted as plain text. A reply containing <img onerror=...> is displayed as those characters. (The tests below send exactly that.)
SSE by hand. It accumulates bytes, splits on a blank line, takes the data: lines and parses the JSON. The three event types map to the three behaviours above.
aria-busy="true" while a reply streams. Without it a screen reader may re-announce the growing message at every word; with it assistive technology can wait until the message is complete. I tested that the attribute is present during streaming and removed afterwards; I did not test any screen reader, so how a particular one behaves is unverified.
Focus returns to the text box after every reply, Stop, or error, so a keyboard user can carry on typing. Enter sends and Shift+Enter adds a line; isComposing avoids sending half-typed text in an input method (Japanese, for example).
Stop cancels the request.AbortController closes the connection; in the test nothing was stored and the server's session was free for the next message straight away (unit tests show that closing the stream cancels generation and releases the lock). Try again re-sends the last message after an error or a stop.
History on load. The page asks /api/history for the stored conversation, so a reload does not look like amnesia.
It follows the visitor's light or dark preference, uses a visible focus outline, makes buttons at least 44 pixels tall, wraps long words, and turns off the blinking "typing" dots when the visitor has asked for reduced motion.
Testing It in a Real Browser
Unit tests cover the server (14 tests with FastAPI's test client: cookie handling, headers, CSRF, size limits, isolation between visitors). They cannot say whether the page works. So I started the real server with a fake provider scripted to produce a normal reply, a hostile reply, a failure half-way and a long reply, and drove the page in real Chrome
through the DevTools Protocol. What happened, from the run:
Check
Result
Structure
role="log" with label "Conversation", role="status", label "Your message" on the textarea, lang="en", one h1, one main; focus starts in the text box
Cookie
document.cookie is an empty string: the HttpOnly session cookie is invisible to script but the chat works
Streaming
Sampling the DOM every 150 ms, the reply grew through 9 different lengths from 0 to 181 characters; while it grew, the message had aria-busy="true", Stop was visible and Send was disabled. Afterwards: no aria-busy left, Stop hidden, Send enabled, focus back in the text box, and the first reply began with the AI disclosure
Hostile reply
Reply text was <img src=x onerror="document.title='HACKED'"> and <b>bold?</b> and <script>…</script>. Result: the page title stayed "Chat", zero img, script or b elements in the log, and the characters displayed literally
Failure half-way
Partial text "The library opens at nine and" stayed visible; an alert read "Sorry, something went wrong. Please try again. (The reply above was cut off.)"; Try again appeared; the server's history still held only the 4 earlier messages (the failed exchange was not stored). Try again produced "Retry worked: we open at nine on weekdays." and history went to 6
Stop
Pressing Stop 700 ms into a 60-word reply left 68 characters on screen, a "Stopped." note, Send enabled, Try again shown; history stayed at 6 (not stored); the next message was answered at once, so the server's session lock had been released
Keyboard
Shift+Enter put a newline in the text box ("line one\nline two") and sent nothing
Reload
The page came back with all 8 stored messages, starting with the first user message
Small screen
At 390 px wide: no horizontal scrolling; the Send button measured 44 × 80 px
Contrast. I read the page's colour variables in the browser in each colour scheme and computed WCAG contrast ratios with the formula in the specification (a small function of my own). WCAG asks for at least 4.5:1 for normal text; the lowest ratio on the page was 5.97:
Pair
Light
Dark
Text on page
17.40
16.02
Text in assistant bubble
15.80
14.19
Text in your bubble
14.27
9.98
Muted note on page
6.90
8.64
Button text on button
6.70
7.84
Outline-style button text on page
6.70
7.74
Error text on assistant bubble
5.97
7.17
That checks the colours I chose, not the whole page. An automated check cannot replace trying the page with a screen reader, with the keyboard alone, and at 200% zoom.
Mistakes while building this
My first CSRF check let requests through when both headers were missing. I wrote it to allow a request with no Origin and no Sec-Fetch-Site, reasoning that "a browser always sends one". Then I read OWASP's cheat sheet, which says to refuse when neither header shows a trusted source. I fixed the code and added tests for a missing header, Origin: null, and Sec-Fetch-Site: same-site; the real browser run still passed, because Chrome sends Sec-Fetch-Site: same-origin on the page's own requests. The takeaway: read the authoritative guidance before writing a security check, and make the safe answer the default.
I stopped a server by killing every Python process. To clean up after the browser test I ran taskkill /IM python.exe, which ends all Python processes on the machine, not just my server. Nothing else of mine was running, so no harm was done, but it could have killed unrelated work. Afterwards I started the server in a way that recorded its process id and stopped exactly that process. The takeaway: stop what you started, by id.
A deprecation warning I left in. Starlette 1.6.0 prints that using httpx with its test client is deprecated in favour of httpx2. The tests pass and I have not migrated; if you are building this later, check which test-client package your Starlette version wants.
What This Chapter Does Not Do
HTTPS and proxies. Cookies need Secure, and streaming through a reverse proxy often needs buffering turned off (the response sends X-Accel-Buffering: no, which is nginx's convention; I had no nginx to test). Chapter 12.
Markdown and rich replies. Plain text only. Rendering Markdown safely needs a sanitising renderer; do not add one casually.
Real streaming from the vendors. Verified against documented formats, not against the services.
Abuse at the front door. The bot's per-user limits apply, but an anonymous cookie is easy to throw away, so a determined abuser gets a fresh budget. IP-based limits at the proxy are the usual answer.
Several server processes. The session lock and the rate-limit buckets live in one process's memory.
Hands-On Exercises
Exercise 1
Start the demo server (py serve_web.py 8765), then write a Python client with httpx that loads the page to get the cookie, tries to chat without an Origin header, chats with one and prints the events, and finally tries from a "second browser" with no cookie. Report all four outcomes.
Measure the cost of buffering: with a fake provider that produces a 20-word reply at 0.05 s per word, time how long until the first event and until the last event for stream_live=True and False, and count the events. Say which mode you would choose for a children's library bot and why.
Write a WCAG contrast checker (the relative-luminance formula) and run it on a palette in which two pairs fail (a pale grey note on white and a light red error on grey). Find replacement colours that pass 4.5:1 and re-run it. What does this check not tell you about the page's accessibility?
Providers: complete() returns a Reply; stream() yields text pieces then one final Reply. A stream with no terminator (message_stop / [DONE]) is an error. Retry and fallback only before the first piece
Anthropic SSE: message_start, content_block_delta/text_delta, message_delta (cumulative output tokens), message_stop, ping, and error events can follow a 200. OpenAI-style: data: chunks with delta.content, usage chunk, [DONE]
Bot.stream_message() yields delta, then final (checked text, changed) or error (partial); nothing stored unless the whole reply passed the checks
Live streaming shows unchecked text then replaces it; buffered shows nothing until checked (measured: first text at 0.06 s vs 1.25 s)
Use fetch + ReadableStream + AbortController (EventSource is GET-only); SSE over HTTP/1.1 is limited to about 6 connections per domain
Session: random id, HMAC-signed, HttpOnly, SameSite=Lax, Secure on HTTPS; anonymous, not a login
CSRF (OWASP): Sec-Fetch-Site: same-origin, else Origin on the allow-list, else refuse; require application/json; SameSite is defence in depth. No CORS middleware when the page and API share an origin
Headers: CSP default-src 'self'; frame-ancestors 'none'; base-uri 'none'; form-action 'self', nosniff, no referrer; allow-listed static files; no public API docs; body size limit
Page: textContent never innerHTML; role="log", aria-busy while streaming, label on the textarea, focus returns to the input, Enter / Shift+Enter, Stop, Try again, history on load, 44 px buttons, contrast at least 4.5:1
Coming next
Build a Chatbot: From API Call to Deployed Bot 9 adds the first chat platform, Telegram: creating a bot with BotFather, polling versus webhooks, commands, the typing indicator, the message-length limit and splitting long replies, groups versus private chats and privacy mode, and an adapter that maps Telegram updates onto the same bot core.
The adapter is tested with recorded updates; running it for real needs a token, which this machine does not have.