NSPClient Reference: The NSP Python SDK Entry Point - NSP

NSPClient is the first object you create in any NSP agent. It owns the underlying HTTP connection pool, injects your API key into every request, and exposes every operation the daemon offers — from listing tracked apps to opening a session on a specific application. You always use it as an async with context manager so the connection pool is released cleanly when your agent exits.

Constructor

class NSPClient:
    def __init__(
        self,
        api_key: str | None = None,
        *,
        host: str = "127.0.0.1",
        port: int = 7842,
        timeout: float = 30.0,
        connect_timeout: float = 5.0,
        max_connections: int = 20,
    ) -> None

Parameter	Type	Default	Description
`api_key`	`str \| None`	`None`	API key. Falls back to the `NSP_API_KEY` environment variable if omitted.
`host`	`str`	`"127.0.0.1"`	Daemon hostname.
`port`	`int`	`7842`	Daemon port.
`timeout`	`float`	`30.0`	Per-request timeout in seconds.
`connect_timeout`	`float`	`5.0`	TCP connection timeout in seconds.
`max_connections`	`int`	`20`	Maximum simultaneous HTTP connections in the pool.

Authentication

The client reads your API key in one of two ways. Set the NSP_API_KEY environment variable and the client picks it up automatically, or pass api_key= directly to the constructor.

# Via environment variable (recommended)
# export NSP_API_KEY=nsp_live_sk_a1b2c3d4...
async with nsp.NSPClient() as client:
    apps = await client.list_apps()

# Explicit key
async with nsp.NSPClient(api_key="nsp_live_sk_a1b2c3d4...") as client:
    apps = await client.list_apps()

Using NSPClient as an async context manager

Always open NSPClient with async with. The context manager starts the connection pool on entry and drains it on exit, even if an exception is raised inside the block.

client_basic.py

import asyncio
import nelieo_nsp as nsp

async def main():
    async with nsp.NSPClient() as client:
        session = await client.attach("Gmail")
        unread = session.get_int("inbox.unread_count")
        print(f"Unread: {unread}")

asyncio.run(main())

For long-lived processes where you cannot use async with, call await client.aclose() manually in a finally block.

Methods

`ping() → bool`

Check whether the daemon is reachable. This method never raises — it returns False on any connection or authentication failure, making it safe as a pre-flight health check.

is_up = await client.ping()
if not is_up:
    print("Daemon not running. Start it with: axon-daemon.exe run")

`list_apps() → list[AppSummary]`

Return all applications currently tracked by the daemon. This call returns the daemon’s cached list and does not trigger new probes. The daemon updates the list automatically as processes start and stop.

apps = await client.list_apps()

for app in apps:
    print(
        f"{app.app_name:25s}  pid={app.pid:6d}  "
        f"runtime={app.runtime:6s}  conf={app.confidence:.2f}  "
        f"status={app.probe_status}"
    )

Each AppSummary has these fields:

Field	Type	Description
`pid`	`int`	OS process ID
`app_id`	`str`	Stable semantic identifier — survives daemon restarts
`app_name`	`str`	Human-readable name
`runtime`	`str`	`v8` / `jvm` / `clr` / `native` / `unknown`
`probe_status`	`str`	`detecting` / `attaching` / `cold_probe` / `active` / `error` / `detached`
`confidence`	`float`	State quality score 0.0–1.0
`object_count`	`int`	Number of SSF state objects
`action_count`	`int`	Number of available actions
`has_state`	`bool`	Whether a state snapshot is available

Apps with has_state: False are still in the cold probe phase. Wait for probe_status to become "active" before reading state.

`get_state(pid, *, fresh=False) → SubstrateState`

Fetch the semantic state snapshot for a specific process. By default this returns the daemon’s cached state in under a millisecond. Pass fresh=True to trigger a new probe cycle, which adds 50ms to 90 seconds of latency depending on the runtime.

# Cached — fast
state = await client.get_state(12847)
print(state.confidence)                           # 0.97
print(state.objects.get("inbox.unread_count"))    # 47

# Fresh — forces a new probe cycle
state = await client.get_state(12847, fresh=True)

Raises: NSPNotFoundError if the PID is not currently tracked.

`attach(app_name) → Session`

Attach to a running application by name and open a Session. This is the primary way to work with any app — you never need to look up PIDs manually.

# Exact name
session = await client.attach("Gmail")

# Partial match (fuzzy matching is on by default)
session = await client.attach("mail")         # matches "Gmail"
session = await client.attach("mission")      # matches "Mission Planner"

# Stable app_id
session = await client.attach("gmail-chrome-12847")

# Force a fresh probe on attach
session = await client.attach("Gmail", fresh=True)

Raises: NSPNotFoundError if no tracked app matches the name.

`watch(pid) → AsyncStream`

Open a raw WebSocket stream for a PID. This is the lower-level entry point for streaming — for most use cases, prefer session.watch() which also keeps the session’s cached state updated.

async with client.watch(pid=12847) as stream:
    async for event in stream:
        print(f"Changed: {event.changed_keys}")
        print(f"Sequence: {event.sequence}")

Complete example

The example below shows the full lifecycle of an NSPClient — connecting, discovering apps, reading state, and executing an action.

monitor_gmail.py

import asyncio
import nelieo_nsp as nsp

async def monitor_gmail():
    async with nsp.NSPClient(api_key="nsp_live_sk_...") as client:
        # Confirm daemon is reachable
        if not await client.ping():
            raise RuntimeError("Daemon not running")

        # List what is tracked
        apps = await client.list_apps()
        print(f"Tracking {len(apps)} app(s):")
        for app in apps:
            print(f"  {app.app_name}  runtime={app.runtime}  conf={app.confidence:.2f}")

        # Attach to Gmail
        session = await client.attach("Gmail")
        print(f"\nAttached: {session.app_name} (conf={session.confidence:.2f})")

        unread = session.get_int("inbox.unread_count")
        print(f"Unread emails: {unread}")

        # Archive the first email (reversible — no verify needed)
        email_id = session.get_str("inbox.emails[0].id")
        result = await session.execute(
            "archive",
            parameters={"email_id": email_id},
        )
        print(f"Archived: {result.success}  latency={result.latency_ms}ms")

asyncio.run(monitor_gmail())

​Constructor

​Authentication

​Using NSPClient as an async context manager

​Methods

​ping() → bool

​list_apps() → list[AppSummary]

​get_state(pid, *, fresh=False) → SubstrateState

​attach(app_name) → Session

​watch(pid) → AsyncStream

​Complete example

Constructor

Authentication

Using NSPClient as an async context manager

Methods

`ping() → bool`

`list_apps() → list[AppSummary]`

`get_state(pid, *, fresh=False) → SubstrateState`

`attach(app_name) → Session`

`watch(pid) → AsyncStream`

Complete example