How to Stream Model Output to a Browser With Server-Sent Events in FastAPI, Stop the Model When the User Leaves, and Resume After a Dropped Connection

Open any AI chat window and the answer appears a few words at a time. Behind that effect is one long HTTP response that the server keeps open and writes to as the model produces tokens. Browsers have a built-in way to read such a response, called Server-Sent Events (SSE), and FastAPI has been able to write one natively since version 0.135.0. A first demo takes ten lines. Making it behave when real people close tabs, lose Wi-Fi and sit behind proxies takes the rest of this tutorial.

You will build the whole thing in small steps and watch each problem happen before you fix it. By the end you will have a FastAPI service that streams a model’s answer to a browser, stops the model when the user leaves, picks up where it left off after a dropped connection, and has a test suite that notices when any of that breaks. Four problems come up on the way, and every one of them fails silently:

1. A token that contains a line break loses it on the wire (Step 2).
2. A stream that ends normally makes the browser start the whole answer again (Step 4).
3. A model keeps generating, and costing money, after the user has left (Step 5).
4. A reconnect repeats or loses text (Step 6).

Everything below ran on Windows 11 with Python 3.13.14, FastAPI 0.162.0 (with Starlette 1.10.0), uvicorn 0.54.0, Microsoft Edge 156.0.4314.50 driven by Playwright 1.64.1, and, for the two optional steps, Ollama 0.32.9 and nginx 1.30.6. Every number you will see was printed by a script on this page. Your timings will differ a little; the pattern should not.

Server-Sent Events in plain words

A normal HTTP response has a start, a body of known size and an end, and the browser shows it once it is complete. A streaming response keeps the connection open and sends the body in pieces as they become ready. SSE is a small agreement about what those pieces look like, so the browser can turn them into JavaScript events for you instead of making you parse raw bytes.

What the stream looks like on the wire

An event stream is plain text with the content type text/event-stream. This is a short piece of a real one, captured from the server you will build in Step 6 and the one in Step 3:

: stream open
retry: 1000

data: "Server-sent "
id: 0

data: "events "
id: 1

event: done
data: [DONE]

Each line is a field name, a colon and a value. Five kinds of line matter:

• data: carries the payload. If an event has several data: lines, the browser joins them with a line break.
• id: labels the event. The browser remembers the most recent one.
• event: names the event type. Without it the type is message.
• retry: tells the browser how many milliseconds to wait before it reconnects.
• A line that starts with a colon is a comment. The browser ignores it, which makes comments handy as keep-alive pings.

A blank line ends an event. Until the blank line arrives, the browser holds the lines it has seen and shows nothing to your code. Two small rules in the WHATWG HTML Standard will matter in Step 2. First, “If value starts with a U+0020 SPACE character, remove it from value.” So exactly one space after the colon is dropped and any further spaces belong to your data. Second, when an event is dispatched, “If the data buffer’s last character is a U+000A LINE FEED (LF) character, then remove the last character from the data buffer.” So several data: lines come back joined by line breaks, with no extra break at the end.

What the browser does with it

In JavaScript you write new EventSource(url). The browser sends a GET request, fires an open event, and fires a message event for every complete event (or an event of the name you chose with event:). The constructor takes only a URL and a withCredentials option, so it cannot send a request body or custom headers. That is why Step 6 creates the job with an ordinary POST and then attaches the stream with a GET.

The behavior that causes most production bugs is what happens when the response ends. The standard says that when the body ends without a network error, the browser reestablishes the connection. It waits for a reconnection time that starts as “an implementation-defined value, probably in the region of a few seconds”, and a retry: line changes it. The new request carries a Last-Event-ID header holding the last id: the browser saw. Two exits exist. If the response status is not 200 or its content type is not text/event-stream, the browser fails the connection, and “Once the user agent has failed the connection, it does not attempt to reconnect.” And “a client can be told to stop reconnecting using the HTTP 204 No Content response code.” Keep these four facts in mind (reconnect on end, retry, Last-Event-ID, the two exits): Steps 4 and 6 are built on them.

When SSE is the right tool

SSE only flows from server to browser, over ordinary HTTP. For a chat answer that is exactly what you need: the question goes up in a normal POST and the answer comes back as a stream. If the browser must also push a steady flow of messages while the stream is open, as in a multiplayer game, a WebSocket fits better.

Prerequisites and setup

• Operating system: I ran everything on Windows 11. The Python code is the same on macOS and Linux; only the command that activates the virtual environment differs.
• Python 3.13 (the version I tested) and a terminal.
• Knowledge: you can read a Python function and a FastAPI route, and you know what async def means. The two short JavaScript files are explained line by line.
• A Chromium-based browser for Steps 4, 6 and 8. I used the Microsoft Edge that ships with Windows 11 through Playwright’s channel="msedge" option. I did not test other browsers.
• Optional: Ollama with the qwen2.5:1.5b model for Step 7, and nginx for Step 8. Every step before them uses a fake model, so you can skip both and lose nothing.

Create a folder, a virtual environment and the file that pins the versions used here:

fastapi==0.162.0
uvicorn==0.54.0
httpx==0.28.1
pytest==9.1.1
playwright==1.64.1

mkdir sse_lab
cd sse_lab
python -m venv venv
venv\Scripts\activate
pip install -r requirements.txt

On macOS or Linux, replace the activation line with source venv/bin/activate. Save every file from this page in sse_lab, next to each other: the scripts import one another by name. To confirm the install, this is what the run script at the end prints for the packages that matter:

fastapi==0.162.0
h11==0.16.0
httpx==0.28.1
playwright==1.64.1
pytest==9.1.1
starlette==1.10.0
uvicorn==0.54.0

Step 1: Build a stand-in for the model

Real models are slow, cost money and give a different answer each time. For learning how streaming behaves you want the opposite: a source that is free, repeatable and that counts its own work. The module below streams a fixed 39-token answer, one token every 50 milliseconds. A token here is a word plus the whitespace after it, so joining all tokens gives back the original text exactly. One token, event. followed by a blank line, contains a paragraph break. That token is a trap on purpose.

"""fake_model.py: a stand-in for an LLM. It streams a fixed answer one token at a time and counts its work."""
import asyncio
import re
import time

ANSWER = (
    "Server-sent events let a server push text to a browser over one long HTTP response. "
    "Each token you see here arrives as its own event.\n\n"
    "If the connection drops, the browser reconnects and says which event it saw last."
)
TOKENS = re.findall(r"\S+\s*", ANSWER)  # words that keep their trailing space or newlines
STATS = {"produced": 0, "finished": 0, "stopped": 0}


def reset_stats() -> dict:
    for key in STATS:
        STATS[key] = 0
    return STATS


async def generate(delay: float = 0.05, fail_after: int | None = None):
    """Async generator: one token every `delay` seconds; fail_after=N makes the backend crash at token N."""
    done = False
    try:
        for n, token in enumerate(TOKENS):
            if fail_after is not None and n >= fail_after:
                raise RuntimeError("model backend crashed")
            await asyncio.sleep(delay)
            STATS["produced"] += 1
            yield token
        done = True
    finally:
        STATS["finished" if done else "stopped"] += 1


def generate_blocking(delay: float = 0.05):
    """Plain generator for `def` endpoints; time.sleep stands in for a blocking model call."""
    done = False
    try:
        for token in TOKENS:
            time.sleep(delay)
            STATS["produced"] += 1
            yield token
        done = True
    finally:
        STATS["finished" if done else "stopped"] += 1

The STATS dictionary is how later steps see wasted work. Every token the model produces adds one to produced. When a generator ends, a finally block adds one to finished if it ran to the end, or to stopped if something cut it short. generate() is an async generator that waits with asyncio.sleep, and generate_blocking() is its plain (synchronous) twin that waits with time.sleep, standing in for a model client that blocks. The fail_after switch makes the fake backend crash at a chosen token, which Step 6 uses.

Step 2: Write an event stream by hand and watch it go wrong

Before you let a library write events, it pays to write a few by hand. You will need three helpers. Do not worry about every line of the first two; their job is to let the later scripts start a server, read a stream and judge the result the way a browser would.

Helpers: start a server, read a stream, parse like a browser

labkit.py starts python -m uvicorn as a child process and waits until the port accepts connections. read_events uses only the standard library’s http.client: it sends a request, reads the response line by line, and groups the lines into events at each blank line. Passing stop_after=5 closes the connection after five events, which is exactly what a user closing the tab does.

"""labkit.py: helpers shared by the step scripts: start a uvicorn server, read an event stream, call JSON endpoints."""
import http.client
import json
import os
import socket
import subprocess
import sys
import time

HERE = os.path.dirname(os.path.abspath(__file__))


def start(app: str, port: int, *extra: str) -> subprocess.Popen:
    """Run `uvicorn app --port port` in the lab folder and wait until it accepts connections."""
    cmd = [sys.executable, "-m", "uvicorn", app, "--port", str(port), "--log-level", "warning", *extra]
    proc = subprocess.Popen(cmd, cwd=HERE)
    for _ in range(100):
        try:
            socket.create_connection(("127.0.0.1", port), timeout=0.2).close()
            return proc
        except OSError:
            time.sleep(0.1)
    raise RuntimeError(f"{app} did not start on port {port}")


def stop(proc: subprocess.Popen) -> None:
    proc.terminate()
    proc.wait(timeout=10)


def call(port: int, method: str, path: str, body: dict | None = None) -> dict:
    conn = http.client.HTTPConnection("127.0.0.1", port, timeout=30)
    payload = json.dumps(body) if body is not None else None
    conn.request(method, path, body=payload, headers={"Content-Type": "application/json"} if payload else {})
    resp = conn.getresponse()
    out = json.loads(resp.read() or b"null")
    conn.close()
    return out


def read_events(port: int, path: str, *, method: str = "GET", headers: dict | None = None,
                body: dict | None = None, stop_after: int | None = None):
    """Open a stream and return (status, headers, chunks); each chunk is (seconds since request, raw text of one event).

    With stop_after=N the connection is closed after N events, the way a user closing the tab would."""
    conn = http.client.HTTPConnection("127.0.0.1", port, timeout=60)
    payload = json.dumps(body) if body is not None else None
    send = {"Accept": "text/event-stream", **(headers or {})}
    if payload:
        send["Content-Type"] = "application/json"
    t0 = time.perf_counter()
    conn.request(method, path, body=payload, headers=send)
    resp = conn.getresponse()
    chunks, current = [], []
    while True:
        line = resp.readline()
        if not line:
            break
        current.append(line.decode("utf-8"))
        if line in (b"\n", b"\r\n"):  # blank line: the event is complete
            chunks.append((time.perf_counter() - t0, "".join(current)))
            current = []
            if stop_after is not None and len(chunks) >= stop_after:
                break
    status, hdrs = resp.status, {k.lower(): v for k, v in resp.getheaders()}
    conn.close()
    return status, hdrs, chunks

