> ## 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.

# NSPClient: Connecting to and Controlling the Daemon

> Full reference for NSPClient — constructor parameters, lifecycle management, and every method for listing apps, reading state, and attaching sessions.

`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

```python theme={null}
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
```

<ParamField path="api_key" type="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.
</ParamField>

<ParamField path="host" type="str" default="127.0.0.1">
  Hostname or IP address of the NSP daemon. Override with the `NSP_HOST` environment variable.
</ParamField>

<ParamField path="port" type="int" default="7842">
  Port the daemon is listening on. Override with the `NSP_PORT` environment variable.
</ParamField>

<ParamField path="timeout" type="float" default="30.0">
  Per-request timeout in seconds. Applies to all HTTP requests except the initial TCP handshake.
</ParamField>

<ParamField path="connect_timeout" type="float" default="5.0">
  TCP connection timeout in seconds. Raises `NSPConnectionError` if the daemon is not reachable within this window.
</ParamField>

<ParamField path="max_connections" type="int" default="20">
  Maximum number of simultaneous HTTP connections in the pool. Increase this if you are making many concurrent requests.
</ParamField>

<ParamField path="verify_ssl" type="bool" default="True">
  Whether to verify SSL certificates. Set to `False` only for local development with a self-signed cert.
</ParamField>

## Lifecycle

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

```python theme={null}
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:

```python theme={null}
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.

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

<ResponseField name="version" type="str">
  Daemon version string, e.g. `"1.0.0"`.
</ResponseField>

<ResponseField name="uptime_seconds" type="int">
  Seconds since the daemon started.
</ResponseField>

***

### `list_apps() → list[AppSummary]`

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

```python theme={null}
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}"
    )
```

<Note>
  `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.
</Note>

***

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

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

<ParamField path="pid" type="int" required>
  The OS process ID of the target application.
</ParamField>

<ParamField path="fresh" type="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).
</ParamField>

```python theme={null}
# 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.

<ParamField path="app_name_or_id" type="str" required>
  The human-readable app name (e.g. `"Gmail"`) or the stable `app_id` string (e.g. `"gmail-chrome-12847"`).
</ParamField>

<ParamField path="fresh" type="bool" default="False">
  Trigger a fresh probe before the session opens. Useful when you need accurate initial state immediately.
</ParamField>

<ParamField path="fuzzy" type="bool" default="True">
  Allow partial name matching. When `True`, `"mail"` matches `"Gmail"` and `"mission"` matches `"Mission Planner"`.
</ParamField>

```python theme={null}
# 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()`.

<ParamField path="pid" type="int" required>
  The exact OS process ID.
</ParamField>

```python theme={null}
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.

<ParamField path="app_name_or_id" type="str" required>
  App name or stable app\_id to wait for.
</ParamField>

<ParamField path="timeout" type="float" default="60.0">
  Maximum seconds to wait before raising `asyncio.TimeoutError`.
</ParamField>

<ParamField path="poll_interval" type="float" default="2.0">
  Seconds between each poll to `list_apps()`.
</ParamField>

<ParamField path="fuzzy" type="bool" default="True">
  Allow partial name matching.
</ParamField>

```python theme={null}
# 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.

```python theme={null}
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

```python theme={null}
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())
```