React to Live State Changes with WebSocket Streaming

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:

pid

int

OS process ID of the app that changed.

app_id

str

Stable semantic identifier for the app (e.g. "gmail-chrome-12847").

sequence

int

Monotonically increasing counter. A gap between two consecutive sequence numbers means events were dropped (queue overflow) or a reconnect occurred.

changed_keys

list[str]

State keys that changed since the previous notification. Use this to filter work — only process events relevant to your agent.

captured_at

datetime

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:

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:

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:

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:

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:

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:

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:

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:

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:

{
  "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:

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

auto_refresh_on_event

bool

default:"false"

When true, the session’s cached state is automatically refreshed every time an event arrives. This is the most ergonomic option for most agents.

auto_reconnect

bool

default:"true"

Automatically re-establishes the WebSocket connection if it drops. Sequence gaps indicate where reconnects occurred.

max_queue

int

default:"128"

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.

Get Started

The Daemon

Runtime Probes

Security

Guides

Why Streaming Beats Polling

The `StateChangeNotification` Object

Basic Streaming with `session.watch()`

Filtering with `changed_keys`

Reconnection Handling

Combining Streaming with `session.refresh()`

Watching Multiple Apps Simultaneously

Streaming + Periodic Polling Together

Low-Level Access: `open_state_stream()`

Raw WebSocket endpoint

`session.watch()` Parameter Reference

​Why Streaming Beats Polling

​The StateChangeNotification Object

​Basic Streaming with session.watch()

​Filtering with changed_keys

​Reconnection Handling

​Combining Streaming with session.refresh()

​Watching Multiple Apps Simultaneously

​Streaming + Periodic Polling Together

​Low-Level Access: open_state_stream()

​Raw WebSocket endpoint

​session.watch() Parameter Reference

Why Streaming Beats Polling

The `StateChangeNotification` Object

Basic Streaming with `session.watch()`

Filtering with `changed_keys`

Reconnection Handling

Combining Streaming with `session.refresh()`

Watching Multiple Apps Simultaneously

Streaming + Periodic Polling Together

Low-Level Access: `open_state_stream()`

Raw WebSocket endpoint

`session.watch()` Parameter Reference