sse_parse.py is a small parser that follows the standard’s rules: join data: lines, drop one leading space, ignore comments, remember the last id:, read retry:, and dispatch an event at each blank line. It deliberately throws away the final piece of the text, which is either empty or a half-received line that a real browser would still be waiting on. A test in Step 9 caught me getting that wrong the first time.

"""sse_parse.py: a small parser that follows the WHATWG rules for reading a text/event-stream."""
import re


def parse(text: str) -> list[dict]:
    """Return the events a browser would dispatch: dicts with type, data, id and retry."""
    events = []
    data, event_type, last_id, retry = [], "", "", None
    lines = re.split(r"\r\n|\n|\r", text)
    lines.pop()  # the last piece is empty, or a half-received line that a browser would still be waiting on
    for line in lines:
        if line == "":  # a blank line dispatches the event
            if data:
                events.append({"type": event_type or "message", "data": "\n".join(data), "id": last_id, "retry": retry})
            data, event_type, retry = [], "", None
            continue
        if line.startswith(":"):  # comment, ignored
            continue
        name, _, value = line.partition(":")
        if value.startswith(" "):
            value = value[1:]  # only ONE leading space is removed
        if name == "data":
            data.append(value)
        elif name == "event":
            event_type = value
        elif name == "id" and "\0" not in value:
            last_id = value
        elif name == "retry" and value.isdigit():
            retry = int(value)
    return events  # an event without its final blank line is discarded, as in a browser

Two hand-written endpoints

FastAPI’s StreamingResponse sends whatever an async generator yields, as it is yielded. With media_type="text/event-stream" the browser treats it as SSE. The first endpoint writes the simplest possible event: data:, the token, and a blank line. The second splits the token at its line breaks and writes one data: line for each part, and it adds an id: that Step 6 will need.

"""app_v1.py: Server-Sent Events written by hand with StreamingResponse (FastAPI 0.162.0, Starlette 1.10.0)."""
from fastapi import FastAPI
from fastapi.responses import StreamingResponse

import fake_model

app = FastAPI()


@app.get("/stats")
def stats() -> dict:
    return fake_model.STATS


@app.post("/stats/reset")
def reset() -> dict:
    return fake_model.reset_stats()


async def naive_events():
    async for token in fake_model.generate():
        yield f"data: {token}\n\n"  # BUG: a newline inside the token ends the event early


async def fixed_events():
    i = 0
    async for token in fake_model.generate():
        lines = "".join(f"data: {part}\n" for part in token.split("\n"))  # one data: line per text line
        yield f"id: {i}\n{lines}\n"
        i += 1


@app.get("/naive")
async def naive() -> StreamingResponse:
    return StreamingResponse(naive_events(), media_type="text/event-stream")


@app.get("/fixed")
async def fixed() -> StreamingResponse:
    return StreamingResponse(fixed_events(), media_type="text/event-stream")

Now read both streams, rebuild the text the way a browser would, and compare it with the model’s answer:

"""demo_wire.py: read the hand-written streams and see what a browser would reconstruct."""
import labkit
import sse_parse
from fake_model import ANSWER

proc = labkit.start("app_v1:app", 8101)
try:
    for path in ("/naive", "/fixed"):
        status, headers, chunks = labkit.read_events(8101, path)
        raw = "".join(text for _, text in chunks)
        events = sse_parse.parse(raw)
        rebuilt = "".join(e["data"] for e in events)
        print(f"GET {path} -> {status} {headers['content-type']}, {len(chunks)} chunks, {len(events)} events")
        print("  first two chunks:", [text for _, text in chunks[:2]])
        token_chunk = next(text for _, text in chunks if "event." in text)
        stray = sum(1 for _, text in chunks if text == "\n")
        print("  paragraph-break token on the wire:", repr(token_chunk), f"followed by {stray} stray blank lines")
        print("  rebuilt text equals the model's answer:", rebuilt == ANSWER)
        if rebuilt != ANSWER:
            print("  rebuilt around the break:", repr(rebuilt[rebuilt.index("Each"):rebuilt.index("Each") + 60]))
finally:
    labkit.stop(proc)

python demo_wire.py

GET /naive -> 200 text/event-stream; charset=utf-8, 41 chunks, 39 events
  first two chunks: ['data: Server-sent \n\n', 'data: events \n\n']
  paragraph-break token on the wire: 'data: event.\n\n' followed by 2 stray blank lines
  rebuilt text equals the model's answer: False
  rebuilt around the break: 'Each token you see here arrives as its own event.If the conn'
GET /fixed -> 200 text/event-stream; charset=utf-8, 39 chunks, 39 events
  first two chunks: ['id: 0\ndata: Server-sent \n\n', 'id: 1\ndata: events \n\n']
  paragraph-break token on the wire: 'id: 24\ndata: event.\ndata: \ndata: \n\n' followed by 0 stray blank lines
  rebuilt text equals the model's answer: True

Look at the naive stream first. The paragraph token carries two line breaks of its own, and the endpoint adds two more, so four line breaks arrive in a row after data: event.. The first ends the data line, the second is a blank line that dispatches the event, and the other two are stray blank lines, which is where the “2 stray blank lines” in the output come from. The parser dispatched an event with the data event. and ignored the empty ones. Nothing failed, but the text the user would see is event.If the conn: the paragraph break is gone. In the fixed stream the token became three data: lines (event., an empty one and another empty one), the browser joins them with line breaks, drops the last, and the text matches the answer exactly.

The rule to remember: a line break inside event data must become a new data: line, or the data must be encoded so that it has no line breaks at all. Step 3 takes the second route.

Step 3: Let FastAPI write the events

Writing the format by hand is educational and error-prone. FastAPI 0.135.0 and later do it for you, and the FastAPI documentation says so directly: “Added in FastAPI 0.135.0.” You declare response_class=EventSourceResponse, make the function a generator, and yield either plain objects or ServerSentEvent instances when you need to set id, event, retry or a comment. The docs add that “Each yielded item is encoded as JSON and sent in the data: field of an SSE event.”

"""app_v2.py: the same stream with FastAPI's built-in Server-Sent Events support (added in FastAPI 0.135.0)."""
import asyncio
from collections.abc import AsyncIterable

from fastapi import FastAPI
from fastapi.responses import FileResponse
from fastapi.sse import EventSourceResponse, ServerSentEvent

import fake_model

app = FastAPI()


@app.get("/")
def page() -> FileResponse:
    return FileResponse("index.html")


@app.get("/stats")
def stats() -> dict:
    return fake_model.STATS


@app.post("/stats/reset")
def reset() -> dict:
    return fake_model.reset_stats()


@app.get("/stream", response_class=EventSourceResponse)
async def stream() -> AsyncIterable[ServerSentEvent]:
    i = 0
    async for token in fake_model.generate():
        yield ServerSentEvent(data=token, id=str(i))  # data is JSON-encoded, so newlines are escaped
        i += 1
    yield ServerSentEvent(raw_data="[DONE]", event="done")


@app.get("/slow", response_class=EventSourceResponse)
async def slow() -> AsyncIterable[ServerSentEvent]:
    yield ServerSentEvent(data="first")
    await asyncio.sleep(17)  # a model that goes quiet longer than the 15 second ping interval
    yield ServerSentEvent(data="second")

Three details are worth understanding. data=token is JSON-encoded, so the token "event.\n\n" travels with escaped line breaks and the browser must call JSON.parse on it. raw_data="[DONE]" sends text as it is, without encoding, which is what you want for a marker that is not JSON. And the /slow route exists only to show one more behavior: it yields once, sleeps for 17 seconds, and yields again.

"""demo_builtin.py: the built-in EventSourceResponse: headers, wire format, JSON-encoded data and the idle ping."""
import json

import labkit
import sse_parse
from fake_model import ANSWER

proc = labkit.start("app_v2:app", 8102)
try:
    status, headers, chunks = labkit.read_events(8102, "/stream")
    print(f"GET /stream -> {status}, {len(chunks)} events")
    for name in ("content-type", "cache-control", "x-accel-buffering", "transfer-encoding"):
        print(f"  {name}: {headers.get(name)}")
    print("  first three chunks:")
    for _, text in chunks[:3]:
        print("   ", repr(text))
    print("  last chunk:", repr(chunks[-1][1]))
    events = sse_parse.parse("".join(text for _, text in chunks))
    rebuilt = "".join(json.loads(e["data"]) for e in events if e["type"] == "message")
    print("  rebuilt text equals the model's answer:", rebuilt == ANSWER)
    print("  paragraph-break token on the wire:", repr(next(text for _, text in chunks if "event." in text)))

    print("GET /slow (the model goes quiet for 17 seconds):")
    _, _, chunks = labkit.read_events(8102, "/slow")
    for seconds, text in chunks:
        print(f"  t={seconds:5.1f}s  {text!r}")
finally:
    labkit.stop(proc)

python demo_builtin.py

GET /stream -> 200, 40 events
  content-type: text/event-stream; charset=utf-8
  cache-control: no-cache
  x-accel-buffering: no
  transfer-encoding: chunked
  first three chunks:
    'data: "Server-sent "\nid: 0\n\n'
    'data: "events "\nid: 1\n\n'
    'data: "let "\nid: 2\n\n'
  last chunk: 'event: done\ndata: [DONE]\n\n'
  rebuilt text equals the model's answer: True
  paragraph-break token on the wire: 'data: "event.\\n\\n"\nid: 24\n\n'
GET /slow (the model goes quiet for 17 seconds):
  t=  0.0s  'data: "first"\n\n'
  t= 15.0s  ': ping\n\n'
  t= 17.0s  'data: "second"\n\n'

Read the output from top to bottom. The headers show what FastAPI does for you without being asked: cache-control: no-cache keeps proxies and browsers from caching the stream, and x-accel-buffering: no is a hint for nginx. The documentation lists three such defaults: a ping comment every 15 seconds of silence “to prevent some proxies from closing the connection”, Cache-Control: no-cache, and “Set a special header X-Accel-Buffering: no to prevent buffering in some proxies like Nginx.” The first line of data shows the JSON encoding: each token is wrapped in double quotes and the paragraph token carries \n\n as an escape (the printout shows each backslash doubled), so it cannot break the event. The rebuilt text matches the answer. The /slow run shows the ping arriving at 15.0 seconds, long before the second token at 17.0 seconds.

The ping idea comes from the standard itself: “Legacy proxy servers are known to, in certain cases, drop HTTP connections after a short timeout. To protect against such proxy servers, authors can include a comment line (one starting with a ‘:’ character) every 15 seconds or so.” In FastAPI 0.162.0 the code reads your generator in a separate task and feeds a small queue, while a second task inserts : ping whenever nothing has arrived for 15 seconds. That is why a model that goes quiet in the middle of its answer does not stop the pings. The interval lives in a private module-level name, _PING_INTERVAL, which FastAPI’s own tests patch. Step 8 does the same in a lab, and you should not rely on it in production code.

