Module

realtime.sse

Server-Sent Events protocol implementation over ASGI.

Handles the full SSE lifecycle: sendstext/event-streamheaders, produces events from an async generator, monitors for client disconnect, and sends periodic heartbeat comments to keep the connection alive.

4Functions

Functions

handle_sse 9 None ▼
Stream Server-Sent Events over an ASGI connection. 1. Sends ``http.response.st…
async
async def handle_sse(event_stream: EventStream, send: Send, receive: Receive, *, kida_env: Environment | None = None, debug: bool = False, retry_ms: int | None = None, close_event: str | None = None, trace_sink: Callable[[str, dict[str, Any]], None] | None = None, extra_headers: tuple[tuple[bytes, bytes], ...] = ()) -> None

Stream Server-Sent Events over an ASGI connection.

  1. Sendshttp.response.start with text/event-streamheaders.
  2. Launches two concurrent tasks:
    • Event producer: consumes the async generator, converts each yielded value to SSE wire format, and sends as ASGI body chunks.
    • Disconnect monitor: awaitshttp.disconnectfrom the client and cancels the producer.
  3. Sends periodic heartbeat comments (:) on idle.
Parameters
Name Type Description
event_stream EventStream
send Send
receive Receive
kida_env Environment | None Default:None
debug bool Default:False
retry_ms int | None Default:None
close_event str | None Default:None
trace_sink Callable[[str, dict[str, Any]], None] | None Default:None
extra_headers tuple[tuple[bytes, bytes], ...] Default:()
_format_event 3 str ▼
Convert a yielded value to SSE wire format. Dispatch: - ``SSEEvent`` -> en…
def _format_event(value: Any, *, default_event: str | None = None, kida_env: Environment | None = None) -> str

Convert a yielded value to SSE wire format.

Dispatch:

  • SSEEvent-> encode as-is
  • Fragment-> render via kida, using its target as event name when set
  • str-> wrap as data
  • dict-> JSON-serialize as data
Parameters
Name Type Description
value Any
default_event str | None Default:None
kida_env Environment | None Default:None
Returns
str
_trace_event_payload 2 dict[str, Any] ▼
Return bounded metadata about a formatted SSE event.
def _trace_event_payload(value: Any, sse_text: str) -> dict[str, Any]
Parameters
Name Type Description
value Any
sse_text str
Returns
dict[str, Any]
_format_error_event 2 str ▼
Format an error as an SSE event for a failed render. For ``Fragment`` values, …
def _format_error_event(value: Any, exc: Exception) -> str

Format an error as an SSE event for a failed render.

ForFragmentvalues, uses the fragment's target as the SSE event name so the error replaces the specific block in the DOM. This lets the developer see exactly which block broke, inline where it should be.

For other value types, sends a genericerrorevent.

Parameters
Name Type Description
value Any
exc Exception
Returns
str