How to Implement Server-Sent Events with SSEResponse in Air

Use air.SSEResponse to wrap an asynchronous generator, which automatically formats chunks into the Server-Sent Events (SSE) protocol with proper headers and data: prefixes.

SSEResponse is a specialized response class in the feldroy/air repository that simplifies real-time streaming to web clients. It inherits from Starlette’s StreamingResponse but adds automatic SSE formatting, making it ideal for pushing live updates to browsers without WebSocket complexity.

How SSEResponse Works

SSEResponse is defined in src/air/responses.py and operates by intercepting chunks from an async iterator and wrapping them in the SSE wire format.

Inheritance and Media Type

The class extends StreamingResponse and hardcodes the SSE media type:

# src/air/responses.py (lines 40-55, 87)

class SSEResponse(StreamingResponse):
    def __init__(self, content, *args, **kwargs):
        super().__init__(content, *args, **kwargs)
        self.media_type = "text/event-stream"  # Line 87

This ensures browsers recognize the stream as text/event-stream, triggering the EventSource API to handle incoming messages correctly.

Chunk Processing Logic

Inside stream_response, the method handles three distinct chunk types:

1. Raw bytes: Passed through unchanged for pre-formatted SSE payloads
2. Air tags: Automatically rendered to HTML via str(tag) before formatting
3. Strings: Split on newlines and wrapped with data: prefixes

The formatting occurs in src/air/responses.py (lines 97-104):

# Pseudocode representation of the chunk handling

if isinstance(chunk, (bytes, memoryview)):
    await send({"type": "http.response.body", "body": chunk, ...})
else:
    text = str(chunk)  # Handles Air tags automatically

    for line in text.splitlines():
        await send(f"data: {line}\n")
    await send("\n")  # Empty line terminates SSE message

Implementing Server-Sent Events: Code Examples

Basic SSE Endpoint

Create an async generator that yields strings, then wrap it with SSEResponse:

import air

app = air.Air()

async def numbers():
    for i in range(5):
        yield f"Number {i}"
        await air.sleep(1)

@app.get("/sse")
async def sse_endpoint() -> air.SSEResponse:
    return air.SSEResponse(numbers())

This streams five messages to the client, one per second, formatted as:

event: message
data: Number 0

event: message
data: Number 1

Streaming Air Tags

SSEResponse automatically renders Air tags to HTML before SSE formatting:

import air

app = air.Air()

async def tag_stream():
    while True:
        yield air.P("Hello from a paragraph tag")
        await air.sleep(2)

@app.get("/tag-sse")
async def tag_sse() -> air.SSEResponse:
    return air.SSEResponse(tag_stream())

The client receives HTML-wrapped content:

event: message
data: <p>Hello from a paragraph tag</p>

Full HTMX Integration Example

The repository includes a complete demo showing SSE with HTMX in examples/server_sent_events.py:

import random
from asyncio import sleep
import air

app = air.Air()

@app.page
def index() -> air.Html | air.Children:
    return air.layouts.mvpcss(
        air.Script(src="https://unpkg.com/htmx-ext-sse@2.2.1/sse.js"),
        air.Title("Server Sent Event Demo"),
        air.H1("Server Sent Event Demo"),
        air.P("Lottery number generator"),
        air.Section(
            hx_ext="sse",
            sse_connect="/lottery-numbers",
            hx_swap="beforeend show:bottom",
            sse_swap="message",
        ),
    )

async def lottery_generator():
    while True:
        numbers = ", ".join(str(random.randint(1, 40)) for _ in range(6))
        yield str(air.Aside(numbers))
        await sleep(1)

@app.page
async def lottery_numbers() -> air.SSEResponse:
    return air.SSEResponse(lottery_generator())

This example demonstrates:

• HTMX SSE extension configuration via hx_ext="sse"
• Continuous streaming using an infinite async generator
• Air tag rendering by yielding air.Aside elements as strings

Testing SSE Output

The test suite in tests/test_responses.py validates the SSE formatting logic:

# tests/test_responses.py (lines 27-46)

async def test_sse_response():
    async def generator():
        yield air.P("Hello")
        yield air.P("World")
    
    response = air.SSEResponse(generator())
    assert response.headers["content-type"] == "text/event-stream; charset=utf-8"
    
    # Verify formatting

    body = await response.body()
    assert "event: message" in body
    assert "data: <p>Hello</p>" in body

Key assertions confirm:

• Content-Type header is correctly set to text/event-stream
• HTML rendering occurs before SSE wrapping
• Event framing includes proper event: and data: fields

Summary

• SSEResponse in src/air/responses.py wraps async generators to produce standards-compliant Server-Sent Events
• Automatic formatting converts Air tags to HTML and wraps content with event: message and data: prefixes
• Media type is hardcoded to text/event-stream for proper browser handling via EventSource
• Multiline support splits strings on newlines and sends each as a separate data: line
• Binary pass-through allows raw bytes to skip formatting for pre-encoded SSE payloads

Frequently Asked Questions

How do I handle connection errors in SSEResponse?

SSEResponse inherits exception handling from Starlette’s StreamingResponse. If the client disconnects, the async generator receives a CancelledError when it next attempts to yield. Wrap your generator logic in try/except asyncio.CancelledError to handle cleanup, such as closing database connections or stopping background tasks.

Can I send custom event types instead of the default "message"?

The current implementation in src/air/responses.py hardcodes event: message for all chunks. To send custom event types like event: update or event: notification, you must either yield pre-formatted SSE strings (which bypass the automatic formatting) or subclass SSEResponse to override the chunk handling logic in stream_response.

Why are my Air tags not rendering to HTML in the SSE stream?

SSEResponse automatically converts non-bytes chunks to strings using str(chunk), which renders Air tags to HTML. If you see raw object representations like <air.P object at 0x...>, ensure you are not wrapping the tag in a class that overrides __str__ unexpectedly, or explicitly convert tags with str() before yielding if using a custom wrapper.

How do I test SSE endpoints in Air?

Use the TestClient from Starlette (which Air inherits) to capture the streaming response. Assert on the content-type header to verify text/event-stream, and read the response body to check for proper SSE formatting with event: and data: lines. For async generators, ensure your test awaits the full body consumption or iterates over the streaming content.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}