Step 4: Read the stream in a browser

The page below opens an EventSource and appends every token to a <pre> element. It also writes timestamps into window.log so a script can read what happened. Four handlers matter. onopen fires when the connection is established. onmessage fires for each token; it decodes the JSON and appends the text. The listener for the done event closes the source. onerror fires when something goes wrong, and it also fires when a response simply ends, a fact that the second experiment below turns into a bug.

<!doctype html>
<meta charset="utf-8">
<title>SSE demo</title>
<pre id="out"></pre>
<script>
const params = new URLSearchParams(location.search);
const closeOnDone = params.get("close") !== "0";
const out = document.getElementById("out");
const t0 = performance.now();
const ms = () => Math.round(performance.now() - t0);
window.log = [];
window.finished = false;

const source = new EventSource("/stream");
source.onopen = () => window.log.push(["open", ms()]);
source.onmessage = (e) => {
  out.textContent += JSON.parse(e.data);
  window.log.push(["token", ms(), e.lastEventId]);
};
source.addEventListener("done", () => {
  window.log.push(["done", ms()]);
  window.finished = true;
  if (closeOnDone) source.close(); // without this the browser reconnects when the response ends
});
source.onerror = () => window.log.push(["error", ms(), source.readyState]);
</script>

The script below uses Playwright to drive Edge without a window. It loads the page, waits for the done event, and reads the log. Then it loads the page again with ?close=0, which skips the source.close() call and so behaves like a page written without the done event, and watches for 12 seconds.

"""demo_browser.py: load the page in headless Edge and measure what a user would see."""
import time

from playwright.sync_api import sync_playwright

import labkit
from fake_model import ANSWER

proc = labkit.start("app_v2:app", 8103)
try:
    with sync_playwright() as p:
        browser = p.chromium.launch(channel="msedge", headless=True)
        print("browser:", browser.version)

        page = browser.new_page()
        requests = []
        page.on("request", lambda r: requests.append(r.url) if r.url.endswith("/stream") else None)
        page.goto("http://127.0.0.1:8103/")
        page.wait_for_function("window.finished === true", timeout=20000)
        log = page.evaluate("window.log")
        text = page.evaluate("document.getElementById('out').textContent")
        tokens = [entry for entry in log if entry[0] == "token"]
        print(f"page closes the source on done: {len(tokens)} tokens, first at {tokens[0][1]} ms, last at {tokens[-1][1]} ms")
        print("  last event id seen:", tokens[-1][2], "| text equals the answer:", text == ANSWER)
        print("  requests to /stream:", len(requests))

        print("page that ignores the done event, watched for 12 seconds:")
        labkit.call(8103, "POST", "/stats/reset")
        page2 = browser.new_page()
        requests2 = []
        page2.on("request", lambda r: requests2.append(r.url) if r.url.endswith("/stream") else None)
        page2.goto("http://127.0.0.1:8103/?close=0")
        time.sleep(12)
        log2 = page2.evaluate("window.log")
        text2 = page2.evaluate("document.getElementById('out').textContent")
        print("  requests to /stream:", len(requests2))
        first_error = next(e for e in log2 if e[0] == "error")
        reopened = next(e for e in log2 if e[0] == "open" and e[1] > first_error[1])
        print("  error events:", sum(1 for entry in log2 if entry[0] == "error"), "| readyState at the first error:", first_error[2],
              f"| the browser waited {reopened[1] - first_error[1]} ms before reconnecting")
        print("  characters on screen:", len(text2), "against", len(ANSWER), "in one answer")
        print("  tokens the model produced:", labkit.call(8103, "GET", "/stats")["produced"])
        browser.close()
finally:
    labkit.stop(proc)

python demo_browser.py

browser: 156.0.4314.50
page closes the source on done: 39 tokens, first at 68 ms, last at 2407 ms
  last event id seen: 38 | text equals the answer: True
  requests to /stream: 1
page that ignores the done event, watched for 12 seconds:
  requests to /stream: 3
  error events: 2 | readyState at the first error: 0 | the browser waited 3012 ms before reconnecting
  characters on screen: 506 against 216 in one answer
  tokens the model produced: 92

The first run is the happy path. The first token reached the page 68 milliseconds after it loaded and the last one at 2,407 milliseconds, a bit more than the 1,950 milliseconds that 39 sleeps of 50 milliseconds add up to, probably because each sleep wakes up slightly late. The last event ID the browser saw was 38, the index of the 39th token, and the text on screen equals the answer. Only one request was made.

The second run is the trap. With no done handler that closes the source, the page made 3 requests in 12 seconds. The first error event reported readyState 0, which is CONNECTING: the browser was not giving up, it was about to retry, and it waited 3,012 milliseconds first, the “few seconds” the standard mentions. Each retry restarted the answer: the screen held 506 characters against the 216 of a single answer, and the model had produced 92 tokens instead of 39. In a real app that second and third answer is a second and third bill.

The fix has two halves. On the server, end every stream with an explicit final event (done here) so the client can tell a finished answer from a dropped connection. On the client, close the source when that event arrives. If you cannot change the client, the standard gives you a server-side exit too: answer the reconnection request with HTTP 204 No Content and the browser stops.

Step 5: Stop the model when the user leaves

When a user closes the tab, the browser closes the connection. Whether your model stops depends entirely on how it is wired to the response. This app has four endpoints that stream the same fake model in four different ways:

"""app_cancel.py: four ways to connect a model to an SSE response, and what each does when the user leaves."""
import asyncio
import gc
from collections.abc import AsyncIterable, Iterable

from fastapi import FastAPI
from fastapi.sse import EventSourceResponse, ServerSentEvent

import fake_model

app = FastAPI()
background: set[asyncio.Task] = set()  # the event loop only keeps weak references to tasks


@app.get("/stats")
def stats() -> dict:
    return fake_model.STATS


@app.post("/stats/reset")
def reset() -> dict:
    return fake_model.reset_stats()


@app.post("/gc")
def collect_garbage() -> dict:
    return {"collected": gc.collect(), **fake_model.STATS}


@app.get("/async", response_class=EventSourceResponse)
async def stream_async() -> AsyncIterable[ServerSentEvent]:
    async for token in fake_model.generate():
        yield ServerSentEvent(data=token)


@app.get("/sync", response_class=EventSourceResponse)
def stream_sync() -> Iterable[ServerSentEvent]:
    for token in fake_model.generate_blocking():
        yield ServerSentEvent(data=token)


async def pump(queue: asyncio.Queue) -> None:
    """Run the model in its own task and hand tokens to the response through a queue."""
    async for token in fake_model.generate():
        await queue.put(token)
    await queue.put(None)


def spawn(coro) -> asyncio.Task:
    task = asyncio.create_task(coro)
    background.add(task)
    task.add_done_callback(background.discard)
    return task


@app.get("/task", response_class=EventSourceResponse)
async def stream_task() -> AsyncIterable[ServerSentEvent]:
    queue: asyncio.Queue = asyncio.Queue()
    spawn(pump(queue))  # nobody ever cancels this task
    while (token := await queue.get()) is not None:
        yield ServerSentEvent(data=token)


@app.get("/task-fixed", response_class=EventSourceResponse)
async def stream_task_fixed() -> AsyncIterable[ServerSentEvent]:
    queue: asyncio.Queue = asyncio.Queue()
    task = spawn(pump(queue))
    try:
        while (token := await queue.get()) is not None:
            yield ServerSentEvent(data=token)
    finally:
        task.cancel()  # runs when the response is cancelled because the client left

/async reads the model directly in an async generator. /sync uses a plain def generator, which FastAPI runs in a worker thread. /task starts the model in its own task and passes tokens through a queue, a pattern you will want in Step 6; it has no cleanup. /task-fixed is the same code with a finally that cancels the task. The spawn helper keeps a reference to every task because, as the Python documentation puts it, “The event loop only keeps weak references to tasks.” The /gc route is a debugging aid for one experiment; never expose something like it in production.

"""demo_cancel.py: close the connection after 5 tokens and count what the model kept producing."""
import time

import labkit

PORT = 8104
proc = labkit.start("app_cancel:app", PORT)
try:
    print(f"{'endpoint':<12} {'produced':>8} {'finished':>8} {'stopped':>8}   (the model has 39 tokens; the client leaves after 5)")
    for path in ("/async", "/sync", "/task", "/task-fixed"):
        labkit.call(PORT, "POST", "/stats/reset")
        labkit.read_events(PORT, path, stop_after=5)
        time.sleep(3)  # longer than the 2 seconds the whole answer takes
        stats = labkit.call(PORT, "GET", "/stats")
        print(f"{path:<12} {stats['produced']:>8} {stats['finished']:>8} {stats['stopped']:>8}")
        if path == "/sync":
            after = labkit.call(PORT, "POST", "/gc")
            print(f"{'  after gc.collect():':<12} {after['produced']:>8} {after['finished']:>8} {after['stopped']:>8}")
finally:
    labkit.stop(proc)

python demo_cancel.py

endpoint     produced finished  stopped   (the model has 39 tokens; the client leaves after 5)
/async              5        0        1
/sync               6        0        0
  after gc.collect():        6        0        1
/task              39        1        0
/task-fixed         5        0        1

Each row is one user who leaves after five tokens, followed by three seconds of waiting, which is longer than the two seconds the whole answer takes.

• /async produced 5 tokens and stopped 1 generator. FastAPI cancelled the generator right after the client left, and the finally block ran. This is the behavior you want.
• /sync produced 6 tokens, one more than the user saw, because the worker thread was already waiting out the next token and finished it. Worse, stopped stayed 0: three seconds later the generator’s finally block had not run. It ran only after I forced a garbage collection, which is what the second line shows. I did not dig into the reason; the likely explanation is that the abandoned generator stays referenced until the garbage collector finds it. The practical rule is not to rely on a sync generator for prompt cleanup, such as closing a connection to your model server.
• /task produced all 39 tokens and finished 1 generator. The response was cancelled, but nobody told the task, so the model worked to the end for an empty room.
• /task-fixed produced 5 and stopped 1, the same as /async. The finally in the endpoint runs when the response is cancelled, and it cancels the task.

The general rule: whatever code starts expensive work must also be the code that cancels it when the response goes away, and the endpoint’s finally block is the natural place. With a fake model the waste is a few tokens. With a real model it is GPU time, which Step 7 measures.

Step 6: Resume after a dropped connection

Phones switch networks, laptops sleep and proxies restart. When that happens mid-answer, the browser reconnects by itself, as you saw, but your server decides whether the answer continues or starts over. Resuming needs four ideas that fit in four sentences.

