httpx-sse

raw JSON →
0.4.3 verified Tue May 12 auth: no python install: stale quickstart: stale

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.

pip install httpx-sse
error AttributeError: module 'httpx' has no attribute 'TransportError'
cause This error occurs when `httpx-sse` is used with a version of `httpx` (>=0.28.0) that has removed or renamed `httpx.TransportError`. Older versions of `httpx-sse` (prior to 0.4.3) subclass `httpx.TransportError`, leading to this `AttributeError` when `httpx` is updated.
fix
Upgrade httpx-sse to version 0.4.3 or newer, or pin your httpx dependency to a version older than 0.28.0. The httpx-sse library version 0.4.3 and above addresses this compatibility issue. pip install --upgrade httpx-sse or pip install httpx==0.27.*
error httpx_sse.SSEError: Expected Content-Type header to be 'text/event-stream', got 'application/json'
cause This error indicates that the server is not sending events with the correct `Content-Type` header, which should be `text/event-stream` as per the Server-Sent Events specification. `httpx-sse` explicitly checks for this header and raises an `SSEError` if it's incorrect.
fix
Ensure that the SSE endpoint on your server correctly sets the Content-Type header to text/event-stream for SSE responses. If you control the server, modify the response headers. If you don't, you might be connecting to a non-SSE endpoint or one that's misconfigured.
error httpx.ReadError: [Errno 104] Connection reset by peer
cause This error occurs when the connection to the SSE server is unexpectedly closed or breaks while `httpx-sse` is attempting to read data. This is a common network-related issue and typically indicates the server closed the connection, or there was an intermediary network problem. `httpx-sse` does not have built-in reconnection logic and will raise this `httpx.ReadError` when the connection is interrupted.
fix
Implement custom reconnection logic in your client code to gracefully handle httpx.ReadError. You can use the sse.id and sse.retry values provided by httpx-sse to resume the event stream. Libraries like stamina can assist with exponential backoff and retries.
error ModuleNotFoundError: No module named 'sse_starlette'
cause This error typically occurs when running examples or tests provided with `httpx-sse` that depend on `sse-starlette`, but `sse-starlette` has not been installed. While `httpx-sse` itself doesn't strictly require `sse-starlette` for basic operation, many examples demonstrating server-side SSE generation or testing setups use it.
fix
Install the sse-starlette library using pip to resolve the missing module. pip install sse-starlette
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.
breaking The `httpx` library is a required dependency. Failure to install it will result in a `ModuleNotFoundError`.
fix Install `httpx` using `pip install httpx`.
breaking A required dependency, `httpx`, was not found. This error occurs when the `httpx` library is not installed in the execution environment, preventing the application from running.
fix Ensure `httpx` is installed in your environment by running `pip install httpx`. If you are installing the main library, ensure its dependencies are correctly resolved (e.g., install it within a virtual environment or avoid `pip install --no-deps`).
python os / libc status wheel install import disk
3.10 alpine (musl) wheel - - 17.8M
3.10 alpine (musl) - - - -
3.10 slim (glibc) wheel 1.5s - 18M
3.10 slim (glibc) - - - -
3.11 alpine (musl) wheel - - 19.7M
3.11 alpine (musl) - - - -
3.11 slim (glibc) wheel 1.6s - 20M
3.11 slim (glibc) - - - -
3.12 alpine (musl) wheel - - 11.6M
3.12 alpine (musl) - - - -
3.12 slim (glibc) wheel 1.4s - 12M
3.12 slim (glibc) - - - -
3.13 alpine (musl) wheel - - 11.3M
3.13 alpine (musl) - - - -
3.13 slim (glibc) wheel 1.4s - 12M
3.13 slim (glibc) - - - -
3.9 alpine (musl) wheel - - 17.3M
3.9 alpine (musl) - - - -
3.9 slim (glibc) wheel 1.7s - 18M
3.9 slim (glibc) - - - -

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))