Python SSE Client

0.0.27 · maintenance · verified Thu Apr 16

The `sseclient` library (version 0.0.27) is a Python client for consuming Server-Sent Events (SSE) streams, also known as EventSource. It provides an `SSEClient` class that acts as an iterator, allowing easy processing of messages from a server. This version, last updated in September 2020, handles automatic reconnections and parses event data. While functional, it is no longer actively developed, with more recent and actively maintained forks (e.g., `sseclient-py`) addressing modern Python features and known issues.

Common errors

```
ValueError: Unterminated string starting at...
```
cause This error often occurs when `json.loads(event.data)` is attempted on incomplete or corrupted JSON data within an SSE event. This can be due to `sseclient` not correctly handling fragmented HTTP chunks for large events.

fix As per warnings, this is a known issue with `sseclient` 0.0.27. Consider using `sseclient-py` which has addressed this, or implement robust error handling around `json.loads` and potentially reassemble fragmented data manually if the server is known to send it that way.
```
requests.exceptions.InvalidURL: Failed to parse: <Response [200]>
```
cause This error happens when attempting to initialize `SSEClient` by passing a `requests.Response` object's URL attribute (`response.url`) instead of the `response` object itself. `SSEClient` expects the raw streaming response object, not just its URL.

fix Initialize the `SSEClient` with the `requests.Response` object directly: `client = SSEClient(response)` where `response = requests.get(url, stream=True, ...)`.
```
requests.exceptions.HTTPError: 405 Client Error: Method Not Allowed
```
cause This typically occurs when the HTTP method used to request the SSE stream (e.g., `requests.post` or `requests.put`) is not supported by the server for SSE connections. SSE streams are almost universally initiated with a `GET` request.

fix Ensure you are using `requests.get()` to initiate the SSE stream. If the server requires parameters, include them as `params` in the `requests.get()` call.

Warnings

breaking Version 0.0.27, due to an unmerged fix for a `read1()` issue, can corrupt data or lose events when consuming long streams or fragmented JSON data, leading to `ValueError: Unterminated string...` during parsing.
Fix: Consider using a more actively maintained fork like `sseclient-py` (mpetazzoni/sseclient) or explicitly apply the patch from Issue #19 (`https://github.com/btubbs/sseclient/issues/19`) if sticking to `btubbs/sseclient`. Ensure your server sends complete, non-fragmented event data if possible.
gotcha The original `sseclient` (btubbs) is not actively maintained since 2020. While it generally works, it may lack support for newer Python versions or not have fixes for recently discovered edge cases.
Fix: For new projects or if encountering persistent issues, evaluate `sseclient-py` (on PyPI, `mpetazzoni/sseclient` on GitHub) which is an actively maintained fork with better Python 3.x and `httpx` integration.
gotcha When using `requests.get(..., stream=True)` with `sseclient`, ensure the `headers={'Accept': 'text/event-stream'}` is included. Without it, some servers might not correctly identify the request as an SSE subscription and might not stream events as expected.
Fix: Always include `headers={'Accept': 'text/event-stream'}` in your `requests.get` call to explicitly request an SSE stream.

Install

pip install sseclient Install with pip

Imports

SSEClient
```
from sseclient import SSEClient
```

Quickstart

This quickstart demonstrates how to connect to an SSE endpoint using `requests` and iterate over incoming Server-Sent Events with `SSEClient`. Each `event` object typically has `data`, and optional `id`, `event` (type), and `retry` attributes.

import requests
from sseclient import SSEClient
import os

# Replace with your actual SSE endpoint URL
sse_url = os.environ.get('SSE_ENDPOINT_URL', 'http://localhost:8000/events')

try:
    # Establish a streaming connection using requests
    response = requests.get(sse_url, stream=True, headers={'Accept': 'text/event-stream'})
    response.raise_for_status() # Raise an exception for HTTP errors

    client = SSEClient(response)

    print(f"Connected to SSE stream at {sse_url}. Listening for events...")
    for event in client.events():
        if event.data:
            print(f"Received Event (ID: {event.id}, Type: {event.event}): {event.data}")
        else:
            print(f"Received heartbeat/empty event (ID: {event.id}, Type: {event.event})")

except requests.exceptions.RequestException as e:
    print(f"Error connecting to SSE stream: {e}")
except Exception as e:
    print(f"An unexpected error occurred: {e}")

view raw JSON →