1. A job is created with a POST and gets an unguessable ID. The model runs in its own task and appends every token to a list, whether or not anyone is watching.
2. A viewer attaches with GET /runs/<id>/events. Each token is sent with an id: equal to its position in the list.
3. When the browser reconnects it sends Last-Event-ID: 6, and the server starts at token 7. The FastAPI docs describe the same pattern: “When a browser reconnects after a connection drop, it sends the last received id in the Last-Event-ID header.”
4. If nobody has been watching for a grace period, the job is cancelled, so abandoned chats do not run forever.

The server

The code is longer than anything so far, so read it in pieces. Run holds one answer: its tokens, its status, how many viewers are attached, and a Condition, which is a lock with a waiting room. Viewers wait in the room until there is a new token or the run has ended, and the producer wakes them all with notify_all. In produce, the status starts as cancelled and is replaced only when the loop ends normally or an ordinary exception happens. A cancellation is not an Exception, so it skips that handler and the status stays cancelled.

attach and detach count viewers. When the last one leaves a running job, detach starts a timer task, the reaper, that cancels the job after GRACE seconds; if someone attaches in time, attach cancels the timer. get_run is a dependency that raises a 404 for an unknown job. It has to be a dependency, because FastAPI resolves dependencies before it starts the response, so the client gets a real HTTP 404. Raising the same exception inside the generator does not work: the /inside route in app_slow.py (Step 8) does exactly that, and a test in Step 9 shows the client still receives HTTP 200 and an empty stream while the server logs a traceback.

The endpoint reads Last-Event-ID through a header parameter named last_event_id, because FastAPI turns underscores into hyphens. It sends a comment event first that carries retry: 1000, so a dropped browser reconnects after one second instead of three. Then it loops: wait for new tokens, send each with its index as the id, and finish with done or with a failure event that carries the reason. The two query parameters are demo switches: drop_after ends the first connection after that many events, which makes the browser reconnect just as it would after a dropped network, and resume=false makes the server ignore Last-Event-ID, so you can see what the browser does to a server that never learned about it.

"""app_v3.py: the model runs on its own, so a dropped stream can resume where it stopped (FastAPI 0.162.0)."""
import asyncio
import os
import secrets
from collections.abc import AsyncIterable
from typing import Annotated

from fastapi import Depends, FastAPI, Header, HTTPException
from fastapi.responses import FileResponse
from fastapi.sse import EventSourceResponse, ServerSentEvent
from pydantic import BaseModel

import fake_model

GRACE = float(os.environ.get("GRACE", "10"))  # seconds a run may keep going with nobody watching
RETRY_MS = 1000  # how long the browser waits before it reconnects
app = FastAPI()
RUNS: dict[str, "Run"] = {}
background: set[asyncio.Task] = set()  # the event loop only keeps weak references to tasks


def spawn(coro) -> asyncio.Task:
    task = asyncio.create_task(coro)
    background.add(task)
    task.add_done_callback(background.discard)
    return task


class Run:
    """One model answer: the tokens so far, who is watching, and how it ended."""

    def __init__(self, fail_after: int | None):
        self.tokens: list[str] = []
        self.status = "running"  # becomes finished, failed or cancelled
        self.error = ""
        self.watchers = 0
        self.attaches: list[str | None] = []  # the Last-Event-ID of every connection, for the demos
        self.changed = asyncio.Condition()
        self.reaper: asyncio.Task | None = None
        self.task = spawn(self.produce(fail_after))

    async def produce(self, fail_after: int | None) -> None:
        status = "cancelled"  # stays this way if the task is cancelled
        try:
            async for token in fake_model.generate(fail_after=fail_after):
                async with self.changed:
                    self.tokens.append(token)
                    self.changed.notify_all()
            status = "finished"
        except Exception as exc:
            status, self.error = "failed", str(exc)
        finally:
            async with self.changed:
                self.status = status
                self.changed.notify_all()

    def attach(self, last_event_id: int | None) -> None:
        self.watchers += 1
        self.attaches.append(None if last_event_id is None else str(last_event_id))
        if self.reaper:  # somebody came back in time
            self.reaper.cancel()
            self.reaper = None

    def detach(self) -> None:
        self.watchers -= 1
        if self.watchers == 0 and self.status == "running":
            self.reaper = spawn(self.reap())

    async def reap(self) -> None:
        await asyncio.sleep(GRACE)  # nobody came back: stop paying for tokens
        self.task.cancel()


class NewRun(BaseModel):
    prompt: str = ""
    fail_after: int | None = None  # demo switch: crash the fake backend at this token


def get_run(run_id: str) -> Run:
    run = RUNS.get(run_id)
    if run is None:
        raise HTTPException(status_code=404, detail="unknown run")
    return run


@app.get("/")
def page() -> FileResponse:
    return FileResponse("index3.html")


@app.get("/stats")
def stats() -> dict:
    return fake_model.STATS


@app.post("/stats/reset")
def reset() -> dict:
    return fake_model.reset_stats()


@app.post("/runs")
async def create_run(body: NewRun) -> dict:
    run_id = secrets.token_urlsafe(16)  # unguessable, and in real life also tied to the signed-in user
    RUNS[run_id] = Run(body.fail_after)
    return {"id": run_id}


@app.get("/runs/{run_id}")
async def show_run(run: Annotated[Run, Depends(get_run)]) -> dict:
    return {"status": run.status, "tokens": len(run.tokens), "watchers": run.watchers, "attaches": run.attaches}


@app.get("/runs/{run_id}/events", response_class=EventSourceResponse)
async def run_events(
    run: Annotated[Run, Depends(get_run)],
    last_event_id: Annotated[int | None, Header()] = None,
    drop_after: int | None = None,  # demo switch: end the first connection after this many events
    resume: bool = True,  # demo switch: False ignores Last-Event-ID, like a server that never learned about it
) -> AsyncIterable[ServerSentEvent]:
    next_index = last_event_id + 1 if resume and last_event_id is not None else 0
    first_connection = last_event_id is None
    run.attach(last_event_id)
    try:
        yield ServerSentEvent(comment="stream open", retry=RETRY_MS)
        while True:
            async with run.changed:
                await run.changed.wait_for(lambda: len(run.tokens) > next_index or run.status != "running")
                batch, status = run.tokens[next_index:], run.status
            for token in batch:
                yield ServerSentEvent(data=token, id=str(next_index))
                next_index += 1
                if first_connection and drop_after is not None and next_index >= drop_after:
                    return  # the response just ends, which the browser cannot tell from a dropped network
            if status != "running" and next_index >= len(run.tokens):
                break
        if status == "finished":
            yield ServerSentEvent(raw_data="[DONE]", event="done")
        else:
            message = run.error or f"the run was {status}"
            yield ServerSentEvent(data={"message": message}, event="failure")
    finally:
        run.detach()

The page

The page first POSTs to /runs to get an ID, then opens the EventSource on that job. It listens for two named events. I called the failure event failure, not error, on purpose. The standard says that if the event type buffer is set, the dispatched event takes that name, so a server event called error would be dispatched as an error event and land in the same onerror handler that reports connection trouble. The error event that an EventSource fires does not carry the HTTP status of a failed stream, so anything your page needs to know about a model failure must travel as an event.

<!doctype html>
<meta charset="utf-8">
<title>Resumable SSE demo</title>
<pre id="out"></pre>
<script>
const params = new URLSearchParams(location.search);
const out = document.getElementById("out");
const t0 = performance.now();
const ms = () => Math.round(performance.now() - t0);
window.log = [];
window.finished = false;

async function main() {
  const body = {prompt: "hello"};
  if (params.has("fail_after")) body.fail_after = Number(params.get("fail_after"));
  const res = await fetch("/runs", {method: "POST", headers: {"Content-Type": "application/json"}, body: JSON.stringify(body)});
  const {id} = await res.json();
  window.runId = id;

  const pass = new URLSearchParams();
  for (const name of ["drop_after", "resume"]) if (params.has(name)) pass.set(name, params.get(name));
  const source = new EventSource(`/runs/${id}/events?${pass}`);
  source.onopen = () => window.log.push(["open", ms()]);
  source.onmessage = (e) => {
    out.textContent += JSON.parse(e.data);
    window.log.push(["token", ms(), e.lastEventId]);
  };
  source.addEventListener("done", () => {
    window.log.push(["done", ms()]);
    window.finished = true;
    source.close();
  });
  source.addEventListener("failure", (e) => {
    window.log.push(["failure", ms(), JSON.parse(e.data).message]);
    window.finished = true;
    source.close();
  });
  source.onerror = () => window.log.push(["error", ms(), source.readyState]);
}
main();
</script>

Try it by hand and in a browser

demo_resume.py does three experiments. First, with a raw HTTP client, it reads seven tokens, leaves, and reconnects with Last-Event-ID: 6 set by hand. Second, in Edge, it lets the server end the first connection after seven events, once with resume working and once with resume=false. Third, it crashes the backend after ten tokens.

"""demo_resume.py: drop a stream after 7 events and resume it, first by hand and then in a real browser."""
import json

from playwright.sync_api import sync_playwright

import labkit
import sse_parse
from fake_model import ANSWER

PORT = 8105
BASE = f"http://127.0.0.1:{PORT}"
proc = labkit.start("app_v3:app", PORT)


def tokens_of(chunks):
    events = sse_parse.parse("".join(text for _, text in chunks))
    return [e for e in events if e["type"] == "message"]


try:
    print("--- by hand, with a raw HTTP client")
    run_id = labkit.call(PORT, "POST", "/runs", {"prompt": "hi"})["id"]
    _, _, first = labkit.read_events(PORT, f"/runs/{run_id}/events", stop_after=8)  # comment + 7 tokens, then leave
    one = tokens_of(first)
    print("first two chunks on the wire:", repr(first[0][1]), repr(first[1][1]))
    print(f"first connection:  {len(one)} tokens, ids {one[0]['id']} to {one[-1]['id']}, then the client left")
    _, _, second = labkit.read_events(PORT, f"/runs/{run_id}/events", headers={"Last-Event-ID": one[-1]["id"]})
    two = tokens_of(second)
    print(f"second connection: {len(two)} tokens, ids {two[0]['id']} to {two[-1]['id']}, last event type {sse_parse.parse(second[-1][1])[0]['type']}")
    ids = [int(e["id"]) for e in one + two]
    text = "".join(json.loads(e["data"]) for e in one + two)
    print("ids are 0..38 with no gap or repeat:", ids == list(range(39)), "| text equals the answer:", text == ANSWER)
    print("what the server saw:", labkit.call(PORT, "GET", f"/runs/{run_id}"))

    print("--- in headless Edge, the server ends the first connection after 7 events")
    with sync_playwright() as p:
        browser = p.chromium.launch(channel="msedge", headless=True)
        cases = (("resume on", "?drop_after=7"), ("server ignores Last-Event-ID", "?drop_after=7&resume=0"))
        for label, query in cases:
            page = browser.new_page()
            event_requests = []
            page.on("request", lambda r: event_requests.append(r.url) if "/events" in r.url else None)
            page.goto(f"{BASE}/{query}")
            page.wait_for_function("window.finished === true", timeout=20000)
            log = page.evaluate("window.log")
            text = page.evaluate("document.getElementById('out').textContent")
            token_ids = [int(e[2]) for e in log if e[0] == "token"]
            repeats = sorted({i for i in token_ids if token_ids.count(i) > 1})
            err = next(e for e in log if e[0] == "error")
            reopened = next(e for e in log if e[0] == "open" and e[1] > err[1])
            run = labkit.call(PORT, "GET", f"/runs/{page.evaluate('window.runId')}")
            print(f"{label}:")
            print(f"  requests to /events: {len(event_requests)} | error readyState: {err[2]} | reconnect delay: {reopened[1] - err[1]} ms")
            print(f"  tokens shown: {len(token_ids)} | repeated ids: {repeats if repeats else 'none'} | text equals the answer: {text == ANSWER}")
            print(f"  Last-Event-ID values the server saw: {run['attaches']}")

        print("--- the backend crashes after 10 tokens")
        page = browser.new_page()
        event_requests = []
        page.on("request", lambda r: event_requests.append(r.url) if "/events" in r.url else None)
        page.goto(f"{BASE}/?fail_after=10")
        page.wait_for_function("window.finished === true", timeout=20000)
        log = page.evaluate("window.log")
        print("  tokens shown:", sum(1 for e in log if e[0] == "token"), "| last log entry:", log[-1][0], log[-1][2], "| requests to /events:", len(event_requests))
        browser.close()
