httpx-sse

0.4.3 · active · verified Sat Mar 28

httpx-sse is a Python library designed to consume Server-Sent Event (SSE) messages, integrating seamlessly with the HTTPX client. It provides both synchronous and asynchronous helpers to connect to SSE endpoints and iterate over the received events. Currently at version 0.4.3, the library is actively maintained with regular updates and fixes.

Warnings

breaking Python 3.7 support was officially dropped as it reached End-of-Life.
Fix: Upgrade to Python 3.8 or newer. The library requires Python >=3.9 since 0.4.0.
gotcha The timing of `SSEError` (for non `text/event-stream` Content-Type) was moved from `connect_sse()` to `iter_sse()`/`aiter_sse()`. This allows inspecting the raw response before content type validation.
Fix: If your code relied on `SSEError` being raised immediately during connection, you might need to adjust error handling to occur during iteration, or explicitly check `event_source.response.headers['Content-Type']` before iterating.
gotcha Incorrect newline parsing that was not compliant with the SSE specification was fixed. If you were working around previous parsing quirks, your code might need adjustment.
Fix: Ensure your SSE server emits events strictly according to the SSE specification to avoid unexpected parsing behavior or silent data loss. Update to `0.4.2` or later for correct parsing.
gotcha A performance issue was introduced in `0.4.2` due to improved line parsing, which was subsequently fixed.
Fix: If you experienced performance degradation after upgrading to `0.4.2`, update to `0.4.3` or newer to resolve the issue.

Install

pip install httpx-sse Install latest stable version

Imports

connect_sse
```
from httpx_sse import connect_sse
```
aconnect_sse
```
from httpx_sse import aconnect_sse
```
ServerSentEvent
```
from httpx_sse import ServerSentEvent
```

Quickstart

Demonstrates how to establish an asynchronous connection to an SSE endpoint using `aconnect_sse` and iterate through incoming `ServerSentEvent` objects. It includes error handling and a common pattern for managing HTTPX client timeouts, which is crucial for long-lived SSE connections.

import asyncio
import httpx
from httpx_sse import aconnect_sse, ServerSentEvent
import os

async def consume_sse_stream(url: str):
    # Ensure a long enough timeout for SSE streams, or set to None
    # if the stream is expected to be indefinite.
    # httpx.AsyncClient(timeout=None) or httpx.Timeout(60.0, connect=5.0)
    async with httpx.AsyncClient(timeout=None) as client:
        try:
            async with aconnect_sse(client, "GET", url) as event_source:
                async for sse in event_source.aiter_sse():
                    print(f"Event: {sse.event}, Data: {sse.data}, ID: {sse.id}, Retry: {sse.retry}")
                    if sse.event == "end":
                        break
        except httpx.RequestError as exc:
            print(f"An error occurred while requesting {exc.request.url!r}: {exc}")
        except Exception as e:
            print(f"An unexpected error occurred: {e}")

# Example usage (replace with your actual SSE endpoint)
# For local testing, you might run a simple SSE server.
SSE_ENDPOINT_URL = os.environ.get("SSE_TEST_URL", "http://localhost:8000/sse")

if __name__ == "__main__":
    print(f"Connecting to SSE endpoint: {SSE_ENDPOINT_URL}")
    asyncio.run(consume_sse_stream(SSE_ENDPOINT_URL))

view raw JSON →