Skip to content

SSEStream

SSEStream builds a text/event-stream response for Server-Sent Events.

Quick Start

from faststrap.presets import SSEStream, sse_event

@app.get("/api/stream")
async def stream():
    async def gen():
        yield sse_event("Hello")
    return SSEStream(gen())

Keep Alive Comments

from faststrap.presets import sse_comment

yield sse_comment("keepalive")

Pair With SSETarget

from faststrap import SSETarget

SSETarget("Waiting...", endpoint="/api/stream")

Security Notes

Use authentication and avoid streaming sensitive data to unauthenticated users. For production, disable proxy buffering.

API Reference

faststrap.presets.streams.SSEStream(events, *, headers=None)

Create a StreamingResponse for Server-Sent Events (SSE).

Parameters:

Name Type Description Default
events Iterable[Any] | AsyncIterable[Any]

Iterable or async iterable of SSE payloads.

 required
headers dict[str, str] | None

Optional extra headers.

 None

Returns:

Type Description
StreamingResponse

StreamingResponse configured for text/event-stream.
Source code in src/faststrap/presets/streams.py 
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
def SSEStream(
    events: Iterable[Any] | AsyncIterable[Any],
    *,
    headers: dict[str, str] | None = None,
) -> StreamingResponse:
    """Create a StreamingResponse for Server-Sent Events (SSE).

    Args:
        events: Iterable or async iterable of SSE payloads.
        headers: Optional extra headers.

    Returns:
        StreamingResponse configured for text/event-stream.
    """
    base_headers = {
        "Cache-Control": "no-cache",
        "Connection": "keep-alive",
        "X-Accel-Buffering": "no",
    }
    if headers:
        base_headers.update(headers)

    body_iter: Iterable[str] | AsyncIterator[str]
    if isinstance(events, AsyncIterable):
        body_iter = _aiter_sse(events)
    else:
        body_iter = _iter_sse(events)

    return StreamingResponse(body_iter, media_type="text/event-stream", headers=base_headers)