finally:
    labkit.stop(proc)

python demo_resume.py

--- by hand, with a raw HTTP client
first two chunks on the wire: ': stream open\nretry: 1000\n\n' 'data: "Server-sent "\nid: 0\n\n'
first connection:  7 tokens, ids 0 to 6, then the client left
second connection: 32 tokens, ids 7 to 38, last event type done
ids are 0..38 with no gap or repeat: True | text equals the answer: True
what the server saw: {'status': 'finished', 'tokens': 39, 'watchers': 0, 'attaches': [None, '6']}
--- in headless Edge, the server ends the first connection after 7 events
resume on:
  requests to /events: 2 | error readyState: 0 | reconnect delay: 1005 ms
  tokens shown: 39 | repeated ids: none | text equals the answer: True
  Last-Event-ID values the server saw: [None, '6']
server ignores Last-Event-ID:
  requests to /events: 2 | error readyState: 0 | reconnect delay: 1005 ms
  tokens shown: 46 | repeated ids: [0, 1, 2, 3, 4, 5, 6] | text equals the answer: False
  Last-Event-ID values the server saw: [None, '6']
--- the backend crashes after 10 tokens
  tokens shown: 10 | last log entry: failure model backend crashed | requests to /events: 1

The first two lines show the wire: a comment event that sets retry: 1000, then the first token with id: 0. By hand, the second connection started at id 7 and ended with the done event, the 39 ids form an unbroken run from 0 to 38, and the server’s own record shows two attaches, the first with no Last-Event-ID and the second with '6'.

In Edge, the browser did the same by itself. It made 2 requests, waited 1,005 milliseconds (the retry: 1000 we set, against 3,012 without it), and the page ended with all 39 tokens, no repeats and the right text. The line below it is the bug: when the server ignores Last-Event-ID, the header still arrives ('6' shows in the server’s record), but the answer restarts at token 0 and the page ends with 46 tokens, the ids 0 to 6 twice, and the wrong text. The browser did its half of the deal; the server must do the other half. The last line shows the failure path: ten tokens, one failure event and a single request, because the page closed the source instead of letting it retry.

What happens when nobody comes back

The grace period decides how much you pay for being able to resume. In this demo it is one second, set through an environment variable, so the output arrives quickly. The default in app_v3.py is 10 seconds.

"""demo_grace.py: what happens to the model when the user leaves for good, and when they come back in time."""
import os
import time

import labkit

PORT = 8106
os.environ["GRACE"] = "1"  # one second, so the demo is quick; the app default is 10
proc = labkit.start("app_v3:app", PORT)
try:
    print("the user leaves after 5 tokens and never returns (GRACE=1):")
    labkit.call(PORT, "POST", "/stats/reset")
    run_id = labkit.call(PORT, "POST", "/runs", {"prompt": "hi"})["id"]
    labkit.read_events(PORT, f"/runs/{run_id}/events", stop_after=6)  # comment + 5 tokens
    for wait in (0.5, 2.0):
        time.sleep(wait)
        run = labkit.call(PORT, "GET", f"/runs/{run_id}")
        print(f"  {run['status']:<9} tokens so far {run['tokens']:>2}  watchers {run['watchers']}  model stats {labkit.call(PORT, 'GET', '/stats')}")
    _, _, chunks = labkit.read_events(PORT, f"/runs/{run_id}/events", headers={"Last-Event-ID": "4"})
    print("  a late reconnect gets the rest of what was produced, then:", repr(chunks[-1][1]))

    print("the user comes back inside the grace period:")
    labkit.call(PORT, "POST", "/stats/reset")
    run_id = labkit.call(PORT, "POST", "/runs", {"prompt": "hi"})["id"]
    labkit.read_events(PORT, f"/runs/{run_id}/events", stop_after=6)
    time.sleep(0.3)
    labkit.read_events(PORT, f"/runs/{run_id}/events", headers={"Last-Event-ID": "4"})
    run = labkit.call(PORT, "GET", f"/runs/{run_id}")
    print(f"  {run['status']}, {run['tokens']} tokens, model stats {labkit.call(PORT, 'GET', '/stats')}")
finally:
    labkit.stop(proc)

python demo_grace.py

the user leaves after 5 tokens and never returns (GRACE=1):
  running   tokens so far 13  watchers 0  model stats {'produced': 13, 'finished': 0, 'stopped': 0}
  cancelled tokens so far 21  watchers 0  model stats {'produced': 21, 'finished': 0, 'stopped': 1}
  a late reconnect gets the rest of what was produced, then: 'event: failure\ndata: {"message": "the run was cancelled"}\n\n'
the user comes back inside the grace period:
  finished, 39 tokens, model stats {'produced': 39, 'finished': 1, 'stopped': 0}

The user left after five tokens. Half a second later the job was still running (13 tokens so far, nobody watching). After the grace period it was cancelled, with 21 tokens produced and the model’s stopped counter at 1, so the cost of the grace period was the tokens made while waiting. A reconnect after that gets whatever was produced, then a failure event that says the run was cancelled. In the second experiment the user came back within the grace period and the job ran to the end with all 39 tokens. The trade-off is yours to set: at the 55 tokens per second a small real model produced in Step 7 (400 tokens in 7.3 seconds), a 10 second grace period can burn about 550 tokens for a chat that never comes back.

Limits of this design

The jobs live in one Python dictionary in one process. With several uvicorn workers, a reconnect can land on a worker that has never heard of the job; the 404 it gets is a non-200 status, which makes the browser fail the connection for good. To scale out you need either sticky routing or a shared store for the token list, which this tutorial did not build or test. The dictionary is also never cleaned, so a real service needs a time limit on finished jobs. And the job ID is only a random string: in a real app every job must belong to a signed-in user and every request must check that ownership, the same class of bug covered in preventing broken object level authorization in FastAPI. If your page and API live on different origins, see how to fix CORS errors in FastAPI, since EventSource follows the same rules.

Step 7: Swap in a real model with Ollama

The fake model taught you the mechanics. A real one adds one question: does stopping your endpoint stop the model? This step answers it with Ollama and the qwen2.5:1.5b model. Download the model once, then start the server so that its log goes to a file next to your scripts (this redirect works in cmd.exe, bash and zsh):

ollama pull qwen2.5:1.5b
ollama serve > ollama_serve.log 2>&1

Ollama’s streaming API answers with one JSON object per line. Its documentation says: “These responses are provided in the newline-delimited JSON format (i.e. the application/x-ndjson content type).” The helper ollama_tokens reads those lines with httpx and yields each token; leaving its async with block, for any reason, closes the connection to Ollama. /chat uses it directly. /chat-leaky repeats the mistake of /task: a forgotten task reads from Ollama while the response forgets about it. Temperature 0 and a limit of 400 tokens make the runs comparable. Note that this endpoint is a POST: SSE works with any HTTP method on the server side, as the FastAPI docs say, and only the browser’s EventSource is limited to GET.

"""app_ollama.py: stream a real model (Ollama) through the same kind of SSE response."""
import asyncio
import json
from collections.abc import AsyncIterable

import httpx
from fastapi import FastAPI
from fastapi.sse import EventSourceResponse, ServerSentEvent
from pydantic import BaseModel

OLLAMA = "http://127.0.0.1:11434"
MODEL = "qwen2.5:1.5b"
app = FastAPI()
background: set[asyncio.Task] = set()


class Prompt(BaseModel):
    text: str


async def ollama_tokens(text: str) -> AsyncIterable[str]:
    """Yield the model's tokens. Ollama streams one JSON object per line (NDJSON) until one says done."""
    body = {"model": MODEL, "messages": [{"role": "user", "content": text}], "stream": True,
            "options": {"temperature": 0, "num_predict": 400}}
    async with httpx.AsyncClient(timeout=None) as client:
        async with client.stream("POST", f"{OLLAMA}/api/chat", json=body) as upstream:
            async for line in upstream.aiter_lines():
                chunk = json.loads(line)
                if chunk.get("done"):
                    return
                yield chunk["message"]["content"]  # leaving this block closes the connection to Ollama


@app.post("/chat", response_class=EventSourceResponse)
async def chat(prompt: Prompt) -> AsyncIterable[ServerSentEvent]:
    i = 0
    async for token in ollama_tokens(prompt.text):
        yield ServerSentEvent(data=token, id=str(i))
        i += 1
    yield ServerSentEvent(raw_data="[DONE]", event="done")


async def pump(text: str, queue: asyncio.Queue) -> None:
    async for token in ollama_tokens(text):
        await queue.put(token)
    await queue.put(None)


@app.post("/chat-leaky", response_class=EventSourceResponse)
async def chat_leaky(prompt: Prompt) -> AsyncIterable[ServerSentEvent]:
    queue: asyncio.Queue = asyncio.Queue()
    task = asyncio.create_task(pump(prompt.text, queue))  # the same mistake as /task in app_cancel.py
    background.add(task)
    task.add_done_callback(background.discard)
    i = 0
    while (token := await queue.get()) is not None:
        yield ServerSentEvent(data=token, id=str(i))
        i += 1
    yield ServerSentEvent(raw_data="[DONE]", event="done")

The script runs a full answer first and then, for each endpoint, leaves after 10 tokens and reads the duration Ollama logged for that request. Ollama prints one log line per request, including how long it took. If you run Ollama some other way, its log lives elsewhere; set the OLLAMA_LOG environment variable to its path.

