Chapter 2 · API Design and API Clients
Webhooks and Streaming with Server-Sent Events
- Page 6 of 22
- 26 min read
The last page ended with a client that polls a job again and again, and with a user waiting for a whole answer. Two patterns remove that waiting. A webhook turns polling around: you give the other service a URL, and it calls you when something happens (a payment succeeded, a document finished processing, a build failed). Streaming lets a server send a response piece by piece while it is still being produced, which is why ChatGPT-style interfaces show words appearing instead of a spinner. Webhooks and streaming are the two ways an AI product learns about events that take time, and both have sharp edges: a webhook URL is a public door anyone can knock on, and a stream is a connection that can break halfway.
You will build both ends of a webhook in FastAPI, a sender that signs and retries, and a receiver that verifies, rejects replays and ignores duplicates. Then you will stream with FastAPI's built-in Server-Sent Events support and read streams from the BitByte Practice API's /v1/chat. Start the practice API in another terminal if it is not running (fastapi run practice_api.py --port 8800).
What you will learn
- How a webhook delivery works, and how to sign it with HMAC-SHA256 over the timestamp and the body.
- Why you compare signatures in constant time, protect against replay with a timestamp window, and deduplicate by event id.
- How a sender retries failed deliveries and why a receiver must answer fast and be idempotent.
- What streaming responses and the Server-Sent Events (SSE) wire format look like, with
EventSourceResponseandServerSentEventfromfastapi.sse. - How to consume an SSE stream with
requestsandhttpx, resume withLast-Event-ID, and handle errors and disconnects.
Set up the project
mkdir -p ~/projects/webhooks-demo
cd ~/projects/webhooks-demo
uv init --bare
uv add -q "fastapi[standard]" requests httpx
# with pip instead:
# python -m venv .venv && source .venv/bin/activate
# pip install -q "fastapi[standard]" requests httpxInitialized project `webhooks-demo`As before, the Python snippets continue one another like notebook cells: run them in uv run python from this folder.
Webhooks: the other service calls you
With polling, you ask "is it done yet?" every few seconds and most answers are "no". With a webhook you register a URL once, and the provider sends an HTTP POST to it when an event happens. Stripe, GitHub, Slack, Twilio and most payment, messaging and AI-platform services work this way; a fine-tuning provider can notify you when a model is ready.
| Polling | Webhook | |
|---|---|---|
| Who starts the call | you | the provider |
| Delay | up to one polling interval | almost none |
| Wasted requests | many | none |
| You must run | just a client | a public HTTPS endpoint that is always up |
| Main risk | load and lag | forged, repeated, late and lost events |
That last row is the whole page. A webhook endpoint is an unauthenticated URL on the internet, so anybody can send it a fake "payment succeeded". And delivery is never perfect: the provider cannot know whether your server received an event if the connection died, so it sends again. A well-built webhook system therefore has four parts:
- Authenticity: every delivery carries a signature that only the sender and the receiver can compute.
- Freshness: the signature covers a timestamp, so a captured delivery cannot be replayed an hour later.
- Retries: the sender repeats a delivery until it gets a
2xx, with backoff. - Deduplication: every event has a stable id, and the receiver processes each id once.
Signing: HMAC-SHA256 over timestamp and body
The sender and receiver share a secret (whsec_demo in this tutorial; a long random string in real life, exchanged once and stored in an environment variable). An HMAC mixes the secret into a hash of the message, so the result cannot be produced or checked without the secret. The signed message is the timestamp, a dot, and the raw body bytes. Put this in a small module that both servers import:
# file: signing.py
import hashlib
import hmac
class SignatureError(Exception):
def __init__(self, message: str, status: int = 401):
super().__init__(message)
self.status = status
def sign(secret: str, timestamp: int, body: bytes) -> str:
message = str(timestamp).encode() + b"." + body
return "sha256=" + hmac.new(secret.encode(), message, hashlib.sha256).hexdigest()
def verify_signature(secret: str, timestamp: str | None, signature: str | None,
body: bytes, now: int, tolerance: int = 300) -> None:
if not timestamp or not signature:
raise SignatureError("Missing signature headers", 400)
try:
sent_at = int(timestamp)
except ValueError:
raise SignatureError("Invalid timestamp", 400)
if abs(now - sent_at) > tolerance:
raise SignatureError("Timestamp outside the allowed window", 400)
expected = sign(secret, sent_at, body)
if not hmac.compare_digest(expected.encode(), signature.encode()):
raise SignatureError("Signature does not match", 401)Try it with a fixed timestamp, so every number is repeatable. The check function takes now as an argument for the same reason: code that reads the clock inside is hard to test.
import json
from signing import SignatureError, sign, verify_signature
SECRET = "whsec_demo"
TIMESTAMP = 1767225600
body = b'{"id":"evt_0001","type":"order.paid"}'
signature = sign(SECRET, TIMESTAMP, body)
print(signature)
def check(label, secret=SECRET, ts=TIMESTAMP, sig=signature, payload=body, now=TIMESTAMP + 10):
try:
verify_signature(secret, str(ts), sig, payload, now)
print(f"{label}: accepted")
except SignatureError as exc:
print(f"{label}: rejected {exc.status} ({exc})")
check("untouched")
check("body changed", payload=b'{"id":"evt_0001","type":"order.refunded"}')
check("wrong secret", secret="whsec_other")
check("an hour later", now=TIMESTAMP + 3600)
check("new timestamp, old signature", ts=TIMESTAMP + 3600, now=TIMESTAMP + 3600)
check("re-serialised JSON", payload=json.dumps(json.loads(body)).encode())
check("no headers", ts="", sig="")sha256=ecb28056ad4c9ecb70a48f76e50c5fb959fea17833ae3ce76e4fc58efc825916
untouched: accepted
body changed: rejected 401 (Signature does not match)
wrong secret: rejected 401 (Signature does not match)
an hour later: rejected 400 (Timestamp outside the allowed window)
new timestamp, old signature: rejected 401 (Signature does not match)
re-serialised JSON: rejected 401 (Signature does not match)
no headers: rejected 400 (Missing signature headers)Each rejection teaches a rule:
- Body changed: any edit to the payload breaks the signature, so the receiver can trust what it parses.
- An hour later: a valid delivery is only valid for a window (here 5 minutes). Without the timestamp inside the signature, an attacker who recorded one genuine message could send it again forever. Moving the timestamp forward does not help them (new timestamp, old signature), because the timestamp is part of what was signed.
- Re-serialised JSON: the same data with different spaces is different bytes, and different bytes mean a different signature. So the receiver must verify the raw request body exactly as received, before parsing it into a Pydantic model and dumping it again.
- Constant-time compare:
hmac.compare_digesttakes the same time whatever the input. A plain==stops at the first different character, and an attacker who measures many responses can slowly learn the correct signature one character at a time (a timing attack). The function needsbytes(or ASCIIstr), which is why the code encodes both sides.
The receiver
The receiver does five things, in this order: read the raw body, verify the signature and timestamp, drop events it has already seen, hand the real work to a background task, and answer with a fast 2xx. It must answer quickly (providers time out after a few seconds and then retry), so slow work never happens inside the request. The /admin/fail endpoint exists only to simulate an outage for this lesson; a real admin endpoint needs authentication.
# file: receiver.py
import json
import os
import time
from fastapi import BackgroundTasks, FastAPI, HTTPException, Request
from pydantic import BaseModel
from signing import SignatureError, verify_signature
app = FastAPI(title="Webhook receiver")
SECRET = os.environ["WEBHOOK_SECRET"]
SEEN: set[str] = set()
PROCESSED: list[str] = []
OUTAGE = {"remaining": 0}
class Outage(BaseModel):
count: int
def handle(event: dict) -> None:
PROCESSED.append(event["id"])
@app.get("/health")
def health():
return {"status": "ok"}
@app.post("/webhooks/orders")
async def receive(request: Request, background: BackgroundTasks):
body = await request.body()
try:
verify_signature(
SECRET,
request.headers.get("x-webhook-timestamp"),
request.headers.get("x-webhook-signature"),
body,
now=int(time.time()),
)
except SignatureError as exc:
raise HTTPException(exc.status, str(exc))
if OUTAGE["remaining"] > 0:
OUTAGE["remaining"] -= 1
raise HTTPException(503, "Temporarily unavailable")
event = json.loads(body)
if event["id"] in SEEN:
return {"status": "duplicate"}
SEEN.add(event["id"])
background.add_task(handle, event)
return {"status": "accepted"}
@app.post("/admin/fail")
def fail_next(outage: Outage):
OUTAGE["remaining"] = outage.count
return OUTAGE
@app.get("/processed")
def processed():
return {"processed": PROCESSED}Two details deserve a second look. The unknown-event check uses the id from the body, which is safe only because the signature already proved the body is genuine. And "seen" is kept in memory here; production code stores event ids in a database table with a unique constraint (or Redis SET NX with an expiry) so that deduplication survives a restart and works across several workers.
The sender
The sender is a service that has events. Its /orders/{id}/pay endpoint creates an order.paid event and delivers it to every subscribed URL after answering its own caller, in a background task. Delivery signs the body, posts it, and on any failure waits and tries again with a growing pause (the policy from the previous page: capped attempts, backoff). Every attempt gets a fresh timestamp and signature, but the same event id and the same body. The pauses here are tiny so the lesson runs fast; real senders wait seconds, minutes and hours, and give up after a day or more.
# file: sender.py
import itertools
import json
import time
import requests
from fastapi import BackgroundTasks, FastAPI
from pydantic import BaseModel
from signing import sign
app = FastAPI(title="Webhook sender")
SUBSCRIPTIONS: list[dict] = []
DELIVERIES: list[dict] = []
EVENT_IDS = itertools.count(1)
RETRY_DELAYS = [0.05, 0.1, 0.2]
class SubscriptionIn(BaseModel):
url: str
secret: str
def deliver(event: dict, subscription: dict) -> None:
body = json.dumps(event, separators=(",", ":"), sort_keys=True).encode()
for attempt in range(1, len(RETRY_DELAYS) + 2):
timestamp = int(time.time())
headers = {
"Content-Type": "application/json",
"X-Webhook-Id": event["id"],
"X-Webhook-Timestamp": str(timestamp),
"X-Webhook-Signature": sign(subscription["secret"], timestamp, body),
}
try:
status = requests.post(subscription["url"], data=body, headers=headers, timeout=(2, 5)).status_code
except requests.RequestException:
status = None
DELIVERIES.append({"event_id": event["id"], "attempt": attempt, "status": status})
if status is not None and 200 <= status < 300:
return
if attempt <= len(RETRY_DELAYS):
time.sleep(RETRY_DELAYS[attempt - 1])
@app.get("/health")
def health():
return {"status": "ok"}
@app.post("/subscriptions", status_code=201)
def subscribe(subscription: SubscriptionIn):
SUBSCRIPTIONS.append(subscription.model_dump())
return {"subscriptions": len(SUBSCRIPTIONS)}
@app.post("/orders/{order_id}/pay")
def pay(order_id: int, background: BackgroundTasks):
event = {"id": f"evt_{next(EVENT_IDS):04d}", "type": "order.paid",
"data": {"order_id": order_id, "amount": 25000}}
for subscription in SUBSCRIPTIONS:
background.add_task(deliver, event, subscription)
return {"event_id": event["id"]}
@app.get("/deliveries")
def deliveries():
return DELIVERIESStart both servers. They share the secret through an environment variable, the way real services do (never hard-code it, and never commit it):
export WEBHOOK_SECRET=whsec_demo
uv run fastapi run receiver.py --port 8001 > receiver.log 2>&1 &
echo $! > receiver.pid
uv run fastapi run sender.py --port 8000 > sender.log 2>&1 &
echo $! > sender.pid
curl -s --retry 20 --retry-connrefused --retry-delay 0 http://localhost:8001/health; echo
curl -s --retry 20 --retry-connrefused --retry-delay 0 http://localhost:8000/health; echo{"status":"ok"}
{"status":"ok"}A delivery with an outage
Register the receiver with the sender, then make the receiver fail the next two deliveries with 503, as a real server does during a deploy. Pay for an order and watch what the sender does:
import requests, time
SENDER, RECEIVER = "http://localhost:8000", "http://localhost:8001"
print(requests.post(f"{SENDER}/subscriptions",
json={"url": f"{RECEIVER}/webhooks/orders", "secret": "whsec_demo"}).json())
print(requests.post(f"{RECEIVER}/admin/fail", json={"count": 2}).json())
print(requests.post(f"{SENDER}/orders/1001/pay").json())
def wait_until(check, timeout=5.0):
deadline = time.monotonic() + timeout
while time.monotonic() < deadline:
value = check()
if value:
return value
time.sleep(0.05)
raise TimeoutError("condition never became true")
deliveries = wait_until(lambda: (d := requests.get(f"{SENDER}/deliveries").json()) and d[-1]["status"] == 200 and d)
for delivery in deliveries:
print(delivery)
print(requests.get(f"{RECEIVER}/processed").json()){'subscriptions': 1}
{'remaining': 2}
{'event_id': 'evt_0001'}
{'event_id': 'evt_0001', 'attempt': 1, 'status': 503}
{'event_id': 'evt_0001', 'attempt': 2, 'status': 503}
{'event_id': 'evt_0001', 'attempt': 3, 'status': 200}
{'processed': ['evt_0001']}Three attempts: two 503, then 200. The receiver processed the event once. In real life the trouble is rarely a clean 503. More often your server handles the event, the response is lost, and the sender, seeing a timeout, sends it again. That is why the receiver deduplicates, and why delivery is "at least once": you must be ready for repeats, and the provider usually does not promise to deliver events in order either. Make the handler idempotent (the same idea as the idempotency keys of the last page) and, where order matters, check a timestamp or version in the event.
Attacks that must fail
Now play the attacker. You do know the secret here (you are testing your own receiver), but the point is to confirm each defence works: a duplicate, a replay of an old message, a forged body, and a stolen signature with a new timestamp.
from signing import sign
import json, time
def send(event, secret="whsec_demo", age=0, signature=None, timestamp=None):
body = json.dumps(event, separators=(",", ":"), sort_keys=True).encode()
ts = timestamp if timestamp is not None else int(time.time()) - age
headers = {"X-Webhook-Timestamp": str(ts),
"X-Webhook-Signature": signature or sign(secret, ts, body)}
response = requests.post(f"{RECEIVER}/webhooks/orders", data=body, headers=headers)
return response.status_code, response.json()
event = {"id": "evt_9001", "type": "order.paid", "data": {"order_id": 7, "amount": 100}}
print("first delivery ", send(event))
print("same event again ", send(event))
print("recorded last hr ", send({**event, "id": "evt_9002"}, age=3600))
print("wrong secret ", send({**event, "id": "evt_9003"}, secret="whsec_guess"))
genuine = int(time.time())
old_signature = sign("whsec_demo", genuine - 3600, json.dumps(event, separators=(",", ":"), sort_keys=True).encode())
print("stolen signature ", send(event, signature=old_signature, timestamp=genuine))
print(requests.get(f"{RECEIVER}/processed").json())first delivery (200, {'status': 'accepted'})
same event again (200, {'status': 'duplicate'})
recorded last hr (400, {'detail': 'Timestamp outside the allowed window'})
wrong secret (401, {'detail': 'Signature does not match'})
stolen signature (401, {'detail': 'Signature does not match'})
{'processed': ['evt_0001', 'evt_9001']}Only the first delivery was processed. The repeat got a 200 duplicate (answering success tells the sender to stop retrying), the old message and the guess signed with the wrong secret were refused, and a signature taken from an old message does not survive a new timestamp. Rules to remember when you run webhooks for real:
- Use HTTPS for the receiver URL, otherwise the body and headers travel in clear text. Checking a sender's IP address is a useful extra layer but not a substitute for signatures.
- Answer fast (well under the provider's timeout): store the event, queue the work, return
2xx. A slow handler makes the provider time out and send the event again. - Return the right code:
2xxmeans "stop retrying". Use4xxfor forged or malformed requests and5xxfor your own failures, so retries happen only when they can help. - Rotate secrets: accept two secrets for a while when you change one. Never log the secret or the full signature header.
- Keep an event log and expose a "redeliver" button or endpoint, plus a dead-letter list for events that failed all retries. Providers retry for hours or days, but sooner or later you will need to replay one by hand.
Stop both servers; the second half of the page needs port 8000 for a different app.
kill $(cat receiver.pid) $(cat sender.pid)
while curl -s -o /dev/null http://localhost:8000/health; do sleep 0.1; done
echo stoppedstoppedStreaming responses
A normal response is sent when it is complete. A streaming response sends the headers at once and then the body in chunks as they are produced; the client reads them as they arrive. HTTP supports this natively (chunked transfer encoding). It matters whenever the full answer is slow but the beginning is already useful:
- LLM output: a long answer takes tens of seconds to finish, but the first token arrives in under a second. Streaming makes the product feel instant and lets the user stop early.
- Progress and logs of a job, live search results, large exports you do not want to hold in memory.
- Agents: showing "searching…", "reading page 3…" and partial answers while tools run.
The lowest-level tool in FastAPI is Starlette's StreamingResponse, which wraps any generator. Here is a tiny server with one plain stream; the other endpoints on it come next. Save it as main.py:
# file: main.py
import asyncio
from collections.abc import AsyncIterable
from typing import Annotated
from fastapi import FastAPI, Header
from fastapi.responses import StreamingResponse
from fastapi.sse import EventSourceResponse, ServerSentEvent
app = FastAPI(title="Streaming demo")
WORDS = "Server-Sent Events are plain text over ordinary HTTP".split()
CLOSED: list[str] = []
@app.get("/health")
def health():
return {"status": "ok"}
@app.get("/plain")
async def plain():
async def chunks():
for word in WORDS:
await asyncio.sleep(0.01)
yield word + " "
return StreamingResponse(chunks(), media_type="text/plain")
@app.get("/stream", response_class=EventSourceResponse)
async def stream(
last_event_id: Annotated[int | None, Header()] = None,
fail_at: int | None = None,
) -> AsyncIterable[ServerSentEvent]:
start = last_event_id or 0
yield ServerSentEvent(comment="stream open", retry=3000)
for number, word in enumerate(WORDS[start:], start=start + 1):
await asyncio.sleep(0.02)
if number == fail_at:
yield ServerSentEvent(data={"message": "the model failed"}, event="error")
return
yield ServerSentEvent(data={"token": word}, event="token", id=str(number))
yield ServerSentEvent(data={"done": True}, event="done")
@app.get("/ticks", response_class=EventSourceResponse)
async def ticks() -> AsyncIterable[ServerSentEvent]:
try:
for number in range(1, 1000):
await asyncio.sleep(0.02)
yield ServerSentEvent(data={"tick": number})
finally:
CLOSED.append("ticks")
@app.get("/closed")
def closed():
return {"closed": len(CLOSED)}uv run fastapi run main.py --port 8000 > server.log 2>&1 &
echo $! > server.pid
curl -s --retry 20 --retry-connrefused --retry-delay 0 http://localhost:8000/health; echo
curl -sN http://localhost:8000/plain; echo{"status":"ok"}
Server-Sent Events are plain text over ordinary HTTPThe words arrived one by one (the -N flag tells curl not to buffer). That is plain chunked text. It works, but the browser or client has no idea where one message ends, what kind it is, or where to resume. That is the gap Server-Sent Events fills.
Server-Sent Events (SSE)
SSE is a tiny standard on top of HTTP: the response has Content-Type: text/event-stream and a body made of events. Each event is a few lines of field: value, and a blank line ends the event.
| Field | Meaning |
|---|---|
data: | The payload. Several data: lines in one event are joined with newlines. Usually JSON. |
event: | The event's name (default message), so a client can handle token, error and done differently. |
id: | An id the client remembers. On reconnect it sends it back in the Last-Event-ID header so the server can resume. |
retry: | How many milliseconds a browser should wait before reconnecting. |
a line starting with : | A comment, ignored by clients. Servers use it as a keep-alive ping so proxies do not close an idle connection. |
Compared with WebSockets, SSE is one-directional (server to client), runs over plain HTTP (so it works with proxies, load balancers, auth headers and HTTP/2), and has reconnection built in. That is exactly what token streaming needs. WebSockets earn their extra complexity when the client must also send a continuous flow of messages (live audio, collaborative editing). Look at the real thing: the practice API's /v1/chat with "stream": true sends a data: line per piece of the answer and a last one with the token usage:
curl -sN -X POST http://localhost:8800/v1/chat \
-H "Authorization: Bearer token-nadia-001" \
-H "Content-Type: application/json" \
-d '{"messages": [{"role": "user", "content": "What is RAG?"}], "stream": true}'data: {"delta": "This "}
data: {"delta": "is "}
data: {"delta": "a "}
data: {"delta": "practice "}
data: {"delta": "answer "}
data: {"delta": "about: "}
data: {"delta": "What "}
data: {"delta": "is "}
data: {"delta": "RAG. "}
data: {"delta": "A "}
data: {"delta": "real "}
data: {"delta": "model "}
data: {"delta": "would "}
data: {"delta": "say "}
data: {"delta": "more. "}
data: {"done": true, "usage": {"prompt_tokens": 3, "completion_tokens": 15, "total_tokens": 18}}Real LLM providers use the same transport with richer events. OpenAI's Responses API streams typed events, and Anthropic's Messages API streams message_start, content_block_start, content_block_delta, content_block_stop, message_delta and message_stop, plus ping and error. Notice a detail of Anthropic's design: error is an event inside a stream that already returned 200. That is unavoidable: once the first byte is sent, the status code cannot change.
FastAPI's built-in SSE
FastAPI has SSE support in the box (no extra package). Three pieces, all visible in the /stream endpoint above:
response_class=EventSourceResponseon the route setsContent-Type: text/event-streamand turns on SSE encoding. It works for any HTTP method, includingPOST, which LLM APIs need because prompts are big.- The endpoint is a generator: a function that uses
yield. Annotating it asAsyncIterable[ServerSentEvent]documents what it sends. - Yield a
ServerSentEvent(data=…, event=…, id=…, retry=…, comment=…)for full control, or yield a plain dict or Pydantic model and it becomes a JSONdata:line. Data is always JSON-encoded (so the string"hello"is sent with its quotes); useraw_data=to send text as it is.eventandidmust be a single line, and FastAPI adds theCache-Control: no-cacheandX-Accel-Buffering: noheaders for you (the second one tells an Nginx proxy not to buffer the stream).
Look at the raw bytes this endpoint sends. The first line is a comment that also sets the retry time, then each token is one named event with an id:
curl -sN http://localhost:8000/stream: stream open
retry: 3000
event: token
data: {"token": "Server-Sent"}
id: 1
event: token
data: {"token": "Events"}
id: 2
event: token
data: {"token": "are"}
id: 3
event: token
data: {"token": "plain"}
id: 4
event: token
data: {"token": "text"}
id: 5
event: token
data: {"token": "over"}
id: 6
event: token
data: {"token": "ordinary"}
id: 7
event: token
data: {"token": "HTTP"}
id: 8
event: done
data: {"done": true}The same endpoint can resume. When a client reconnects with Last-Event-ID: 5 (a browser's EventSource does this by itself), the server continues from word 6:
curl -sN -H "Last-Event-ID: 5" http://localhost:8000/stream | grep -E "^(id|event: done)"id: 6
id: 7
id: 8
event: doneConsuming SSE from Python
A browser has EventSource, but it can only send GET requests, so chat UIs usually use fetch with a stream reader. In Python you read the response line by line and apply the format yourself. The parser below is about twenty lines: it collects fields until a blank line, ignores comments, and joins multi-line data. Two details matter: set the encoding to UTF-8 (requests assumes Latin-1 for text/* without a charset, which garbles Bangla), and pass chunk_size=None so lines are handed to you as they arrive instead of after a buffer fills.
def parse_sse(lines):
"""Turn the lines of an SSE response into events: dicts with event, data and id."""
event = {"event": "message", "data": [], "id": None}
for line in lines:
if line == "":
if event["data"]:
yield {**event, "data": "\n".join(event["data"])}
event = {"event": "message", "data": [], "id": None}
elif line.startswith(":"):
continue
else:
field, _, value = line.partition(":")
value = value.removeprefix(" ")
if field == "data":
event["data"].append(value)
elif field in ("event", "id"):
event[field] = value
CHAT = "http://localhost:8800/v1/chat"
AUTH = {"Authorization": "Bearer token-nadia-001"}
question = {"messages": [{"role": "user", "content": "What is retrieval augmented generation?"}], "stream": True}
pieces, usage = [], None
with requests.post(CHAT, json=question, headers=AUTH, stream=True, timeout=(3, 30)) as response:
response.raise_for_status()
response.encoding = "utf-8"
for event in parse_sse(response.iter_lines(decode_unicode=True, chunk_size=None)):
data = json.loads(event["data"])
if data.get("done"):
usage = data["usage"]
else:
pieces.append(data["delta"])
print(len(pieces), "pieces; first three:", pieces[:3])
print("".join(pieces).strip())
print(usage)17 pieces; first three: ['This ', 'is ', 'a ']
This is a practice answer about: What is retrieval augmented generation. A real model would say more.
{'prompt_tokens': 5, 'completion_tokens': 17, 'total_tokens': 22}The same with httpx, which you will use for production clients on the next-but-one page. Its stream() context manager does not read the body until you ask, and iter_lines() already decodes UTF-8. The timeout is a per-chunk read timeout, so a healthy stream that runs for minutes is never cut:
import httpx
with httpx.stream("POST", CHAT, json=question, headers=AUTH, timeout=httpx.Timeout(30.0, connect=3.0)) as response:
response.raise_for_status()
text = "".join(json.loads(e["data"]).get("delta", "") for e in parse_sse(response.iter_lines()))
print(text.strip())This is a practice answer about: What is retrieval augmented generation. A real model would say more.Next, the events from our own endpoint, with names and ids. A helper reads tokens and can stop after a few events (leaving the with block closes the connection, which is exactly what a user pressing "stop" does), and can resume from an id:
APP = "http://localhost:8000"
def read_tokens(stop_after=None, last_event_id=None, params=None):
headers = {"Last-Event-ID": last_event_id} if last_event_id else {}
got = []
with requests.get(f"{APP}/stream", params=params, headers=headers, stream=True, timeout=5) as response:
response.encoding = "utf-8"
for event in parse_sse(response.iter_lines(decode_unicode=True, chunk_size=None)):
if event["event"] == "token":
got.append((event["id"], json.loads(event["data"])["token"]))
if stop_after and len(got) >= stop_after:
break
elif event["event"] == "error":
got.append(("error", json.loads(event["data"])["message"]))
return got
first = read_tokens(stop_after=3)
print("connection dropped after:", first)
rest = read_tokens(last_event_id=first[-1][0])
print("resumed:", rest)
print("whole sentence:", " ".join(word for _, word in first + rest))
print("model failure:", read_tokens(params={"fail_at": 3}))connection dropped after: [('1', 'Server-Sent'), ('2', 'Events'), ('3', 'are')]
resumed: [('4', 'plain'), ('5', 'text'), ('6', 'over'), ('7', 'ordinary'), ('8', 'HTTP')]
whole sentence: Server-Sent Events are plain text over ordinary HTTP
model failure: [('1', 'Server-Sent'), ('2', 'Events'), ('error', 'the model failed')]Resuming works because the server numbers its events and can restart from any number. An LLM stream cannot resume in the middle of a model call, so an AI backend typically either restarts the generation or keeps the partial text per request id. The last line shows the error pattern: the HTTP status was already 200, so the failure travels as an error event, and the client must handle it as part of the stream.
Disconnects and cancellation
When the client leaves (closed tab, "stop generating", dead phone connection) the server should stop work, because an LLM call that nobody reads still costs money. FastAPI cancels the generator when it notices the disconnect; the finally in /ticks proves it ran. The stream below would never end on its own; the client takes three events and walks away:
with requests.get(f"{APP}/ticks", stream=True, timeout=5) as response:
response.encoding = "utf-8"
for count, event in enumerate(parse_sse(response.iter_lines(decode_unicode=True, chunk_size=None)), start=1):
print(event["data"])
if count == 3:
break
print("server noticed:", wait_until(lambda: requests.get(f"{APP}/closed").json()["closed"] or None)){"tick": 1}
{"tick": 2}
{"tick": 3}
server noticed: 1Put cleanup (cancel the upstream provider call, record how many tokens were used) in a try … finally around the loop. A few more production notes: send a comment ping every 15 to 30 seconds on quiet streams (FastAPI does this for you when a generator is idle); turn off response buffering and compression in any proxy in front of the stream; and remember that every open stream holds a connection, so cap how many one user may open. Building the streaming chat endpoint of a full LLM backend is the project of Building an LLM Backend with Streaming.
Real-world use
- Payment, source-control and chat platforms push events by webhook (Stripe, GitHub, Slack). Their libraries verify signatures exactly as above.
- AI platforms notify by webhook when a batch job, a fine-tune or a long agent run finishes, and stream tokens over SSE for interactive answers.
- Chat products stream a model's answer through your own backend to the browser, adding auth, usage limits and logging on the way.
- Agent frameworks and MCP use SSE-style streaming over HTTP to send progress and partial results while tools run.
- Your own services use webhooks to tell each other "document indexed" or "model deployed" without polling.
Common mistakes
- Verifying the parsed JSON instead of the raw body. Re-serialising changes the bytes and the signature never matches (or, worse, you skip verification to make it "work"). Read
await request.body(), verify, then parse. - Comparing signatures with
==or skipping the timestamp. The first leaks timing, the second allows replays forever. Usehmac.compare_digestand a tolerance window of a few minutes. - Doing the work inside the webhook request. A slow handler makes the sender time out and retry, which causes duplicates. Store, enqueue, answer
2xx, deduplicate by event id. - Assuming webhooks arrive once and in order. They arrive at least once, sometimes late and out of order. Make handlers idempotent and check versions or timestamps.
- Treating a stream like a normal response. Proxies that buffer, clients without a read timeout, no handling of an
errorevent, and no cleanup on disconnect. Test withcurl -N, disable buffering, and usetry … finally.
Try it yourself
- Easy: use
signfromsigning.pyto compute the signature of the bodyb'{"id":"evt_9"}'with the secretwhsec_exerciseand timestamp1767225600, and print it. - Medium: use
verify_signatureto check that signature atnow = 1767225610(should pass), with the body changed tob'{"id":"evt_8"}'(should fail), and atnow = 1767229600(should fail). Print what each check does. - Medium: with
httpx, stream/v1/chatfor the question "Why use webhooks?" and print how many pieces arrived, the joined answer, andtotal_tokensfrom the final event. - Hard: write
read_until_done(drop_after), which reads/stream, simulates a dropped connection afterdrop_aftertoken events on the first connection, reconnects withLast-Event-ID, and keeps going until thedoneevent. Return the full sentence and the number of connections used.
Answers
# 1. Easy
exercise_body = b'{"id":"evt_9"}'
exercise_signature = sign("whsec_exercise", 1767225600, exercise_body)
print(exercise_signature)
# 2. Medium
def try_verify(payload, now):
try:
verify_signature("whsec_exercise", "1767225600", exercise_signature, payload, now)
return "accepted"
except SignatureError as exc:
return f"rejected: {exc}"
print(try_verify(exercise_body, 1767225610))
print(try_verify(b'{"id":"evt_8"}', 1767225610))
print(try_verify(exercise_body, 1767229600))
# 3. Medium
payload = {"messages": [{"role": "user", "content": "Why use webhooks?"}], "stream": True}
deltas, total = [], None
with httpx.stream("POST", CHAT, json=payload, headers=AUTH, timeout=30.0) as response:
response.raise_for_status()
for event in parse_sse(response.iter_lines()):
data = json.loads(event["data"])
if data.get("done"):
total = data["usage"]["total_tokens"]
else:
deltas.append(data["delta"])
print(len(deltas), "pieces")
print("".join(deltas).strip())
print("total_tokens:", total)
# 4. Hard
def read_until_done(drop_after):
tokens, last_id, connections = [], None, 0
while True:
connections += 1
headers = {"Last-Event-ID": last_id} if last_id else {}
seen, finished = 0, False
with requests.get(f"{APP}/stream", headers=headers, stream=True, timeout=5) as response:
response.encoding = "utf-8"
for event in parse_sse(response.iter_lines(decode_unicode=True, chunk_size=None)):
if event["event"] == "done":
finished = True
break
if event["event"] == "token":
tokens.append(json.loads(event["data"])["token"])
last_id = event["id"]
seen += 1
if connections == 1 and seen == drop_after:
break
if finished:
return " ".join(tokens), connections
print(read_until_done(drop_after=4))sha256=06ad48768c3077908ecc479f6b333b30dec6e25fe743f7490bbad8a40e7d4643
accepted
rejected: Signature does not match
rejected: Timestamp outside the allowed window
15 pieces
This is a practice answer about: Why use webhooks. A real model would say more.
total_tokens: 18
('Server-Sent Events are plain text over ordinary HTTP', 2)Stop the server when you are done:
kill $(cat server.pid)Summary
- A webhook is the provider calling your URL when an event happens. Delivery is at least once, possibly late and out of order, so verify, deduplicate and keep handlers idempotent.
- Sign
timestamp + "." + raw bodywith HMAC-SHA256 and a shared secret; the receiver recomputes it on the raw bytes, compares withhmac.compare_digest, and rejects timestamps outside a few minutes. Retries reuse the event id and get a fresh signature. - Answer webhooks fast with a
2xx, do the work in the background, use4xxfor bad requests and5xxfor your own failures, and keep an event log with redelivery. - Streaming sends the body while it is being produced. SSE is its standard format:
data,event,idandretryfields, a blank line between events, and:comments as keep-alives. FastAPI has it built in:response_class=EventSourceResponseplus a generator yieldingServerSentEventobjects. - Read streams line by line with
requests(stream=True, UTF-8) orhttpx.stream(); resume withLast-Event-ID; treat anerrorevent as part of the stream; and clean up infinallywhen the client disconnects.
Next: Professional API Clients with httpx and asyncio takes the client side of everything you have built so far, retries, rate limits, pagination and streams, and turns it into fast, well-tested asynchronous client code.