> ## Documentation Index
> Fetch the complete documentation index at: https://docs.nelieo.com/llms.txt
> Use this file to discover all available pages before exploring further.
# React to Live State Changes with WebSocket Streaming
> Use session.watch() to subscribe to NSP state changes, handle StateChangeNotification events, reconnect on drops, and access the raw WebSocket endpoint.
The NSP daemon emits a `StateChangeNotification` over WebSocket every time a tracked application's state changes. Instead of polling `session.refresh()` in a loop and burning through your rate limit budget, you open a single persistent connection and react to changes the instant they arrive.
## Why Streaming Beats Polling
| Approach | Reaction Latency | CPU Overhead | Rate Limit Impact |
| -------------------- | ---------------- | ------------ | -------------------- |
| Poll every 500ms | 0–500ms avg | High | 2 req/s per app |
| Poll every 100ms | 0–100ms avg | Very high | 10 req/s per app |
| **WebSocket stream** | **50–100ms** | **Minimal** | **Not rate-limited** |
The `/watch` WebSocket endpoint is not subject to the daemon's rate limiter. You can hold as many concurrent streams open as you need without affecting your API key's request quota.
## The `StateChangeNotification` Object
Every event your stream yields is a `StateChangeNotification` with these fields:
OS process ID of the app that changed.
Stable semantic identifier for the app (e.g. `"gmail-chrome-12847"`).
Monotonically increasing counter. A gap between two consecutive sequence numbers means events were dropped (queue overflow) or a reconnect occurred.
State keys that changed since the previous notification. Use this to filter work — only process events relevant to your agent.
Timestamp at which the daemon captured the changed state.
## Basic Streaming with `session.watch()`
The recommended pattern is `session.watch(auto_refresh_on_event=True)`. This refreshes `session.state` automatically on every event, so you can call `session.get_*()` directly inside the loop without a separate `refresh()` call:
```python theme={null}
import asyncio
import nelieo_nsp as nsp
async def main():
async with nsp.NSPClient(api_key="sk_nsp_...") as client:
session = await client.attach("Gmail")
async with session.watch(auto_refresh_on_event=True) as stream:
async for event in stream:
# Filter: only act on inbox changes
if "inbox.unread_count" not in event.changed_keys:
continue
unread = session.get_int("inbox.unread_count")
print(f"[{event.captured_at}] Unread count → {unread}")
if unread > 0:
await handle_new_email(session)
async def handle_new_email(session):
subject = session.get_str("inbox.emails[0].subject")
from_addr = session.get_str("inbox.emails[0].from")
thread_id = session.get_str("inbox.emails[0].thread_id")
print(f"New: '{subject}' from {from_addr}")
# Auto-archive newsletters
if "newsletter" in subject.lower():
email_id = session.get_str("inbox.emails[0].id")
await session.execute("archive", parameters={"email_id": email_id})
asyncio.run(main())
```
## Filtering with `changed_keys`
Always check `event.changed_keys` before doing any work. Most state change events will touch fields your agent doesn't care about:
```python theme={null}
async with session.watch(auto_refresh_on_event=True) as stream:
async for event in stream:
# Only process inbox-related changes
if not any(k.startswith("inbox.") for k in event.changed_keys):
continue
# Process new email at position 0
if "inbox.emails[0].id" in event.changed_keys:
subject = session.get_str("inbox.emails[0].subject")
print(f"🔔 New email: {subject}")
```
## Reconnection Handling
By default `session.watch()` reconnects automatically if the WebSocket connection drops. You can detect reconnections by watching for gaps in `event.sequence`:
```python theme={null}
async def watch_with_reconnect_detection(session):
last_seq = 0
async with session.watch(auto_reconnect=True) as stream:
async for event in stream:
if last_seq > 0 and event.sequence > last_seq + 1:
missed = event.sequence - last_seq - 1
print(f"⚠️ Reconnected — missed {missed} event(s), refreshing state")
# Force a fresh probe to get current state after the gap
await session.refresh(fresh_probe=True)
last_seq = event.sequence
print(f"[seq={event.sequence}] {event.changed_keys}")
```
After detecting a sequence gap, call `session.refresh(fresh_probe=True)` rather than relying on cached state. The gap means you missed some transitions — a fresh probe gives you the current ground truth.
## Combining Streaming with `session.refresh()`
For most agents, `auto_refresh_on_event=True` is all you need. If you need to force a full re-probe (for example, after acting on the state), call `refresh()` explicitly:
```python theme={null}
async with session.watch() as stream:
async for event in stream:
if "inbox.unread_count" not in event.changed_keys:
continue
# Manually refresh before reading (no auto_refresh_on_event)
await session.refresh()
unread = session.get_int("inbox.unread_count")
if unread > 0:
email_id = session.get_str("inbox.emails[0].id")
# Execute an action, then force a fresh probe to confirm outcome
await session.execute("mark_read", parameters={"email_id": email_id})
await session.refresh(fresh_probe=True)
print(f"Marked as read. Still unread: {session.get_int('inbox.unread_count')}")
```
## Watching Multiple Apps Simultaneously
Open one stream per app and gather them with `asyncio.gather`:
```python theme={null}
import asyncio
import nelieo_nsp as nsp
async def watch_app(session):
async with session.watch(auto_refresh_on_event=True) as stream:
async for event in stream:
await react(session, event)
async def react(session, event):
if session.app_name == "Gmail":
if "inbox.unread_count" in event.changed_keys:
print(f" Gmail unread: {session.get_int('inbox.unread_count')}")
elif session.app_name == "Mission Planner":
if "flight.altitude_m" in event.changed_keys:
print(f" Altitude: {session.get_float('flight.altitude_m'):.1f}m")
async def main():
async with nsp.NSPClient(api_key="sk_nsp_...") as client:
apps = await client.list_apps()
sessions = []
for app in apps:
if app.has_state and app.confidence >= 0.80:
s = await client.session(app.pid)
sessions.append(s)
print(f"Watching {len(sessions)} app(s)...")
await asyncio.gather(*[watch_app(s) for s in sessions])
asyncio.run(main())
```
## Streaming + Periodic Polling Together
Sometimes you want real-time reaction to events *and* a periodic summary. Run both patterns concurrently with `asyncio.gather`:
```python theme={null}
import asyncio
import time
import nelieo_nsp as nsp
async def periodic_summary(session, interval_secs: int = 30):
"""Log an inbox summary every 30 seconds."""
while True:
try:
await session.refresh()
unread = session.get_int("inbox.unread_count")
print(f"[{time.strftime('%H:%M:%S')}] Summary — Unread: {unread}")
except nsp.NSPRateLimitError:
await asyncio.sleep(5)
continue
await asyncio.sleep(interval_secs)
async def reactive_handler(session):
"""React immediately to new email arrivals."""
async with session.watch(auto_refresh_on_event=True) as stream:
async for event in stream:
if "inbox.emails[0].id" in event.changed_keys:
subject = session.get_str("inbox.emails[0].subject")
print(f"🔔 New email: {subject}")
async def main():
async with nsp.NSPClient(api_key="sk_nsp_...") as client:
session = await client.attach("Gmail")
# Run both patterns concurrently
await asyncio.gather(
periodic_summary(session),
reactive_handler(session),
)
asyncio.run(main())
```
## Low-Level Access: `open_state_stream()`
If you need direct WebSocket access — for example, to forward events to a non-Python consumer or to build your own buffering layer — use the low-level `client.open_state_stream()` function:
```python theme={null}
async with nsp.NSPClient(api_key="sk_nsp_...") as client:
apps = await client.list_apps()
target_pid = apps[0].pid
# Returns a raw StateStream without the session convenience layer
async with client.watch(pid=target_pid) as stream:
async for event in stream:
print(f"pid={event.pid} seq={event.sequence} keys={event.changed_keys}")
```
### Raw WebSocket endpoint
The underlying WebSocket URL is:
```text theme={null}
ws://localhost:7842/substrate/v1/watch/{pid}?key=sk_nsp_...
```
You can connect to this directly from any WebSocket client. Each message is a JSON object:
```json theme={null}
{
"pid": 12847,
"app_id": "gmail-chrome-12847",
"sequence": 142,
"changed_keys": ["inbox.unread_count", "inbox.emails[0].is_read"],
"captured_at": "2026-07-20T13:00:00.123Z"
}
```
To test the stream from the command line with `websocat`:
```bash theme={null}
websocat "ws://localhost:7842/substrate/v1/watch/12847?key=sk_nsp_..."
```
The raw WebSocket endpoint does not validate `changed_keys` against an app schema. If the session is torn down and re-created (e.g. the app restarts), `prev_capture_id` may not be meaningful across the boundary. Use `sequence` gaps to detect these transitions.
## `session.watch()` Parameter Reference
When `true`, the session's cached state is automatically refreshed every time an event arrives. This is the most ergonomic option for most agents.
Automatically re-establishes the WebSocket connection if it drops. Sequence gaps indicate where reconnects occurred.
Size of the internal event buffer. When the buffer is full, the oldest event is dropped. Increase this if your handler is slower than the daemon's emit rate.