# Server-Sent Events

URL: /docs/streaming/server-sent-events/
Section: streaming
Tags: sse, real-time, events, htmx

--------------------------------------------------------------------------------

What Are SSE? Server-Sent Events (SSE) are a standard browser API for receiving a stream of events from the server over a persistent HTTP connection. Unlike WebSockets, SSE is: One-directional -- server pushes to client Plain HTTP -- no protocol upgrade, no special infrastructure Auto-reconnecting -- the browser reconnects automatically Text-based -- simple text/event-stream format EventStream Return an EventStream from a route handler to start pushing events: from chirp import EventStream @app.route("/events") async def events(): async def stream(): while True: data = await get_next_update() yield data return EventStream(stream()) The generator yields values. Chirp formats them as SSE wire protocol and sends them to the client. Yield Types The generator can yield different types: async def stream(): # String -- sent as SSE data field yield "Hello, World!" # Dict -- JSON-serialized as SSE data yield {"count": 42, "status": "ok"} # Fragment -- rendered via kida, sent as named SSE event # target= becomes the SSE event name (default: "fragment") yield Fragment("components/notification.html", "alert", target="notification", message="New alert") # SSEEvent -- full control over event type, id, retry yield SSEEvent(data="custom", event="ping", id="1") SSEEvent For fine-grained control, yield SSEEvent objects: from chirp import SSEEvent async def stream(): yield SSEEvent( data="User joined", event="user-join", # Event type (client filters on this) id="evt-42", # Last-Event-ID for reconnection retry=5000, # Reconnection interval in ms ) Real-Time HTML with htmx The killer pattern: combine SSE with htmx to push rendered HTML fragments in real-time. Server: @app.route("/notifications") async def notifications(): async def stream(): async for event in notification_bus.subscribe(): yield Fragment("components/notification.html", "alert", target="notification", message=event.message, time=event.timestamp, ) return EventStream(stream()) Client (using htmx SSE extension): <div hx-ext="sse" sse-connect="/notifications"> <div sse-swap="notification" hx-swap="beforeend"> <!-- Fragments are swapped in here --> </div> </div> Important: sse-swap must be on a child of the sse-connect element, not the same element. htmx uses querySelectorAll internally, which does not include the root element itself. The server renders HTML, the browser swaps it in. Zero client-side JavaScript for the rendering logic. Live Dashboard Example A more complete example -- a dashboard that streams stats updates: @app.route("/dashboard/live") async def live_stats(): async def stream(): while True: stats = await get_current_stats() # Fragment.target becomes the SSE event name. # No need to wrap in SSEEvent -- chirp handles it. yield Fragment("dashboard.html", "stats_panel", target="stats-update", stats=stats) await asyncio.sleep(5) return EventStream(stream()) <section hx-ext="sse" sse-connect="/dashboard/live" hx-disinherit="hx-target hx-swap"> <div id="stats" sse-swap="stats-update"> {# Initial stats rendered server-side #} {% block stats_panel %} ... {% endblock %} </div> </section> Tip: Add hx-disinherit="hx-target hx-swap" on the sse-connect element to prevent layout-level hx-target from bleeding into SSE swap targets. Without this, SSE fragments can accidentally replace the wrong region. Connection Lifecycle Chirp manages the SSE connection lifecycle automatically: Event producer -- consumes the generator, formats events, sends as ASGI body chunks Disconnect monitor -- watches for http.disconnect, cancels the producer Heartbeat -- sends : heartbeat comments on idle to keep the connection alive The heartbeat uses asyncio.shield to avoid cancelling the pending generator coroutine when sending keep-alive comments. Testing SSE Use the TestClient.sse() method: from chirp.testing import TestClient async def test_notifications(): async with TestClient(app) as client: result = await client.sse("/notifications", max_events=3) assert len(result.events) == 3 assert "notification" in result.events[0].data See Testing Assertions for SSE testing details. Next Steps Streaming HTML -- Progressive page rendering Fragments -- How fragments are rendered Assertions -- Testing SSE endpoints

--------------------------------------------------------------------------------

Metadata:
- Word Count: 542
- Reading Time: 3 minutes