NSPClient: Connecting to and Controlling the Daemon - NSP

NSPClient is the main entry point for the nelieo-nsp SDK. Every interaction with the NSP daemon starts here. It manages the underlying httpx.AsyncClient connection pool, handles API key authentication, and exposes methods for discovering applications, reading state, and opening sessions.

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,
        verify_ssl: bool = True,
    ) -> None

api_key

str | None

default:"None"

Your NSP API key. If omitted, the client reads the NSP_API_KEY environment variable. Raises NSPAuthError on the first request if neither is set.

host

str

default:"127.0.0.1"

Hostname or IP address of the NSP daemon. Override with the NSP_HOST environment variable.

port

int

default:"7842"

Port the daemon is listening on. Override with the NSP_PORT environment variable.

timeout

float

default:"30.0"

Per-request timeout in seconds. Applies to all HTTP requests except the initial TCP handshake.

connect_timeout

float

default:"5.0"

TCP connection timeout in seconds. Raises NSPConnectionError if the daemon is not reachable within this window.

max_connections

int

default:"20"

Maximum number of simultaneous HTTP connections in the pool. Increase this if you are making many concurrent requests.

verify_ssl

bool

default:"True"

Whether to verify SSL certificates. Set to False only for local development with a self-signed cert.

Lifecycle

NSPClient is an async context manager. Use async with to ensure the underlying connection pool is cleanly closed when you are done:

import asyncio
import nelieo_nsp as nsp

async def main():
    async with nsp.NSPClient(api_key="sk_nsp_...") as client:
        apps = await client.list_apps()
        # client is automatically closed when the block exits

asyncio.run(main())

For long-lived clients — for example, inside a web server — manage the lifecycle manually:

client = nsp.NSPClient(api_key="sk_nsp_...")
await client.__aenter__()

try:
    apps = await client.list_apps()
finally:
    await client.aclose()

Methods

`health() → dict`

Return daemon version, uptime, and basic health metadata. Use this to confirm the daemon is running before issuing other requests.

info = await client.health()
print(f"Version : {info['version']}")
print(f"Uptime  : {info['uptime_seconds']}s")

version

str

Daemon version string, e.g. "1.0.0".

uptime_seconds

int

Seconds since the daemon started.

`list_apps() → list[AppSummary]`

List every application currently tracked by the daemon. Returns an empty list if no apps are tracked.

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}"
    )

list_apps() returns cached data from the daemon’s internal registry. It does not trigger a fresh probe. The daemon updates this list automatically as processes start and stop.

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

Fetch the current semantic state for a tracked process by PID.

pid

int

required

The OS process ID of the target application.

fresh

bool

default:"False"

When True, forces the daemon to run a fresh probe cycle before returning. This adds 50 ms–90 s of latency depending on the runtime (v8 is fast; CLR and JVM are slower on first attach).

# Use cached state — fast (~1ms round-trip)
state = await client.get_state(12847)
print(f"Confidence : {state.confidence:.2f}")
print(f"Unread     : {state.get_int('inbox.unread_count')}")
print(f"Latest     : {state.get_str('inbox.emails[0].subject')}")

# Force a fresh probe — accurate, but slower
state = await client.get_state(12847, fresh=True)

Raises: NSPNotFoundError if the PID is not in the daemon’s tracked list.

`attach(app_name_or_id, *, fresh=False, fuzzy=True) → NSPSession`

Attach to a running application by name or stable app_id. This is the most convenient way to open a session.

app_name_or_id

str

required

The human-readable app name (e.g. "Gmail") or the stable app_id string (e.g. "gmail-chrome-12847").

fresh

bool

default:"False"

Trigger a fresh probe before the session opens. Useful when you need accurate initial state immediately.

fuzzy

bool

default:"True"

Allow partial name matching. When True, "mail" matches "Gmail" and "mission" matches "Mission Planner".

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

# Partial match (fuzzy=True by default)
session = await client.attach("mail")       # matches "Gmail"
session = await client.attach("planner")    # matches "Mission Planner"

