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.

2Functions

Functions

handle_sse 7 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) -> 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
_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, wrap with event: fragment
  • 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