NSPSession: Reading State and Executing Actions in NSP - NSP

A Session (returned as NSPSession) is the workhorse object in any NSP agent. You get one by calling client.attach() with an app name, and from that point on you never pass PIDs or app IDs around — the session carries that context internally. It caches the app’s state locally, gives you typed accessors for individual keys, enforces action reversibility rules before requests leave Python, and exposes a WebSocket stream so you can react to changes in real time.

Creating a session

Call client.attach() with the app’s name. Fuzzy matching is on by default, so partial names work.

# Attach by display name
session = await client.attach("Gmail")

# Partial match
session = await client.attach("mission")   # matches "Mission Planner"

# Force a fresh probe on attach (slower, but guarantees up-to-date state)
session = await client.attach("Gmail", fresh=True)

Session properties

Property	Type	Description
`session.state`	`SubstrateState`	The cached state snapshot. Call `refresh()` to update it.
`session.pid`	`int`	OS process ID
`session.app_id`	`str`	Stable semantic app identifier
`session.app_name`	`str`	Human-readable name
`session.confidence`	`float`	Probe confidence of the cached state (0.0–1.0)
`session.actions`	`dict[str, ActionSchema]`	All available actions and their schemas
`session.clr`	`CLRSessionProxy`	.NET-specific helpers (CLR apps only)
`session.jvm`	`JVMSessionProxy`	Java-specific helpers (JVM apps only)

Reading state

Typed accessors

The typed helpers (get_str, get_int, get_bool) read from the locally cached state. They never raise on a missing key — they return the default value you provide (or None if you omit it).

# String value
subject = session.get_str("inbox.emails[0].subject")
sender  = session.get_str("inbox.emails[0].from")

# Integer value
unread = session.get_int("inbox.unread_count", 0)

# Boolean value
is_read = session.get_bool("inbox.emails[0].is_read")

# Float value
altitude = session.get_float("flight.altitude_m", 0.0)

Accessing the full state object

For bulk reads or when you need metadata about the snapshot itself, use session.state directly.

state = session.state

# Read a key from the objects map
from_addr = state.objects.get("inbox.emails[0].from")

# Get all keys that share a prefix
inbox_keys = state.keys_with_prefix("inbox.")
# Returns: {'inbox.unread_count': 47, 'inbox.total_count': 1284, ...}

# Inspect probe metadata
print(f"Confidence: {state.confidence}")
print(f"Captured at: {state.captured_at}")
print(f"Runtime: {state.runtime}")

Refreshing state

State is cached locally from the last probe cycle. Call refresh() to pull the latest cached snapshot from the daemon (~1ms), or pass fresh_probe=True to force a new probe cycle.

# Pull latest cached state from the daemon
state = await session.refresh()

# Force a completely fresh probe cycle (50ms–90s depending on runtime)
state = await session.refresh(fresh_probe=True)

Iterating available actions

session.actions is a dict[str, ActionSchema]. Iterate it to inspect what operations the app supports before deciding which one to call.

for name, schema in session.actions.items():
    flag = " ⚠️  IRREVERSIBLE" if schema.reversibility == "irreversible_write" else ""
    print(f"  {name}: {schema.signature}{flag}")

A typical output for Gmail looks like:

  send_reply: fn(thread_id: string, body: string, cc?: string[]) -> bool ⚠️  IRREVERSIBLE
  archive: fn(email_id: string) -> bool

Executing actions

`session.execute(action, parameters, verify, verify_expression, verify_timeout_ms)`

Call execute() to invoke any action the app exposes. The SDK checks the action’s reversibility tier and enforces safety rules before sending the request to the daemon. Reversible action — no verification required:

result = await session.execute(
    "archive",
    parameters={"email_id": "msg_18821af3"},
)
print(f"Success: {result.success}  latency={result.latency_ms}ms")

Irreversible action — verify=True and a verify_expression are required. The expression is evaluated against the post-action state; the call fails if it does not evaluate to true within verify_timeout_ms.

result = await session.execute(
    "send_reply",
    parameters={
        "thread_id": "thread_7af2b3",
        "body": "Thanks, I'll review this shortly.",
    },
    verify=True,
    verify_expression="inbox.sent_count > 0",
    verify_timeout_ms=5000,
)
print(f"Reply sent: {result.success}  verified: {result.verification.passed}")

Parameter	Type	Default	Description
`action`	`str`	required	Action name from `session.actions`
`parameters`	`dict`	`{}`	Parameters matching the action schema
`verify`	`bool \| None`	`None`	Enable post-action verification. Automatically `True` for `irreversible_write` actions.
`verify_expression`	`str \| None`	`None`	JavaScript expression evaluated against post-action state
`verify_timeout_ms`	`int`	`5000`	How long to poll for state settlement (100–30000ms)

The ActionResponse object returned by execute() has these key fields:

Field	Type	Description
`success`	`bool`	`True` if the action ran and verification passed
`latency_ms`	`int`	Total wall time from request to response
`verification`	`VerificationOutcome \| None`	Verification result, including `passed`, `changed_fields`, and `explanation`
`state_delta`	`dict \| None`	Before/after diff of changed state keys
`raw_return`	`Any \| None`	Raw return value for CLR/JVM method invocations

Calling execute() on an irreversible_write action without verify=True raises NSPIrreversibleNotConfirmedError immediately, before any network request is made. This is a client-side safety check — it cannot be silenced.

Real-time streaming

Call session.watch() to open a WebSocket stream and receive a StateChangeNotification on every probe cycle that detects a change. See the Streaming page for full details.

async with session.watch() as stream:
    async for event in stream:
        print(f"Changed keys: {event.changed_keys}")
        print(f"Sequence:     {event.sequence}")
        await session.refresh()   # Pull the updated state

Complete example

email_responder.py

import asyncio
import nelieo_nsp as nsp

async def email_responder():
    async with nsp.NSPClient() as client:
        session = await client.attach("Gmail")
        print(f"Connected to Gmail (conf={session.confidence:.2f})")

        unread = session.get_int("inbox.unread_count")
        print(f"Processing {unread} unread email(s)...")

        for i in range(min(unread, 10)):  # Cap at 10 for safety
            await session.refresh()

            thread_id = session.get_str(f"inbox.emails[{i}].thread_id")
            subject   = session.get_str(f"inbox.emails[{i}].subject")
            sender    = session.get_str(f"inbox.emails[{i}].from")
            is_read   = session.get_bool(f"inbox.emails[{i}].is_read")

            if not is_read:
                print(f"  Replying to: '{subject}' from {sender}")

                result = await session.execute(
                    "send_reply",
                    parameters={
                        "thread_id": thread_id,
                        "body": "Thank you for your email. I'll get back to you shortly.",
                    },
                    verify=True,
                    verify_expression="inbox.sent_count > 0",
                )

                status = "✅" if result.success else "❌"
                print(f"  {status} Sent in {result.latency_ms}ms")

asyncio.run(email_responder())

​Creating a session

​Session properties

​Reading state

​Typed accessors

​Accessing the full state object

​Refreshing state

​Iterating available actions

​Executing actions

​session.execute(action, parameters, verify, verify_expression, verify_timeout_ms)

​Real-time streaming

​Complete example

Creating a session

Session properties

Reading state

Typed accessors

Accessing the full state object

Refreshing state

Iterating available actions

Executing actions

`session.execute(action, parameters, verify, verify_expression, verify_timeout_ms)`

Real-time streaming

Complete example