Real-Time WebSocket Streaming with the NSP Python SDK - NSP

The NSP daemon broadcasts a StateChangeNotification over WebSocket every time a probe cycle detects that an application’s state has changed. The Python SDK wraps this in an async generator — you iterate over it with async for and receive events the moment they arrive, without any polling. Streaming is the right approach when your agent needs to react quickly to changes, build monitoring loops, or trigger actions as soon as a condition appears in state.

Basic streaming with `session.watch()`

session.watch() is the recommended entry point. It returns an async context manager that yields StateChangeNotification objects.

streaming_basic.py

import asyncio
import nelieo_nsp as nsp

async def watch_gmail():
    async with nsp.NSPClient() as client:
        session = await client.attach("Gmail")

        async with session.watch() as stream:
            async for event in stream:
                print(f"Changed: {event.changed_keys}")
                print(f"Sequence: {event.sequence}")
                print(f"Captured at: {event.captured_at}")

asyncio.run(watch_gmail())

StateChangeNotification fields

Every event the stream yields is a StateChangeNotification with these fields:

Field	Type	Description
`pid`	`int`	OS process ID of the tracked application
`app_id`	`str`	Stable semantic app identifier
`sequence`	`int`	Monotonic counter. A gap means dropped events or reconnection.
`changed_keys`	`list[str]`	Dot-notation state keys that changed since the last notification
`captured_at`	`datetime`	ISO 8601 timestamp of the state capture

Refreshing state on each event

Events tell you which keys changed, but they don’t carry the new values. Call session.refresh() inside the loop to pull the updated state, or enable auto_refresh_on_event=True to have the SDK do it automatically.

async with session.watch() as stream:
    async for event in stream:
        if any(k.startswith("inbox.") for k in event.changed_keys):
            await session.refresh()
            unread = session.get_int("inbox.unread_count")
            print(f"Unread: {unread}")

async with session.watch(auto_refresh_on_event=True) as stream:
    async for event in stream:
        # session.state is always current here
        unread = session.get_int("inbox.unread_count")
        print(f"[{event.captured_at}] Unread: {unread}")

        if "inbox.emails[0]" in event.changed_keys:
            subject = session.get_str("inbox.emails[0].subject")
            print(f"New email: {subject}")

`client.watch(pid)` vs `session.watch()`

Both entry points subscribe to the same underlying WebSocket endpoint. The difference is context: session.watch() knows the session’s PID and app name, and can optionally keep the session’s cached state updated. client.watch(pid) is a raw stream — use it when you need to stream events for a PID without a full session object.

# Via session (recommended)
async with session.watch() as stream:
    async for event in stream:
        print(event.changed_keys)

# Via client (raw)
async with client.watch(pid=12847) as stream:
    async for event in stream:
        print(event.changed_keys)

Handling disconnects and reconnection

By default, session.watch() reconnects automatically if the WebSocket connection drops. You can detect reconnections by checking for gaps in event.sequence.

reconnect_detection.py

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:
                dropped = event.sequence - last_seq - 1
                print(f"⚠️  Sequence gap: {last_seq} → {event.sequence} "
                      f"({dropped} event(s) dropped or reconnection occurred)")

            last_seq = event.sequence
            print(f"seq={event.sequence}  changed={event.changed_keys}")

For scenarios where you need explicit control over reconnection timing, disable auto-reconnect and implement exponential backoff:

manual_reconnect.py

import asyncio
import nelieo_nsp as nsp

async def watch_with_backoff(session):
    backoff = 1.0
    max_backoff = 60.0

    while True:
        try:
            async with session.watch(auto_reconnect=False) as stream:
                async for event in stream:
                    backoff = 1.0   # reset on successful event
                    print(f"seq={event.sequence}  keys={event.changed_keys}")
        except nsp.NSPConnectionError:
            print(f"Connection lost. Retrying in {backoff:.0f}s...")
            await asyncio.sleep(backoff)
            backoff = min(backoff * 2, max_backoff)

Sequence gaps

event.sequence is a monotonically increasing integer. If you observe event.sequence > last_sequence + 1, one of the following occurred:

The internal event queue overflowed and the oldest events were dropped (see queue configuration below).
The WebSocket connection dropped and was re-established, causing the daemon to start a new sequence.
The daemon restarted.

Treat sequence gaps as a signal to call await session.refresh() and re-read current state rather than attempting to reconstruct what changed during the gap.

Watching multiple apps simultaneously

Combine asyncio.gather with per-session watch tasks to monitor several apps at the same time.

watch_all.py

import asyncio
import nelieo_nsp as nsp

async def watch_all():
    async with nsp.NSPClient() as client:
        apps = await client.list_apps()
        sessions = [await client.attach(app.app_name) for app in apps]

        async def watch_one(s):
            async with s.watch() as stream:
                async for event in stream:
                    print(f"[{s.app_name}] {event.changed_keys}")

        await asyncio.gather(*[watch_one(s) for s in sessions])

asyncio.run(watch_all())

Queue configuration

Events are buffered in an internal queue. If your consumer processes events more slowly than the daemon emits them, the oldest events are dropped to make room. Increase max_queue if your agent does significant work inside the loop.

async with session.watch(max_queue=512) as stream:
    async for event in stream:
        await do_heavy_work(event)   # won't overflow at 512-event buffer

Watch events do not count against your action rate limit. You can keep a stream open indefinitely without affecting how many actions per second your agent can execute.

Bounding a watch session by time

Use asyncio.timeout (Python 3.11+) to stop watching after a fixed duration:

bounded_watch.py

import asyncio
import nelieo_nsp as nsp

async def watch_for_60s(session):
    try:
        async with asyncio.timeout(60.0):
            async with session.watch() as stream:
                async for event in stream:
                    print(event.changed_keys)
    except asyncio.TimeoutError:
        print("Watch window closed after 60 seconds.")

​Basic streaming with session.watch()

​StateChangeNotification fields

​Refreshing state on each event

​client.watch(pid) vs session.watch()

​Handling disconnects and reconnection

​Sequence gaps

​Watching multiple apps simultaneously

​Queue configuration

​Bounding a watch session by time

Basic streaming with `session.watch()`

StateChangeNotification fields

Refreshing state on each event

`client.watch(pid)` vs `session.watch()`

Handling disconnects and reconnection

Sequence gaps

Watching multiple apps simultaneously

Queue configuration

Bounding a watch session by time