Live feed over SSE
This example streams HTML to the browser after the page loads — a feed of
notifications that appear one by one, with no client-side JavaScript. You return
anEventStreamfrom a route, and an async generator yields events over a
long-lived connection. Each yielded value can be a plain string, a structured
SSEEvent (custom event name and id), or a rendered Fragment(...)that htmx
swaps into the page.
Minimal example
Two routes: one renders the page shell, the other returns anEventStreamwhose
generator yields a mix ofSSEEvent and Fragmentvalues.
import asyncio
from pathlib import Path
from chirp import App, AppConfig, EventStream, Fragment, SSEEvent, Template
config = AppConfig(
template_dir=Path(__file__).parent / "templates",
worker_mode="async",
sse_close_event="close",
)
app = App(config=config)
_NOTIFICATIONS = [
{"title": "Welcome", "message": "You are now connected to the live feed."},
{"title": "Update", "message": "New deployment started."},
{"title": "Alert", "message": "CPU usage above 90% on worker-3."},
{"title": "Resolved", "message": "CPU usage back to normal."},
]
@app.route("/")
def index():
return Template("feed.html")
@app.route("/events", referenced=True)
def events():
async def generate():
# Structured SSEEvent — custom event name, kept off the htmx message channel.
yield SSEEvent(data="connected", event="status")
# Fragment events — rendered HTML pushed to the browser and swapped by htmx.
for notification in _NOTIFICATIONS:
await asyncio.sleep(1.5)
yield Fragment(
"feed.html",
"notification",
title=notification["title"],
message=notification["message"],
)
return EventStream(generate())
if __name__ == "__main__":
app.run()
Source: examples/standalone/sse/app.py.
Setworker_mode="async"— the stream holds a connection open and awaits between
events. This maintained example follows Chirp's htmx 2 default: aFragment's
target becomes the named event consumed bysse-swap. The exact htmx 4 preview
instead emitshx-sse:connect; its rendered fragments are unnamed HTML and an
explicit target becomes a DOM id inside<hx-partial>. Named SSEEvents remain
literal DOM events. See the SSE guide for the side-by-side migration contract.
Run it
What the contract checks cover here
These checks run at startup viaapp.check()— they are static analysis of your
routes and templates, separate from the per-event runtime boundary described
above.
- Event cross-references (
sse_crossref). Eachsse-swaptarget in the markup is matched against the event names the connected route emits — thetargeton aFragment(...), theeventon anSSEEvent(...), and any declaredSSEContract.event_types. A target that no event produces is flagged as a likely typo. - Self-swap (
sse_self_swap). Ansse-swapon the same element assse-connectnever matches (htmx excludes the root element), so it errors. - Connect scope (
sse_scope). Ansse-connectinside a broadhx-targetscope without mitigation errors, since inherited swaps would clobber the wrong region.