NSP Python SDK Error Handling: Exceptions and Retries - NSP

The NSP Python SDK uses a structured exception hierarchy rooted at NSPError. Every exception the SDK raises is a subclass of it, so you can always catch nsp.NSPError as a broad fallback, or catch specific subclasses to handle individual failure modes precisely. All exceptions carry a human-readable message, a machine-readable code, and the HTTP status_code that the daemon returned (or 0 for connection-level failures).

Exception hierarchy

NSPError                                # Base class for all NSP errors
├── NSPConnectionError                  # Daemon unreachable (no HTTP response)
├── NSPAuthError                        # Invalid or missing API key (401)
├── NSPNotFoundError                    # PID or app not tracked (404)
├── NSPRateLimitError                   # Too many requests (429)
├── NSPTimeoutError                     # Request or fresh probe timed out (504)
├── NSPConfidenceTooLowError            # State confidence below action threshold (503)
├── NSPProbeNotReadyError               # Cold probe still in progress (503)
├── NSPActionError                      # Action executed but failed at app level (422)
│   ├── NSPVerificationFailedError      # Action ran but verify_expression didn't pass (422)
│   └── NSPIrreversibleNotConfirmedError # Missing verify=True on irreversible action
└── NSPServerError                      # Unexpected daemon error (5xx)

Base error fields

Every NSPError subclass exposes:

e.message      # str — human-readable description
e.code         # str — machine-readable error code
e.status_code  # int — HTTP status code (0 for connection-level failures)

Exception reference

`NSPAuthError` (401)

The API key is missing, invalid, or has been revoked. Check e.code for one of missing_key, invalid_key, or disabled_key.

try:
    apps = await client.list_apps()
except nsp.NSPAuthError as e:
    print(f"Auth failed [{e.code}]: {e.message}")
    # Generate or rotate a key at platform.nelieo.com

`NSPNotFoundError` (404)

The requested PID or app name is not currently tracked by the daemon. This is not a fatal error — the app may simply not be running yet.

try:
    session = await client.attach("Gmail")
except nsp.NSPNotFoundError:
    print("Gmail not tracked — is Chrome running with --remote-debugging-port=9222?")
    apps = await client.list_apps()
    print(f"Currently tracked: {[a.app_name for a in apps]}")

`NSPRateLimitError` (429)

You have exceeded the daemon’s rate limit for this API key (default: 100 state reads/sec, 10 actions/sec). Respect the backoff period and retry.

import asyncio

async def with_rate_limit_retry(fn, max_retries: int = 5):
    for attempt in range(max_retries):
        try:
            return await fn()
        except nsp.NSPRateLimitError:
            if attempt == max_retries - 1:
                raise
            wait = 2 ** attempt   # 1, 2, 4, 8, 16s
            print(f"Rate limited. Retrying in {wait}s (attempt {attempt + 1}/{max_retries})...")
            await asyncio.sleep(wait)

`NSPProbeNotReadyError` (503)

The cold probe is still in progress. The daemon has detected the application but has not yet completed its first JVMTI or CLR bridge attachment. Retry with exponential backoff.

import asyncio

async def get_state_with_backoff(client, pid: int):
    backoff = 2.0
    for attempt in range(8):
        try:
            return await client.get_state(pid)
        except nsp.NSPProbeNotReadyError:
            print(f"Probe not ready — waiting {backoff:.0f}s (attempt {attempt + 1})")
            await asyncio.sleep(backoff)
            backoff = min(backoff * 2, 60.0)
    raise RuntimeError(f"Probe never became ready for PID {pid}")

`NSPVerificationFailedError` (422)

The action executed successfully at the runtime level, but the post-action verify_expression evaluated to false before the verify_timeout_ms elapsed. The side effect has already occurred.

try:
    result = await session.execute(
        "send_reply",
        parameters={"thread_id": thread_id, "body": "Acknowledged."},
        verify=True,
        verify_expression="inbox.sent_count > 0",
    )
except nsp.NSPVerificationFailedError as e:
    # The reply WAS sent — state just didn't match the expression in time.
    # Do NOT retry blindly — you may send a duplicate.
    print(f"Action ran but verify_expression did not pass: {e.message}")
    await session.refresh()
    actual_count = session.get_int("inbox.sent_count")
    print(f"Current sent count: {actual_count}")

NSPVerificationFailedError does not mean the action failed. It means the action ran and the post-action state did not match your verify_expression within the timeout. If you retry without checking current state first, you risk executing the action a second time. Always call await session.refresh() and inspect the actual state before deciding whether to retry.

`NSPIrreversibleNotConfirmedError`

