Skip to content

Streaming Server-Sent Events from asyncio

Not every push feature needs WebSockets. Progress bars, notification badges, log tails, model-token streams and dashboards are one-directional: the server sends, the browser listens. Server-sent events do that over an ordinary HTTP response with a text/event-stream body, which means no protocol upgrade, no separate gateway path, automatic reconnection in the browser's EventSource, and resume support built into the protocol through Last-Event-ID. The parts that go wrong are all in the details: frames that are almost-but-not-quite the right format, proxies that buffer the stream until it looks broken, clients that disconnect without the server noticing, and a slow reader that quietly consumes memory on the server. This guide builds an SSE endpoint with an async generator, verified end to end with an HTTP client, and covers each of those details.

Prerequisites

The four fields of an SSE frame 4 stacked layers from event: price to data: {...}. The four fields of an SSE frame event: price names the event type clients subscribe by name id: 42 resume token returned as Last-Event-ID retry: 2000 browser reconnect delay milliseconds data: {...} one line per payload line blank line ends the frame A missing blank line, or a raw newline inside data, breaks the frame.

1. Emit correctly framed events

An SSE frame is a set of field: value lines followed by a blank line. Four fields matter: data (repeated for multi-line payloads), event (a name the client can subscribe to), id (the resume token) and retry (the client's reconnect delay in milliseconds). A line beginning with : is a comment.

import json


def sse_frame(data, event: str | None = None, event_id: int | str | None = None,
              retry_ms: int | None = None) -> bytes:
    lines: list[str] = []
    if event is not None:
        lines.append(f"event: {event}")
    if event_id is not None:
        lines.append(f"id: {event_id}")                       # becomes Last-Event-ID on reconnect
    if retry_ms is not None:
        lines.append(f"retry: {retry_ms}")
    for line in json.dumps(data).splitlines():
        lines.append(f"data: {line}")                         # every line needs its own data:
    return ("\n".join(lines) + "\n\n").encode()               # blank line terminates the frame


print(sse_frame({"seq": 0, "price": 100}, event="price", event_id=0, retry_ms=2000).decode())

The two rules that break implementations: a payload containing a newline must be split across several data: lines, and the frame ends with a blank line — a single trailing newline leaves the client waiting. Serialising to JSON first, then splitting, handles both without special cases.

Verify: printing the frame shows event, id, retry and data lines followed by an empty line.

2. Serve the stream from an async generator

The endpoint returns a streaming response whose body is an async generator. The headers matter as much as the body: the content type identifies the protocol, and the caching and buffering headers keep intermediaries from holding the stream.

# pip install starlette httpx
import asyncio

from starlette.applications import Starlette
from starlette.requests import Request
from starlette.responses import StreamingResponse
from starlette.routing import Route

EVENTS = [{"seq": i, "price": 100 + i} for i in range(6)]


async def event_source(request: Request) -> StreamingResponse:
    last_id = request.headers.get("last-event-id")
    start = int(last_id) + 1 if last_id is not None else 0     # resume, see step 3

    async def body():
        yield b": stream open\n\n"                             # a comment flushes buffering proxies
        for event in EVENTS[start:]:
            if await request.is_disconnected():                # step 4
                return
            yield sse_frame(event, event="price", event_id=event["seq"], retry_ms=2000)
            await asyncio.sleep(0.01)

    return StreamingResponse(body(), media_type="text/event-stream", headers={
        "Cache-Control": "no-store",                           # never cache a live stream
        "X-Accel-Buffering": "no",                             # nginx: do not buffer
        "Connection": "keep-alive",
    })


app = Starlette(routes=[Route("/events", event_source)])

X-Accel-Buffering: no is the header most teams discover late: with default nginx settings the proxy buffers the response and the browser receives nothing until the buffer fills or the request ends, which looks exactly like a broken endpoint. The opening comment serves the same purpose for intermediaries that flush on first bytes.

Verify: the response's content type is text/event-stream, and the first bytes arrive before the generator finishes.

3. Support resume with Last-Event-ID

When a browser's EventSource reconnects, it sends the last id it received in a Last-Event-ID header. A server that honours it can replay only what the client missed, which turns a reconnect into a gap-free continuation.

import asyncio

import httpx


async def main() -> None:
    transport = httpx.ASGITransport(app=app)
    async with httpx.AsyncClient(transport=transport, base_url="http://sse") as client:
        async with client.stream("GET", "/events") as response:
            print("content-type:", response.headers["content-type"])
            frames, buffer = [], ""
            async for chunk in response.aiter_text():
                buffer += chunk
                while "\n\n" in buffer:
                    frame, buffer = buffer.split("\n\n", 1)
                    frames.append(frame)
            print("frames received:", len(frames))

        async with client.stream("GET", "/events",
                                 headers={"Last-Event-ID": "3"}) as response:
            text = "".join([chunk async for chunk in response.aiter_text()])
            print("after resume:", [l for l in text.splitlines() if l.startswith("id:")])


asyncio.run(main())

The first request received seven frames — the opening comment plus six events — and the resumed request returned only id: 4 and id: 5. Where the source cannot replay, send the client a snapshot event first and then live data, so the gap is explicit. The retry: field lets the server tune the browser's reconnect delay, which is the SSE equivalent of the backoff discussed in reconnecting WebSocket clients with backoff.

Verify: the resumed stream contains only events after the supplied ID, and no event is delivered twice.

A reconnect that loses nothing 5 stages from server sends id: 3 to server replays 4+. A reconnect that loses nothing server sends id: 3 client stores it connection drops proxy or deploy browser retries after retry ms Last-Event-ID: 3 sent as a header server replays 4+ no gap, no duplicates The browser does the reconnecting; the server only has to honour the header.

4. Notice when the client goes away

A client that closes the tab does not tell the generator. Without a check, the server keeps producing events into a dead connection — and keeps holding the resources behind them. Starlette exposes request.is_disconnected(); the underlying ASGI mechanism is a http.disconnect message.

import asyncio

from starlette.requests import Request
from starlette.responses import StreamingResponse

active_streams = {"n": 0}


async def supervised_source(request: Request) -> StreamingResponse:
    async def body():
        active_streams["n"] += 1
        try:
            while True:
                if await request.is_disconnected():
                    return                                     # stop producing immediately
                yield sse_frame({"ts": asyncio.get_running_loop().time()}, event="tick")
                await asyncio.sleep(0.5)
        finally:
            active_streams["n"] -= 1                            # release per-stream resources
            # unsubscribe from the bus, close cursors, decrement metrics here

    return StreamingResponse(body(), media_type="text/event-stream")

The finally block is the important half: the generator is closed when the response ends, so subscriptions and counters are released deterministically. Long-lived streams should also check disconnection while waiting for data, not only between events — combine the wait with a timeout so a client that left during a quiet period is noticed within one interval.

Verify: active_streams["n"] returns to zero after clients disconnect, and no generator keeps producing for a closed response.

5. Bound the per-client queue and keep the stream alive

A real endpoint does not have its events in a list; it subscribes to a bus. That introduces the two remaining production concerns: a slow client must not grow the server's memory, and an idle stream must not be closed by an intermediary's read timeout.

import asyncio
import contextlib

from starlette.requests import Request
from starlette.responses import StreamingResponse

HEARTBEAT_SECONDS = 15.0


async def bus_source(request: Request, bus) -> StreamingResponse:
    queue: asyncio.Queue[bytes] = asyncio.Queue(maxsize=100)    # bounded per client
    dropped = 0

    def on_event(payload: bytes) -> None:
        nonlocal dropped
        try:
            queue.put_nowait(payload)
        except asyncio.QueueFull:
            dropped += 1                                        # slow client: drop, never block

    async def body():
        unsubscribe = bus.subscribe(on_event)
        try:
            yield b": stream open\n\n"
            while True:
                try:
                    async with asyncio.timeout(HEARTBEAT_SECONDS):
                        payload = await queue.get()
                except TimeoutError:
                    yield b": keep-alive\n\n"                    # keeps proxies from timing out
                    continue
                if await request.is_disconnected():
                    return
                yield payload
        finally:
            unsubscribe()
            if dropped:
                print(f"client dropped {dropped} events")

    return StreamingResponse(body(), media_type="text/event-stream")

The heartbeat comment costs four bytes and prevents the most common "SSE stops working after a minute" report, which is a proxy or load balancer closing an idle connection. The bounded queue with a drop counter is the same policy as fanning out a queue to multiple consumer groups: one slow reader loses events rather than consuming the server's memory or slowing everyone else.

Verify: an idle stream emits a keep-alive comment every interval, and a client that stops reading accumulates drops while the server's memory stays flat.

Server-sent events or WebSockets? A grid of 5 rows by 2 columns. Server-sent events or WebSockets? property server-sent events WebSockets direction server to client only both directions reconnect built into EventSource you implement it resume Last-Event-ID header application protocol proxies plain HTTP response needs upgrade support payloads UTF-8 text frames text or binary Choose SSE when the data only flows one way; that is most dashboards and progress streams.

Verification

An SSE endpoint is correct when:

  • Frames are well formed: multi-line payloads use repeated data: lines, and every frame ends with a blank line.
  • Streaming is not buffered: the first bytes reach the client immediately, with text/event-stream, Cache-Control: no-store and buffering disabled at the proxy.
  • Resume works: Last-Event-ID replays only missed events, with no duplicates.
  • Disconnects are detected: the generator stops producing and its finally releases subscriptions, with the active-stream count returning to zero.
  • Slow clients are bounded: per-client queues are bounded with a drop counter, and idle streams send heartbeat comments.

Pitfalls & edge cases

  • Missing the blank line. A frame without a terminating empty line is buffered by the client until the next one arrives, which looks like a one-event delay.
  • Newlines inside data. A raw newline in the payload silently ends the field; split the payload across data: lines.
  • Compression middleware. Gzip middleware may buffer the stream; exclude text/event-stream from compression.
  • HTTP/1.1 connection limits. Browsers allow about six connections per origin over HTTP/1.1, so several SSE tabs can starve other requests; HTTP/2 multiplexes them, as covered in HTTP/2 connection multiplexing with httpx.
  • SSE where bidirectional is needed. SSE is server-to-client only; a client that must send messages needs WebSockets or ordinary requests alongside the stream.

Frequently Asked Questions

How do I send server-sent events from an async Python app?

Return a streaming response whose body is an async generator yielding SSE frames, with media type text/event-stream. Each frame is one or more field lines — data, and optionally event, id and retry — followed by a blank line. Add Cache-Control: no-store and disable proxy buffering.

Why does my SSE stream arrive only when the response ends?

Something is buffering it. Most often it is a reverse proxy: set X-Accel-Buffering: no for nginx, or the equivalent for your proxy, and make sure compression middleware does not buffer text/event-stream. Sending a comment line as the first bytes also prompts many intermediaries to flush.

How does Last-Event-ID work in server-sent events?

The server includes an id field with each event, and the browser's EventSource remembers the last one. When it reconnects it sends that value in a Last-Event-ID header, so the server can replay only the events after it. Servers that cannot replay should send a snapshot event before resuming live data.

How does the server know when an SSE client disconnects?

ASGI delivers an http.disconnect message, which frameworks expose — in Starlette, await request.is_disconnected(). Check it between events and after waiting for data, and put subscription cleanup in the generator's finally block so resources are released when the response ends.