aiohttp-sse-client

0.2.1 · maintenance · verified Thu Apr 16

aiohttp-sse-client is a Python client library for consuming Server-Sent Events (SSE) streams, built on top of `aiohttp`. It provides a simple asynchronous interface for connecting to SSE sources, processing incoming events, and handling automatic reconnections. The current version is 0.2.1, released in February 2021, indicating an infrequent release cadence. Note that a fork, `aiohttp-sse-client2`, exists to address maintenance and support newer Python versions.

Common errors

```
asyncio.TimeoutError: Operation timed out
```
cause The default `aiohttp.ClientSession` timeout (usually 5 minutes) expires on a long-lived SSE connection that appears idle to the client, even if the server is sending occasional keep-alives.

fix Pass a `aiohttp.ClientSession` with a `aiohttp.ClientTimeout(total=None)` or a very large `total` value to `EventSource` to disable or extend the timeout. (e.g., `session = aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=None)); EventSource(url, session=session)`).
```
ClientOSError: [Errno 104] Connection reset by peer
```
cause The server or an intermediary closed the connection unexpectedly, or the client-side timeout in `aiohttp.ClientSession` was reached, causing the connection to be forcefully closed.

fix Implement robust `try...except ConnectionError` or specific `aiohttp.ClientConnectorError` handling. For long-running streams, also configure `aiohttp.ClientTimeout(total=None)` to prevent client-side timeouts from causing resets.

Warnings

breaking Version 0.2.0 dropped official support for Python 3.5. Subsequent versions (including 0.2.1) require Python 3.6 or newer.
Fix: Ensure your environment uses Python 3.6 or newer. For Python 3.7+ support, you might consider the `aiohttp-sse-client2` fork.
gotcha The underlying `aiohttp.ClientSession` used by `EventSource` has a default total timeout (typically 5 minutes). For long-lived SSE connections, this can lead to `asyncio.TimeoutError` or `ClientOSError: [Errno 104] Connection reset by peer` even with server-side keep-alives.
Fix: When initializing `EventSource`, pass a custom `aiohttp.ClientSession` with an explicitly configured timeout. For indefinite connections, use `aiohttp.ClientTimeout(total=None)`, or a sufficiently large numeric value. Example: `session = aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=None)); sse_client.EventSource(url, session=session)`.
gotcha Asynchronous network operations are inherently prone to disconnections. Client applications must implement robust error handling and reconnection logic to gracefully manage these failures.
Fix: Implement comprehensive `try...except` blocks around the event iteration to handle `ConnectionError`, `aiohttp.ClientConnectorError`, `asyncio.TimeoutError`, and other network-related exceptions. Consider adding retry mechanisms with exponential backoff if a simple `pass` is insufficient for your use case.

Install

pip install aiohttp-sse-client Install stable release

Imports

EventSource
```
from aiohttp_sse_client import client as sse_client # then use sse_client.EventSource
```
The primary class for connecting to and consuming an SSE stream. It's typically imported with an alias for brevity.

Quickstart

This quickstart demonstrates connecting to a public SSE stream and iterating over incoming events. It includes robust error handling for common connection issues and configures `aiohttp.ClientTimeout` for long-running connections to prevent premature timeouts.

import asyncio
import aiohttp
from aiohttp_sse_client import client as sse_client

async def listen_to_sse(url):
    session = aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=None))
    try:
        async with sse_client.EventSource(url, session=session) as event_source:
            print(f"Connected to SSE stream at {url}")
            async for event in event_source:
                print(f"Event ID: {event.id}, Type: {event.type}, Data: {event.data}")
    except aiohttp.ClientConnectorError as e:
        print(f"Connection error: {e}")
    except ConnectionError as e:
        print(f"Stream disconnected: {e}")
    except asyncio.TimeoutError:
        print("SSE stream timed out. Ensure server sends keep-alives or configure ClientTimeout.")
    finally:
        await session.close()

async def main():
    # Example public SSE stream (e.g., Wikimedia recent changes)
    sse_url = "https://stream.wikimedia.org/v2/stream/recentchange"
    await listen_to_sse(sse_url)

if __name__ == '__main__':
    asyncio.run(main())

view raw JSON →