> ## Documentation Index
> Fetch the complete documentation index at: https://docs.westyx.dev/llms.txt
> Use this file to discover all available pages before exploring further.

# Python SDK - Stream Observer

> Opt-in structured callbacks for SSE lifecycle events in the Python SDK.

Pass a `StreamObserver` via `NexusConfig.stream_observer` to receive structured callbacks for every SSE lifecycle transition. The observer is the **canonical** way to wire SDK state into ops dashboards, custom telemetry, or load-test tooling - far more reliable than scraping `NexusLogger` output by substring match.

The observer is **opt-in**. When `NexusConfig.stream_observer` is `None` the SDK uses `NOOP_STREAM_OBSERVER` at zero overhead.

## The interface

```python theme={null}
from datetime import datetime
from typing import Optional, Protocol, runtime_checkable

@runtime_checkable
class StreamObserver(Protocol):
    def on_connected(self) -> None: ...
    def on_disconnected(self, cause: Optional[BaseException]) -> None: ...
    def on_event(self, name: str) -> None: ...
    def on_reconnect_attempt(self, attempt: int) -> None: ...
    def on_fallback(self, reason: str) -> None: ...
    def on_quarantined(self, reason: str, expires_at: datetime) -> None: ...  # v0.4.0
```

The protocol is structural - any object with the matching method signatures qualifies. Methods are looked up via `getattr`, so consumers may omit any method they don't care about (the SDK handles missing-attribute cases gracefully).

## Hook semantics

| Hook                                            | Fires when                                                                                      | Argument                                                       |
| ----------------------------------------------- | ----------------------------------------------------------------------------------------------- | -------------------------------------------------------------- |
| `on_connected()`                                | `GET /v1/stream` returns `200 OK` and the SDK is ready to read events                           | none                                                           |
| `on_disconnected(cause)`                        | An active stream connection ends - clean disconnect, transport error, or server-initiated close | `None` for clean shutdown, the underlying exception otherwise  |
| `on_event(name)`                                | A whitelisted SSE event arrives, **before** the SDK runs the event-triggered sync               | event name string (e.g. `"config.updated"`)                    |
| `on_reconnect_attempt(n)`                       | The SDK is about to back off and retry after a transport error                                  | 1-based attempt counter (max 3 with current threshold)         |
| `on_fallback(reason)`                           | The SDK enters a degraded state                                                                 | `FALLBACK_REASON_MAX_ERRORS` or `FALLBACK_REASON_RATE_LIMITED` |
| `on_quarantined(reason, expires_at)` *(v0.4.0)* | Server returned `429` with a quarantine body on either the sync or stream path                  | `reason` string; `expires_at` timezone-aware `datetime`        |

### `on_fallback` reasons

```python theme={null}
from westyx_nexus import (
    FALLBACK_REASON_MAX_ERRORS,
    FALLBACK_REASON_RATE_LIMITED,
)
```

| Constant         | When                                                                 | Stream after                                            |
| ---------------- | -------------------------------------------------------------------- | ------------------------------------------------------- |
| `'max-errors'`   | 3 consecutive transport errors - backoff exhausted, the SDK gives up | TTL polling only; reconnect requires `connect_stream()` |
| `'rate-limited'` | server returned `429 Too Many Requests` (tier connection-limit hit)  | depends on the response - see below                     |

**`rate-limited` semantics (v0.3.1+).** This reason now fires in **both** of these situations:

* **Server included a parseable `Retry-After` header.** The SDK fires `on_fallback(FALLBACK_REASON_RATE_LIMITED)`, sleeps the indicated delay clamped to `[5, 300]` seconds, and then reconnects automatically. The 429 does **not** count against `_MAX_STREAM_ERRORS`. Expect a matching `on_connected()` shortly after the sleep window.
* **No parseable `Retry-After`.** Legacy behaviour: the SDK fires `on_fallback(FALLBACK_REASON_RATE_LIMITED)`, closes the stream, and stays on TTL polling. No automatic reconnect - `connect_stream()` is required to recover.

## Threading contract

For the **synchronous** `NexusClient`, observer methods are called **synchronously from the SDK's stream thread**. For the **async** `AsyncNexusClient`, observer methods are called **synchronously from the asyncio task** running the stream - they are NOT awaited even if you define them as coroutines.

<Warning>
  In both cases observer methods MUST return promptly. The SDK does not protect against blocking observers.
</Warning>

If your observer does anything substantial (network I/O, lock contention, disk writes), copy the argument and dispatch the work to your own thread / task before returning:

```python theme={null}
import queue, threading
class SlowObserver:
    def __init__(self) -> None:
        self.q: queue.Queue[str] = queue.Queue(maxsize=1024)
        threading.Thread(target=self._drain, daemon=True).start()

    def on_event(self, name: str) -> None:
        try:
            self.q.put_nowait(name)
        except queue.Full:
            pass  # Drop on overflow - backpressure on the SDK would stall reads.

    def _drain(self) -> None:
        while True:
            event = self.q.get()
            ... # process(event)
```

## If your observer raises

*Changed in v0.11.0.*

An exception from an observer method is caught and absorbed - a faulty observer can never break the stream, stall a sync, or propagate into your own code paths.

It is also **logged**. The SDK reports it through [`NexusLogger`](/sdks/python/configuration) at error level, naming the hook that failed and attaching the original exception as the cause:

```
nexus: stream observer on_connected() raised - ignoring: division by zero
```

<Warning>
  Before v0.11.0 the exception was discarded with no trace anywhere, so an observer that threw on every event was indistinguishable from one that was never called. If you set no logger the fault is still absorbed - it just goes unreported, as before.
</Warning>

This follows the [OpenTelemetry error-handling specification](https://opentelemetry.io/docs/specs/otel/error-handling/), which covers the same situation from both sides: "API methods that accept external callbacks MUST handle all errors", and "whenever the library suppresses an error that would otherwise have been exposed to the user, the library SHOULD log the error".

A missing hook is not a fault - the observer protocol is partial by design. Implement only the callbacks you need; the rest are skipped silently.

## Counter pattern

```python theme={null}
from dataclasses import dataclass
from threading import Lock
from westyx_nexus import NexusClient, NexusConfig

@dataclass
class CounterObserver:
    events: int = 0
    reconnects: int = 0
    fallbacks: int = 0
    _lock: Lock = Lock()

    def on_event(self, name: str) -> None:
        with self._lock:
            self.events += 1

    def on_reconnect_attempt(self, attempt: int) -> None:
        with self._lock:
            self.reconnects += 1

    def on_fallback(self, reason: str) -> None:
        with self._lock:
            self.fallbacks += 1

obs = CounterObserver()
client = NexusClient.create(NexusConfig(
    base_url="https://...",
    api_key="wxs_...",
    stream_observer=obs,
))
```

## Active-connection gauge pattern

```python theme={null}
class GaugeObserver:
    def __init__(self) -> None:
        self.connected = False

    def on_connected(self) -> None:
        self.connected = True

    def on_disconnected(self, cause) -> None:
        self.connected = False

    def on_fallback(self, reason: str) -> None:
        self.connected = False
```

`on_fallback` is included because after the 3-strike `FALLBACK_REASON_MAX_ERRORS` case the SDK will not reconnect - the gauge should stay `False` until you call `connect_stream()` again. For `FALLBACK_REASON_RATE_LIMITED` with a `Retry-After` (v0.3.1+) the SDK *does* reconnect on its own; the gauge will flip back to `True` on the next `on_connected()`.