"""demo_ollama.py: stream a real model, then leave early and read Ollama's own log to see who kept generating."""
import json
import os
import re
import time

import httpx

import labkit
import sse_parse

PORT = 8130
OLLAMA_LOG = os.environ.get("OLLAMA_LOG", os.path.join(labkit.HERE, "ollama_serve.log"))
PROMPT = {"text": "Write a long story about a lighthouse keeper who finds a message in a bottle."}


def last_chat_duration() -> str:
    """The duration Ollama logged for its most recent POST /api/chat."""
    lines = [l for l in open(OLLAMA_LOG, encoding="utf-8", errors="replace") if '"/api/chat"' in l]
    return re.search(r"\|\s*([\d.]+(?:ms|s|m[\d.]+s))\s*\|", lines[-1]).group(1)


httpx.post("http://127.0.0.1:11434/api/chat", json={"model": "qwen2.5:1.5b", "messages": [{"role": "user", "content": "hi"}],
           "stream": False, "options": {"num_predict": 1}}, timeout=120)  # load the model so the timings below are fair
proc = labkit.start("app_ollama:app", PORT)
try:
    print("a full answer through /chat:")
    started = time.perf_counter()
    _, headers, chunks = labkit.read_events(PORT, "/chat", method="POST", body=PROMPT)
    full_seconds = time.perf_counter() - started
    tokens = [json.loads(e["data"]) for e in sse_parse.parse("".join(text for _, text in chunks)) if e["type"] == "message"]
    print(f"  first token after {chunks[0][0]:.2f} s, {len(tokens)} tokens in {full_seconds:.1f} s, first words: {''.join(tokens[:8])!r}")
    print(f"  Ollama logged {last_chat_duration()} for the request")

    for path, label in (("/chat", "proper cancellation"), ("/chat-leaky", "model pumped by a forgotten task")):
        started = time.perf_counter()
        labkit.read_events(PORT, path, method="POST", body=PROMPT, stop_after=10)
        left_at = time.perf_counter() - started
        time.sleep(12)  # long enough for a full answer to finish
        print(f"the user leaves after 10 tokens ({left_at:.2f} s) through {path} ({label}): Ollama logged {last_chat_duration()}")
finally:
    labkit.stop(proc)

python demo_ollama.py

a full answer through /chat:
  first token after 0.47 s, 400 tokens in 7.3 s, first words: 'The old lighthouse stood tall on the'
  Ollama logged 6.9926562s for the request
the user leaves after 10 tokens (0.75 s) through /chat (proper cancellation): Ollama logged 330.6639ms
the user leaves after 10 tokens (0.89 s) through /chat-leaky (model pumped by a forgotten task): Ollama logged 7.1854155s

A full answer took 7.3 seconds for 400 tokens, with the first token after 0.47 seconds, and Ollama logged 6.99 seconds for it. When the user left after 10 tokens, the properly cancelled endpoint ended the Ollama request after 330.66 milliseconds. The leaky endpoint let Ollama run for 7.19 seconds, the whole answer, for a user who had left after less than a second. That is the difference between stopping the model and paying for GPU time that nobody reads, and it costs one finally block to avoid.

Step 8: Check the proxy and the browser’s connection limit

In production your app rarely faces the browser directly. A reverse proxy such as nginx, a load balancer or a CDN sits in between, and each of them has opinions about long responses. The browser has its own.

nginx in front of the stream

nginx can hold back a proxied response in buffers before it passes the data on. Its documentation says that “When buffering is disabled, the response is passed to a client synchronously, immediately as it is received”, and that an upstream can switch buffering per response with an X-Accel-Buffering header. FastAPI sets that header for you; a hand-written endpoint does not. The configuration below defines five routes on port 8180: a hand-written stream through default settings, the built-in stream, the hand-written stream with proxy_buffering off, and two routes with a deliberately short proxy_read_timeout of 3 seconds. The docs describe that timeout as set “only between two successive read operations, not for the transmission of the whole response”, and say “If the proxied server does not transmit anything within this time, the connection is closed.” The nginx default is 60 seconds.

worker_processes 1;
error_log logs/error.log warn;
pid logs/nginx.pid;
events { worker_connections 64; }