# Stable app_id match (never changes across restarts)
session = await client.attach("gmail-chrome-12847")

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

Raises: NSPNotFoundError if no tracked app matches the given name or ID.

`attach_by_pid(pid, *, fresh=False) → NSPSession`

Attach to a specific process by exact PID. Use this when you already know the PID from list_apps().

pid

int

required

The exact OS process ID.

apps = await client.list_apps()
pid  = apps[0].pid

session = await client.attach_by_pid(pid)

`wait_for_app(app_name_or_id, *, timeout=60.0, poll_interval=2.0, fuzzy=True) → NSPSession`

Poll the daemon until a matching app appears, then attach and return the session. Use this when your agent starts before the target application.

app_name_or_id

str

required

App name or stable app_id to wait for.

timeout

float

default:"60.0"

Maximum seconds to wait before raising asyncio.TimeoutError.

poll_interval

float

default:"2.0"

Seconds between each poll to list_apps().

fuzzy

bool

default:"True"

Allow partial name matching.

# Wait up to 30 seconds for Mission Planner to launch
session = await client.wait_for_app("Mission Planner", timeout=30.0)
print(f"Attached: {session.app_name}")

Raises: asyncio.TimeoutError if the app does not appear within timeout seconds.

`execute_action(request) → ActionResponse`

Low-level action dispatch. For most use cases, prefer session.execute() — it handles PID binding and state refresh automatically.

from nelieo_nsp import ActionRequest

request = ActionRequest(
    pid=12847,
    action="archive",
    parameters={"email_id": "msg_18821af3"},
    verify_expression="inbox.emails[0].id != 'msg_18821af3'",
)
response = await client.execute_action(request)
print(f"Success : {response.success}")
print(f"Latency : {response.latency_ms}ms")

Complete Example

import asyncio
import nelieo_nsp as nsp

async def monitor_gmail():
    async with nsp.NSPClient(api_key="sk_nsp_...") as client:
        # Confirm daemon is healthy
        info = await client.health()
        print(f"Daemon {info['version']} running for {info['uptime_seconds']}s")

        # Wait for Gmail to appear (Chrome may still be launching)
        print("Waiting for Gmail...")
        session = await client.wait_for_app("Gmail", timeout=30.0)
        print(f"Attached: {session.app_name}  conf={session.confidence:.2f}")

        # Read initial state
        unread = session.get_int("inbox.unread_count")
        print(f"Unread: {unread}")

        # Show every available action and its reversibility class
        print(f"Actions ({len(session.actions)}):")
        for name, schema in session.actions.items():
            flag = " ⚠ IRREVERSIBLE" if schema.is_destructive else ""
            print(f"  {name}: {schema.signature}{flag}")

        # Stream state changes for 60 seconds
        print("Streaming for 60s (Ctrl-C to stop)...")
        async with asyncio.timeout(60.0):
            async with session.watch() as stream:
                async for event in stream:
                    await session.refresh()
                    new_unread = session.get_int("inbox.unread_count")
                    print(f"  changed={event.changed_keys}  unread={new_unread}")

asyncio.run(monitor_gmail())

​Constructor

​Lifecycle

​Methods

​health() → dict

​list_apps() → list[AppSummary]

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

​attach(app_name_or_id, *, fresh=False, fuzzy=True) → NSPSession

​attach_by_pid(pid, *, fresh=False) → NSPSession

​wait_for_app(app_name_or_id, *, timeout=60.0, poll_interval=2.0, fuzzy=True) → NSPSession

​execute_action(request) → ActionResponse

​Complete Example

Constructor

Lifecycle

Methods

`health() → dict`

`list_apps() → list[AppSummary]`

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

`attach(app_name_or_id, *, fresh=False, fuzzy=True) → NSPSession`

`attach_by_pid(pid, *, fresh=False) → NSPSession`

`wait_for_app(app_name_or_id, *, timeout=60.0, poll_interval=2.0, fuzzy=True) → NSPSession`

`execute_action(request) → ActionResponse`

Complete Example