# How to Implement Server-Sent Events with SSEResponse in Air

> Implement Server-Sent Events with air SSREsponse by wrapping async generators. Automatically format chunks for SSE with headers and data prefixes. Stream real-time updates efficiently.

- Repository: [Feldroy/air](https://github.com/feldroy/air)
- Tags: how-to-guide
- Published: 2026-03-01

---

**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`](https://github.com/feldroy/air/blob/main/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:

```python

# 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`](https://github.com/feldroy/air/blob/main/src/air/responses.py) (lines 97-104):

```python

# 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`:

```python
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:

```python
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`](https://github.com/feldroy/air/blob/main/examples/server_sent_events.py):

```python
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`](https://github.com/feldroy/air/blob/main/tests/test_responses.py) validates the SSE formatting logic:

```python

# 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`](https://github.com/feldroy/air/blob/main/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`](https://github.com/feldroy/air/blob/main/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.