http {
    access_log off;
    server {
        listen 127.0.0.1:8180;

        # hand-written SSE, which sends no X-Accel-Buffering header; nginx talks HTTP/1.0 to the app by default
        location /v1/ { proxy_pass http://127.0.0.1:8107/; }

        # FastAPI's built-in SSE, which sends X-Accel-Buffering: no
        location /v2/ { proxy_pass http://127.0.0.1:8108/; }

        # the same hand-written app, with buffering switched off for this location
        location /v1-off/ {
            proxy_pass http://127.0.0.1:8107/;
            proxy_buffering off;
        }

        # a proxy that hangs up on silent connections after 3 seconds (nginx's default is 60)
        location /idle/ {
            proxy_pass http://127.0.0.1:8109/;
            proxy_read_timeout 3s;
        }
        location /idle-ping/ {
            proxy_pass http://127.0.0.1:8110/;
            proxy_read_timeout 3s;
        }
    }
}

app_slow.py is a small app whose model goes quiet for six seconds after its first token. It reads its ping interval from an environment variable by patching FastAPI’s private _PING_INTERVAL (the shortcut I warned about in Step 3), so the lab can compare the default 15 seconds with 1 second. The same file serves a blank page for the second experiment in this step, and the /inside route that Step 9 tests.

"""app_slow.py: a model that goes quiet for 6 seconds, with a configurable keep-alive ping interval."""
import asyncio
import os
from collections.abc import AsyncIterable

import fastapi.routing
from fastapi import FastAPI, HTTPException
from fastapi.responses import HTMLResponse
from fastapi.sse import EventSourceResponse, ServerSentEvent

# FastAPI keeps its 15 second ping interval in a private name that its own tests monkeypatch; the lab does the same.
fastapi.routing._PING_INTERVAL = float(os.environ.get("PING_INTERVAL", "15"))

app = FastAPI()


@app.get("/blank")
def blank() -> HTMLResponse:
    return HTMLResponse("<!doctype html><title>blank</title>")


@app.get("/quiet", response_class=EventSourceResponse)
async def quiet(silence: float = 6) -> AsyncIterable[ServerSentEvent]:
    yield ServerSentEvent(data="before the pause")
    await asyncio.sleep(silence)
    yield ServerSentEvent(data="after the pause")


@app.get("/inside", response_class=EventSourceResponse)
async def inside() -> AsyncIterable[ServerSentEvent]:
    raise HTTPException(status_code=404, detail="unknown run")  # too late: by now the response is already a stream
    yield ServerSentEvent(data="never sent")

demo_proxy.py starts four app servers and nginx, then measures the arrival time of the first, the twentieth and the last event on each route, and finally tests the idle timeout. To run it you need the nginx executable: set the NGINX_EXE environment variable to its path (on Windows I used the zip from nginx.org; on Linux or macOS use your package manager’s binary).

"""demo_proxy.py: put nginx in front of the app, then measure buffering and the idle timeout."""
import os
import subprocess
import time

import labkit

NGINX = os.environ.get("NGINX_EXE", "nginx")
PREFIX = os.path.join(labkit.HERE, "nginx_run")
CONF = os.path.join(labkit.HERE, "nginx.conf")
for folder in ("logs", "temp"):
    os.makedirs(os.path.join(PREFIX, folder), exist_ok=True)
base = [NGINX, "-p", PREFIX, "-c", CONF]

apps = [labkit.start("app_v1:app", 8107), labkit.start("app_v2:app", 8108)]
os.environ["PING_INTERVAL"] = "15"
apps.append(labkit.start("app_slow:app", 8109))
os.environ["PING_INTERVAL"] = "1"
apps.append(labkit.start("app_slow:app", 8110))
web = subprocess.Popen(base + ["-g", "daemon off;"])
time.sleep(1.5)
try:
    routes = [
        ("direct to uvicorn, hand-written SSE", 8107, "/fixed"),
        ("nginx default, hand-written SSE (no header)", 8180, "/v1/fixed"),
        ("nginx default, built-in SSE (X-Accel-Buffering: no)", 8180, "/v2/stream"),
        ("nginx proxy_buffering off, hand-written SSE", 8180, "/v1-off/fixed"),
    ]
    print(f"{'route':<54}{'1st event':>10}{'20th':>8}{'last':>8}")
    for label, port, path in routes:
        status, headers, chunks = labkit.read_events(port, path)
        times = [seconds for seconds, _ in chunks]
        print(f"{label:<54}{times[0]:>9.2f}s{times[19]:>7.2f}s{times[-1]:>7.2f}s")

    print("\nnginx proxy_read_timeout 3s, model silent for 6 s after its first token:")
    for label, path in (("ping every 15 s (the default)", "/idle/quiet"), ("ping every 1 s", "/idle-ping/quiet")):
        started = time.perf_counter()
        status, headers, chunks = labkit.read_events(8180, path)
        elapsed = time.perf_counter() - started
        kinds = ["comment" if text.startswith(":") else text.split("data: ")[1].strip() for _, text in chunks]
        print(f"  {label:<30} stream ended after {elapsed:4.1f} s, received {kinds}")
finally:
    subprocess.run(base + ["-s", "stop"], capture_output=True)
    web.wait(timeout=10)
    for proc in apps:
        labkit.stop(proc)

python demo_proxy.py

route                                                  1st event    20th    last
direct to uvicorn, hand-written SSE                        0.09s   1.32s   2.54s
nginx default, hand-written SSE (no header)                0.06s   1.28s   2.51s
nginx default, built-in SSE (X-Accel-Buffering: no)        0.10s   1.32s   2.54s
nginx proxy_buffering off, hand-written SSE                0.06s   1.29s   2.51s

nginx proxy_read_timeout 3s, model silent for 6 s after its first token:
  ping every 15 s (the default)  stream ended after  3.0 s, received ['"before the pause"']
  ping every 1 s                 stream ended after  6.1 s, received ['"before the pause"', 'comment', 'comment', 'comment', 'comment', 'comment', '"after the pause"']

The first table did not show what the documentation warns about. On nginx 1.30.6 for Windows, with default settings and 39 small events, every route delivered its first event after about 0.1 seconds and its last after about 2.5 seconds, including the hand-written stream that sends no header. I could not reproduce buffering here, so I will not claim that nginx always delays a stream. Other proxies, CDNs and versions behave differently, and the header costs nothing, so send it, and measure through your own stack with this same script.

The second experiment did reproduce a proxy problem. With the default 15 second ping, nginx cut the stream after 3.0 seconds, having delivered one event, because 3 seconds passed without a byte. With a ping every second the stream stayed open for the full 6.1 seconds and delivered five pings and then the second event. If a proxy or load balancer in your path has an idle timeout shorter than your ping interval, you will see exactly this: an answer that stops in the middle, and a browser that reconnects.

Six connections per host

Browsers limit how many HTTP/1.1 connections they open to one host, and every open stream holds one for as long as it lasts. MDN warns that “When not used over HTTP/2, SSE suffers from a limitation to the maximum number of open connections, which can be especially painful when opening multiple tabs, as the limit is per browser and is set to a very low number (6).” The script opens eight streams from one tab and then splits them across two tabs.

"""demo_limit.py: how many EventSource connections can one browser hold open to one host over HTTP/1.1?"""
from playwright.sync_api import sync_playwright

import labkit

PORT = 8111
URL = f"http://127.0.0.1:{PORT}"
OPEN_MANY = """async (count) => {
  window.opened = 0;
  window.sources = [];
  for (let i = 0; i < count; i++) {
    const source = new EventSource('/quiet?silence=60');
    source.onopen = () => window.opened++;
    window.sources.push(source);
  }
  await new Promise((resolve) => setTimeout(resolve, 2000));
  return window.opened;
}"""

proc = labkit.start("app_slow:app", PORT)
try:
    with sync_playwright() as p:
        browser = p.chromium.launch(channel="msedge", headless=True)
        print("browser:", browser.version)
        context = browser.new_context()

        page = context.new_page()
        page.goto(f"{URL}/blank")
        print("one tab, 8 EventSources to the same host: open after 2 s =", page.evaluate(OPEN_MANY, 8))
        page.evaluate("window.sources[0].close()")
        page.wait_for_timeout(1000)
        print("  after closing one of them, connections that have ever opened =", page.evaluate("window.opened"), "(the 7th was waiting for the freed slot)")
        page.close()

        a, b = context.new_page(), context.new_page()
        a.goto(f"{URL}/blank")
        b.goto(f"{URL}/blank")
        in_a = a.evaluate(OPEN_MANY, 4)
        in_b = b.evaluate(OPEN_MANY, 4)
        print(f"two tabs of one browser, 4 EventSources each: tab A opened {in_a}, tab B opened {in_b}, total {in_a + in_b}")
        browser.close()
finally:
    labkit.stop(proc)

python demo_limit.py

browser: 156.0.4314.50
one tab, 8 EventSources to the same host: open after 2 s = 6
  after closing one of them, connections that have ever opened = 7 (the 7th was waiting for the freed slot)
two tabs of one browser, 4 EventSources each: tab A opened 4, tab B opened 2, total 6

Edge opened 6 of the 8 streams in one tab. The seventh stayed waiting until I closed one stream, and then it took the free slot. Two tabs of the same browser shared the limit: tab A got its 4 and tab B only 2, a total of 6. The practical consequences are to open one stream per tab, never one per component, and to serve streams over HTTP/2 in production, where, according to MDN, “the maximum number of simultaneous HTTP streams is negotiated between the server and the client (defaults to 100).” I tested only HTTP/1.1.

Step 9: Test the stream, then break it on purpose

Streaming bugs are timing bugs, so they hide from tests that check only a final response. The suite below starts real uvicorn servers and checks what travels on the wire: the parser rules, the hand-written format, the built-in headers, event ids, resuming, the 404, failures, the grace period, the ping, the exception raised inside a generator, and the three cancellation cases from Step 5. Each fixture starts its server once for the whole module, and serve sets environment variables such as GRACE for the child process only.

"""test_sse.py: the streaming behaviour of the lab apps, checked against real uvicorn servers."""
import json
import os
import threading
import time

import pytest

import labkit
import sse_parse
from fake_model import ANSWER


def serve(app, port, **env):
    old = {key: os.environ.get(key) for key in env}
    os.environ.update(env)
    proc = labkit.start(app, port)
    for key, value in old.items():  # restore the environment once the child has inherited it
        os.environ.pop(key, None) if value is None else os.environ.__setitem__(key, value)
    return proc


@pytest.fixture(scope="module")
def v1():
    proc = serve("app_v1:app", 8121)
    yield 8121
    labkit.stop(proc)


@pytest.fixture(scope="module")
def v2():
    proc = serve("app_v2:app", 8122)
    yield 8122
    labkit.stop(proc)


@pytest.fixture(scope="module")
def v3():
    proc = serve("app_v3:app", 8123, GRACE="1")
    yield 8123
    labkit.stop(proc)


@pytest.fixture(scope="module")
def cancel():
    proc = serve("app_cancel:app", 8124)
    yield 8124
    labkit.stop(proc)


@pytest.fixture(scope="module")
def pinger():
    proc = serve("app_slow:app", 8125, PING_INTERVAL="0.5")
    yield 8125
    labkit.stop(proc)


def parsed(chunks):
    return sse_parse.parse("".join(text for _, text in chunks))


def new_run(port, **body):
    return labkit.call(port, "POST", "/runs", {"prompt": "hi", **body})["id"]


def wait_for_status(port, run_id, wanted, timeout=6):
    deadline = time.time() + timeout
    while time.time() < deadline:
        run = labkit.call(port, "GET", f"/runs/{run_id}")
        if run["status"] == wanted:
            return run
        time.sleep(0.1)
    raise AssertionError(f"run never became {wanted}: {run}")


# --- the parser follows the WHATWG rules

def test_parser_joins_data_lines_with_newlines():
    assert sse_parse.parse("data: a\ndata: b\n\n")[0]["data"] == "a\nb"


def test_parser_removes_only_one_leading_space():
    assert sse_parse.parse("data:  x\n\n")[0]["data"] == " x"


def test_parser_ignores_comments_and_reads_retry():
    events = sse_parse.parse(": ping\n\nretry: 500\ndata: x\n\n")
    assert [e["data"] for e in events] == ["x"] and events[0]["retry"] == 500


def test_parser_discards_an_event_that_never_got_its_blank_line():
    assert sse_parse.parse("data: x\n\ndata: y\n") == sse_parse.parse("data: x\n\n")


# --- the wire format

def test_hand_written_stream_loses_the_paragraph_break(v1):
    _, _, chunks = labkit.read_events(v1, "/naive")
    assert "".join(e["data"] for e in parsed(chunks)) != ANSWER


def test_hand_written_stream_with_one_data_line_per_text_line_is_exact(v1):
    _, _, chunks = labkit.read_events(v1, "/fixed")
    assert "".join(e["data"] for e in parsed(chunks)) == ANSWER


def test_built_in_stream_sends_the_documented_headers(v2):
    status, headers, _ = labkit.read_events(v2, "/stream", stop_after=1)
    assert status == 200
    assert headers["content-type"].startswith("text/event-stream")
    assert headers["cache-control"] == "no-cache"
    assert headers["x-accel-buffering"] == "no"


def test_built_in_stream_json_encodes_tokens_so_newlines_survive(v2):
    _, _, chunks = labkit.read_events(v2, "/stream")
    events = parsed(chunks)
    assert "".join(json.loads(e["data"]) for e in events if e["type"] == "message") == ANSWER
    assert events[-1]["type"] == "done"


def test_idle_stream_gets_ping_comments(pinger):
    _, _, chunks = labkit.read_events(pinger, "/quiet?silence=2")
    assert any(text == ": ping\n\n" for _, text in chunks)


# --- resuming with Last-Event-ID

def test_event_ids_count_up_from_zero(v3):
    run_id = new_run(v3)
    _, _, chunks = labkit.read_events(v3, f"/runs/{run_id}/events")
    tokens = [e for e in parsed(chunks) if e["type"] == "message"]
    assert [int(e["id"]) for e in tokens] == list(range(39))


def test_resume_returns_only_later_events(v3):
    run_id = new_run(v3)
    _, _, chunks = labkit.read_events(v3, f"/runs/{run_id}/events", headers={"Last-Event-ID": "10"})
    tokens = [e for e in parsed(chunks) if e["type"] == "message"]
    assert tokens[0]["id"] == "11" and len(tokens) == 28


def test_resume_after_the_last_token_gets_only_done(v3):
    run_id = new_run(v3)
    labkit.read_events(v3, f"/runs/{run_id}/events")
    _, _, chunks = labkit.read_events(v3, f"/runs/{run_id}/events", headers={"Last-Event-ID": "38"})
    assert [e["type"] for e in parsed(chunks)] == ["done"]


def test_dropped_then_resumed_stream_has_no_gap_and_no_repeat(v3):
    run_id = new_run(v3)
    _, _, first = labkit.read_events(v3, f"/runs/{run_id}/events?drop_after=7")
    seen = [e for e in parsed(first) if e["type"] == "message"]
    assert len(seen) == 7 and "done" not in [e["type"] for e in parsed(first)]
    _, _, second = labkit.read_events(v3, f"/runs/{run_id}/events?drop_after=7", headers={"Last-Event-ID": seen[-1]["id"]})
    seen += [e for e in parsed(second) if e["type"] == "message"]
    assert [int(e["id"]) for e in seen] == list(range(39))
    assert "".join(json.loads(e["data"]) for e in seen) == ANSWER


def test_two_viewers_of_one_run_both_get_everything(v3):
    run_id = new_run(v3)
    results = []

    def watch():
        _, _, chunks = labkit.read_events(v3, f"/runs/{run_id}/events")
        results.append(len([e for e in parsed(chunks) if e["type"] == "message"]))

    threads = [threading.Thread(target=watch) for _ in range(2)]
    [t.start() for t in threads]
    [t.join() for t in threads]
    assert results == [39, 39]


def test_unknown_run_is_a_plain_404(v3):
    status, headers, _ = labkit.read_events(v3, "/runs/nope/events")
    assert status == 404 and not headers["content-type"].startswith("text/event-stream")


def test_an_exception_inside_the_generator_cannot_become_a_404(pinger):
    status, headers, chunks = labkit.read_events(pinger, "/inside")
    assert status == 200 and headers["content-type"].startswith("text/event-stream") and chunks == []


# --- failures and the grace period

def test_backend_crash_arrives_as_a_failure_event_not_as_done(v3):
    run_id = new_run(v3, fail_after=10)
    _, _, chunks = labkit.read_events(v3, f"/runs/{run_id}/events")
    events = parsed(chunks)
    assert len([e for e in events if e["type"] == "message"]) == 10
    assert events[-1]["type"] == "failure" and "crashed" in json.loads(events[-1]["data"])["message"]


def test_leaving_for_good_stops_the_model_after_the_grace_period(v3):
    run_id = new_run(v3)
    labkit.read_events(v3, f"/runs/{run_id}/events", stop_after=6)
    run = wait_for_status(v3, run_id, "cancelled")
    assert run["tokens"] < 39 and run["watchers"] == 0


def test_coming_back_inside_the_grace_period_keeps_the_run_alive(v3):
    run_id = new_run(v3)
    labkit.read_events(v3, f"/runs/{run_id}/events", stop_after=6)
    time.sleep(0.3)
    _, _, chunks = labkit.read_events(v3, f"/runs/{run_id}/events", headers={"Last-Event-ID": "4"})
    assert parsed(chunks)[-1]["type"] == "done"
    assert wait_for_status(v3, run_id, "finished")["tokens"] == 39


# --- stopping the model when the user leaves

def leave_after(port, path, tokens=5):
    labkit.call(port, "POST", "/stats/reset")
    labkit.read_events(port, path, stop_after=tokens)
    time.sleep(2.5)
    return labkit.call(port, "GET", "/stats")


def test_async_generator_stops_within_a_token_of_the_client_leaving(cancel):
    stats = leave_after(cancel, "/async")
    assert stats["produced"] <= 6 and stats["stopped"] == 1


def test_model_in_a_forgotten_task_runs_to_the_end_for_nobody(cancel):
    assert leave_after(cancel, "/task")["produced"] == 39


def test_cancelling_the_task_in_finally_stops_the_model(cancel):
    stats = leave_after(cancel, "/task-fixed")
    assert stats["produced"] <= 6 and stats["stopped"] == 1

python -m pytest test_sse.py -v

PASSED test_parser_joins_data_lines_with_newlines
PASSED test_parser_removes_only_one_leading_space
PASSED test_parser_ignores_comments_and_reads_retry
PASSED test_parser_discards_an_event_that_never_got_its_blank_line
PASSED test_hand_written_stream_loses_the_paragraph_break
PASSED test_hand_written_stream_with_one_data_line_per_text_line_is_exact
PASSED test_built_in_stream_sends_the_documented_headers
PASSED test_built_in_stream_json_encodes_tokens_so_newlines_survive
PASSED test_idle_stream_gets_ping_comments
PASSED test_event_ids_count_up_from_zero
PASSED test_resume_returns_only_later_events
PASSED test_resume_after_the_last_token_gets_only_done
PASSED test_dropped_then_resumed_stream_has_no_gap_and_no_repeat
PASSED test_two_viewers_of_one_run_both_get_everything
PASSED test_unknown_run_is_a_plain_404
PASSED test_an_exception_inside_the_generator_cannot_become_a_404
PASSED test_backend_crash_arrives_as_a_failure_event_not_as_done
PASSED test_leaving_for_good_stops_the_model_after_the_grace_period
PASSED test_coming_back_inside_the_grace_period_keeps_the_run_alive
PASSED test_async_generator_stops_within_a_token_of_the_client_leaving
PASSED test_model_in_a_forgotten_task_runs_to_the_end_for_nobody
PASSED test_cancelling_the_task_in_finally_stops_the_model
22 passed in 38.55s

All 22 tests pass. This suite also earned its place by failing once: my first parser kept the empty piece left at the end of the text and treated it as a blank line, so a half-received event was dispatched. test_parser_discards_an_event_that_never_got_its_blank_line failed, and the lines.pop() line in sse_parse.py is the fix.

Passing tests prove little unless they can fail. mutations.py copies the lab, plants one bug at a time (an off-by-one resume, events without ids, a reaper that never runs, a crash reported as success and so on), runs the suite and reports whether it noticed.

"""mutations.py: plant one bug at a time in a copy of the lab and check that the test suite notices."""
import os
import shutil
import subprocess
import sys

HERE = os.path.dirname(os.path.abspath(__file__))
SKIP = {"venv", "nginx_run", "__pycache__", ".pytest_cache", "mut"}

MUTANTS = [
    ("resume repeats the last event", "app_v3.py", "last_event_id + 1 if resume", "last_event_id if resume"),
    ("events carry no id", "app_v3.py", "ServerSentEvent(data=token, id=str(next_index))", "ServerSentEvent(data=token)"),
    ("nobody ever stops an abandoned run", "app_v3.py", "self.reaper = spawn(self.reap())", "pass"),
    ("a returning viewer does not cancel the reaper", "app_v3.py", "            self.reaper.cancel()\n", "            pass\n"),
    ("a crashed backend is reported as finished", "app_v3.py", 'status, self.error = "failed", str(exc)', 'status = "finished"'),
    ("hand-written stream sends one data line per token", "app_v1.py", 'lines = "".join(f"data: {part}\\n" for part in token.split("\\n"))', 'lines = f"data: {token}\\n"'),
    ("the forgotten task is no longer cancelled", "app_cancel.py", "        task.cancel()  # runs when", "        pass  # runs when"),
    ("the ping interval is ignored", "app_slow.py", 'float(os.environ.get("PING_INTERVAL", "15"))', "15.0"),
]


def run_suite(folder: str) -> list[str]:
    result = subprocess.run([sys.executable, "-m", "pytest", "test_sse.py", "-q", "-x", "-p", "no:cacheprovider"],
                            cwd=folder, capture_output=True, text=True, env={**os.environ, "COLUMNS": "200"})
    return [line.split(" - ")[0].replace("FAILED test_sse.py::", "") for line in result.stdout.splitlines() if line.startswith("FAILED")]


if __name__ == "__main__":
    caught = 0
    for name, filename, old, new in MUTANTS:
        folder = os.path.join(HERE, "mut")
        shutil.rmtree(folder, ignore_errors=True)
        shutil.copytree(HERE, folder, ignore=lambda d, names: [n for n in names if n in SKIP])
        path = os.path.join(folder, filename)
        source = open(path, encoding="utf-8").read()
        assert source.count(old) == 1, f"pattern for '{name}' found {source.count(old)} times"
        open(path, "w", encoding="utf-8", newline="\n").write(source.replace(old, new))
        failed = run_suite(folder)
        caught += bool(failed)
        print(f"{'caught' if failed else 'MISSED':<7}{name:<52}{failed[0] if failed else ''}")
    shutil.rmtree(os.path.join(HERE, "mut"), ignore_errors=True)
    print(f"{caught} of {len(MUTANTS)} planted bugs caught")

python mutations.py

caught resume repeats the last event                       test_resume_returns_only_later_events
caught events carry no id                                  test_event_ids_count_up_from_zero
caught nobody ever stops an abandoned run                  test_leaving_for_good_stops_the_model_after_the_grace_period
caught a returning viewer does not cancel the reaper       test_dropped_then_resumed_stream_has_no_gap_and_no_repeat
caught a crashed backend is reported as finished           test_backend_crash_arrives_as_a_failure_event_not_as_done
caught hand-written stream sends one data line per token   test_hand_written_stream_with_one_data_line_per_text_line_is_exact
caught the forgotten task is no longer cancelled           test_cancelling_the_task_in_finally_stops_the_model
caught the ping interval is ignored                        test_idle_stream_gets_ping_comments
8 of 8 planted bugs caught

All 8 planted bugs were caught, each by the test you would expect.

Common mistakes and how to spot them

Check that everything works end to end

run_all.py runs every demo, the test suite and the mutation check, and saves each output under out/. It takes about six minutes. Pass job names to run only some of them, for example python run_all.py demo_wire demo_builtin demo_browser if you skipped Ollama and nginx.

"""run_all.py: run every step script, pytest and the mutation check; save each output under out/."""
import os
import re
import subprocess
import sys
import time

HERE = os.path.dirname(os.path.abspath(__file__))
OUT = os.path.join(HERE, "out")
os.makedirs(OUT, exist_ok=True)
env = {**os.environ, "COLUMNS": "170", "PYTHONIOENCODING": "utf-8"}

KEEP = {"fastapi", "starlette", "uvicorn", "httpx", "pytest", "playwright", "h11"}  # the packages worth printing

JOBS = [
    ("versions", [sys.executable, "-m", "pip", "list", "--disable-pip-version-check", "--format=freeze"]),
    ("demo_wire", [sys.executable, "demo_wire.py"]),
    ("demo_builtin", [sys.executable, "demo_builtin.py"]),
    ("demo_browser", [sys.executable, "demo_browser.py"]),
    ("demo_cancel", [sys.executable, "demo_cancel.py"]),
    ("demo_resume", [sys.executable, "demo_resume.py"]),
    ("demo_grace", [sys.executable, "demo_grace.py"]),
    ("demo_proxy", [sys.executable, "demo_proxy.py"]),
    ("demo_limit", [sys.executable, "demo_limit.py"]),
    ("demo_ollama", [sys.executable, "demo_ollama.py"]),
    ("pytest", [sys.executable, "-m", "pytest", "test_sse.py", "-v", "-p", "no:cacheprovider"]),
    ("mutations", [sys.executable, "mutations.py"]),
]

if __name__ == "__main__":
    only = sys.argv[1:]
    for name, cmd in JOBS:
        if only and name not in only:
            continue
        started = time.time()
        result = subprocess.run(cmd, cwd=HERE, capture_output=True, text=True, encoding="utf-8", errors="replace", env=env)
        text = result.stdout + (("\n[stderr]\n" + result.stderr) if result.stderr.strip() else "")
        if name == "versions":
            text = "\n".join(line for line in text.splitlines() if line.split("==")[0].lower() in KEEP)
        if name == "pytest":
            rows = [re.sub(r"^test_sse.py::(\S+) (PASSED|FAILED).*$", r"\2 \1", line)
                    for line in text.splitlines() if "::" in line and ("PASSED" in line or "FAILED" in line)]
            text = "\n".join(rows + [line.strip("= ") for line in text.splitlines() if re.search(r"\d+ (passed|failed)", line)][-1:])
        open(os.path.join(OUT, f"{name}.txt"), "w", encoding="utf-8", newline="\n").write(text)
        print(f"{name:<16} exit {result.returncode}  {time.time() - started:5.1f} s  {len(text):>6} chars", flush=True)

python run_all.py

You are done when every job prints exit 0, out/pytest.txt ends with 22 passed, and out/mutations.txt ends with 8 of 8 planted bugs caught. If a browser job fails to start, check that Playwright can launch your browser; if a timing line differs from the page, compare ratios, not exact numbers.

Where to go next

Four directions follow naturally. First, put the token list in a shared store so any worker can resume any job, and add a time limit for finished jobs. Second, protect the endpoints: cap concurrent streams per user, as in hardening a FastAPI rate limiter, and shed load before streams pile up, as in adding backpressure and load shedding. Third, treat the ping as a heartbeat and think about what your own timeouts mean; detecting failed nodes with heartbeats goes deeper. Fourth, if your agent also calls tools while it streams, cutting AI agent latency with streaming and parallel tool calls pairs well with this post.

Sources

• WHATWG HTML Standard, Server-sent events: the event stream format, parsing rules, reconnection and Last-Event-ID.
• MDN, Using server-sent events: the six-connection limit and HTTP/2.
• FastAPI documentation, Server-Sent Events (SSE): EventSourceResponse, ServerSentEvent, resuming, SSE with POST and the built-in defaults.
• nginx documentation, ngx_http_proxy_module: proxy_buffering, X-Accel-Buffering and proxy_read_timeout.
• Ollama documentation, Streaming: newline-delimited JSON responses.
• Python documentation, asyncio Coroutines and Tasks: keeping references to tasks.