You called execute() on an irreversible_write action without setting verify=True. This is a client-side check — no network request was made.

# ❌ This raises NSPIrreversibleNotConfirmedError immediately:
await session.execute("send_reply", parameters={...}, verify=False)

# ✅ Correct — verify=True with an expression:
await session.execute(
    "send_reply",
    parameters={"thread_id": thread_id, "body": "On it."},
    verify=True,
    verify_expression="inbox.sent_count > 0",
)

`NSPTimeoutError` (504)

A fresh probe timed out before returning a state snapshot. Fall back to the cached state and retry the fresh read later.

try:
    state = await client.get_state(pid, fresh=True)
except nsp.NSPTimeoutError:
    print("Fresh probe timed out — using cached state")
    state = await client.get_state(pid, fresh=False)

`NSPConfidenceTooLowError` (503)

The probe’s confidence score is below the minimum required to execute the requested action. This typically happens immediately after a cold probe, or after the app navigates to a new view and the probe must re-label the state.

try:
    await session.execute("send_reply", parameters={...}, verify=True,
                          verify_expression="inbox.sent_count > 0")
except nsp.NSPConfidenceTooLowError:
    print(f"Confidence {session.confidence:.2f} too low — warming up probe...")
    await asyncio.sleep(5)
    await session.refresh(fresh_probe=True)
    # Retry once

Robust agent loop example

The example below shows how to handle every meaningful exception in a single agent loop. Copy it as a starting template for production agents.

robust_agent.py

import asyncio
import nelieo_nsp as nsp

async def robust_agent():
    async with nsp.NSPClient() as client:

        # Wait for Gmail with probe-ready retry
        for attempt in range(5):
            try:
                session = await client.attach("Gmail")
                break
            except nsp.NSPNotFoundError:
                print(f"Gmail not found (attempt {attempt + 1}/5) — retrying in 5s...")
                await asyncio.sleep(5)
        else:
            raise RuntimeError("Gmail never appeared in tracked apps")

        await session.refresh(fresh_probe=True)

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

        thread_id = session.get_str("inbox.emails[0].thread_id")

        for attempt in range(3):
            try:
                result = await session.execute(
                    "send_reply",
                    parameters={
                        "thread_id": thread_id,
                        "body": "Thanks — I'll follow up shortly.",
                    },
                    verify=True,
                    verify_expression="inbox.sent_count > 0",
                    verify_timeout_ms=8000,
                )
                print(f"✅ Reply sent in {result.latency_ms}ms")
                break

            except nsp.NSPIrreversibleNotConfirmedError:
                # Programming error — do not retry
                print("❌ Must set verify=True for irreversible actions")
                raise

            except nsp.NSPVerificationFailedError:
                # Action ran — check state before retrying
                print("⚠️  verify_expression failed — checking state before retry")
                await session.refresh()
                current_count = session.get_int("inbox.sent_count")
                print(f"   Current sent count: {current_count}")
                # If count increased, the reply was sent — break
                break

            except nsp.NSPConfidenceTooLowError:
                print(f"⚠️  Confidence {session.confidence:.2f} — waiting for probe...")
                await asyncio.sleep(5)
                await session.refresh(fresh_probe=True)

            except nsp.NSPRateLimitError:
                wait = 2 ** attempt
                print(f"⚠️  Rate limited — backing off {wait}s")
                await asyncio.sleep(wait)

            except nsp.NSPProbeNotReadyError:
                wait = 2 ** attempt
                print(f"⚠️  Probe not ready — retrying in {wait}s")
                await asyncio.sleep(wait)

            except nsp.NSPAuthError:
                print("❌ Invalid API key — check NSP_API_KEY")
                raise

            except nsp.NSPConnectionError:
                print("❌ Daemon unreachable — is axon-daemon running?")
                raise

asyncio.run(robust_agent())

​Exception hierarchy

​Base error fields

​Exception reference

​NSPAuthError (401)

​NSPNotFoundError (404)

​NSPRateLimitError (429)

​NSPProbeNotReadyError (503)

​NSPVerificationFailedError (422)

​NSPIrreversibleNotConfirmedError

​NSPTimeoutError (504)

​NSPConfidenceTooLowError (503)

​Robust agent loop example

Exception hierarchy

Base error fields

Exception reference

`NSPAuthError` (401)

`NSPNotFoundError` (404)

`NSPRateLimitError` (429)

`NSPProbeNotReadyError` (503)

`NSPVerificationFailedError` (422)

`NSPIrreversibleNotConfirmedError`

`NSPTimeoutError` (504)

`NSPConfidenceTooLowError` (503)

Robust agent loop example