> ## 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.
# NSPSession: Read Application State and Run Actions
> Full reference for NSPSession — typed state accessors, action execution, cached vs. fresh state, watch streaming, and wait-for condition helpers.
`NSPSession` is the high-level interface for interacting with a single attached application. It bundles the process ID and app identifier together so you never have to pass them around manually. Every state read, action call, and streaming subscription goes through a session.
## Creating a Session
You create a session by calling one of the `attach` methods on `NSPClient`:
```python theme={null}
# Most common — attach by app name
session = await client.attach("Gmail")
# Attach by PID
session = await client.attach_by_pid(12847)
# Force a fresh probe before the session opens
session = await client.attach("Gmail", fresh=True)
```
## Session Properties
| Property | Type | Description |
| -------------------- | ------------------------- | -------------------------------------------------- |
| `session.state` | `SubstrateState` | Cached state snapshot. Call `refresh()` to update. |
| `session.pid` | `int` | OS process ID |
| `session.app_id` | `str` | Stable semantic app identifier |
| `session.app_name` | `str` | Human-readable application name |
| `session.confidence` | `float` | Probe confidence of the cached state (0.0–1.0) |
| `session.actions` | `dict[str, ActionSchema]` | Available actions keyed by action name |
| `session.clr` | `CLRSessionProxy` | .NET-specific helpers (CLR apps only) |
| `session.jvm` | `JVMSessionProxy` | Java-specific helpers (JVM apps only) |
## Reading State
### Typed Accessor Methods
Use the typed getters to read individual values from the cached state. These methods never raise — they return the provided default (or `None`) when a key is missing:
```python theme={null}
# Generic — returns Any
value = session.get("inbox.unread_count")
# Typed — returns the specified type, default on missing key
count = session.get_int("inbox.unread_count", default=0)
alt = session.get_float("flight.altitude_m", default=0.0)
is_read = session.get_bool("inbox.emails[0].is_read", default=False)
subject = session.get_str("inbox.emails[0].subject", default="")
```
### Prefix Query
Retrieve all state keys that share a common prefix as a flat dictionary:
```python theme={null}
# Returns {"inbox.unread_count": 47, "inbox.total_count": 1284, ...}
inbox_data = session.get_keys_with_prefix("inbox.")
for key, value in inbox_data.items():
print(f" {key}: {value}")
```
### Accessing the Raw State Object
For operations not covered by the typed helpers, work directly with the `SubstrateState` object:
```python theme={null}
state = session.state
# Direct field access via the objects mapping
sender = state.objects.get("inbox.emails[0].from")
# Check whether an action can safely run at current confidence
if state.can_execute("send_reply"):
print("Safe to send")
else:
print(f"Confidence {session.confidence:.2f} is too low")
# Look up action schema metadata
schema = state.action("archive")
if schema:
print(f"Min confidence : {schema.min_confidence_required}")
print(f"Reversibility : {schema.reversibility}")
print(f"Is destructive : {schema.is_destructive}")
```
## Cached State vs. Fresh State
`session.state` holds the **last snapshot received** from the daemon. It does not automatically update. You have two ways to get newer data:
Fetches the latest cached snapshot from the daemon (\~1 ms). Use this after an action or when you need a more recent read without triggering a full probe.
Forces the daemon to run a new probe cycle before returning (50 ms–90 s). Use this only when you need a guaranteed up-to-date reading.
```python theme={null}
# Fast — fetches daemon's latest cached state
state = await session.refresh()
# Slow but accurate — triggers a fresh probe first
state = await session.refresh(fresh_probe=True)
```
## Executing Actions
### `execute()` Method
```python theme={null}
result = await session.execute(
action,
parameters={},
verify=None,
verify_expression=None,
verify_timeout_ms=5000,
auto_refresh=True,
)
```
The action name, as it appears in `session.actions`.
Key-value pairs matching the action's parameter schema.
Enable post-action verification. Automatically set to `True` for `IrreversibleWrite` actions — you cannot set this to `False` for irreversible actions without raising `NSPIrreversibleNotConfirmedError`.
A JavaScript expression evaluated against the post-action state. If it returns falsy, the action is marked as failed.
How long (in milliseconds) to poll for state settlement after the action runs. Range: 100–30000.
When `True`, refreshes `session.state` automatically after the action completes.
### Usage Examples
```python theme={null}
# Read-type action — no verification needed
result = await session.execute(
"get_thread",
parameters={"thread_id": "msg_18821af3"},
)
print(result.result) # {"subject": "...", "body": "...", ...}
# Reversible write — verify is optional
result = await session.execute(
"archive",
parameters={"email_id": "msg_18821af3"},
)
print(f"Archived: {result.success} latency={result.latency_ms}ms")
# Irreversible write — verify=True is REQUIRED
result = await session.execute(
"send_reply",
parameters={
"thread_id": "msg_18821af3",
"body": "Thanks, I will review this shortly.",
},
verify=True,
verify_expression="inbox.sent_count > 0",
verify_timeout_ms=5000,
)
print(f"Sent: {result.success} verified: {result.verification.satisfied}")
```
### `ActionResponse` Fields
`True` if the action ran and (when enabled) verification passed.
Total round-trip time from dispatch to response, in milliseconds.
The raw return value of the action as reported by the runtime. `null` for actions that return nothing (UI interactions, void methods).
Present when `verify=True`. Contains `satisfied: bool`, `actual_value`, `expression`, `evaluated_at`, and `latency_ms`. See [`VerificationResult`](/models/verification).
Human-readable failure description when `success` is `False`.
## Watch Stream
Open a real-time WebSocket stream of state-change events:
```python theme={null}
# Basic stream — you manage refreshes manually
async with session.watch() as stream:
async for event in stream:
print(f"Changed: {event.changed_keys}")
print(f"Seq : {event.sequence}")
print(f"At : {event.captured_at}")
await session.refresh()
# Auto-refresh — session.state is updated before each iteration
async with session.watch(auto_refresh_on_event=True) as stream:
async for event in stream:
unread = session.get_int("inbox.unread_count")
print(f"Unread: {unread}")
# Tune queue size and reconnect behavior
async with session.watch(max_queue=64, auto_reconnect=True) as stream:
async for event in stream:
...
```
See the [Streaming](/sdk/python/streaming) page for complete documentation.
## Wait Patterns
### `wait_for(predicate, *, timeout, poll_interval, fresh_probe)`
Poll the state until a lambda condition is satisfied:
```python theme={null}
# Wait until the email is marked as read
state = await session.wait_for(
lambda s: s.get_bool("inbox.emails[0].is_read"),
timeout=30.0,
poll_interval=1.0,
)
print("Email is now read")
# Wait for flight mode to become GUIDED, using fresh probes
state = await session.wait_for(
lambda s: s.get_str("flight.mode") == "GUIDED",
timeout=60.0,
poll_interval=0.5,
fresh_probe=True,
)
```
### `wait_for_key(key, *, expected_value, timeout, poll_interval, fresh_probe)`
Wait for a specific key to reach an exact value:
```python theme={null}
state = await session.wait_for_key(
"flight.mode",
expected_value="GUIDED",
timeout=30.0,
)
print(f"Mode is now: {state.get_str('flight.mode')}")
```
### Waiting for Change After an Action
Capture a baseline value, execute the action, then wait for the value to change:
```python theme={null}
before = session.get_int("inbox.sent_count")
await session.execute(
"send_reply",
parameters={"thread_id": "msg_18821af3", "body": "On my way."},
verify=True,
)
await session.wait_for(
lambda s: s.get_int("inbox.sent_count") > before,
timeout=10.0,
)
print("Reply confirmed sent")
```
## Complete Agent Example
```python theme={null}
import asyncio
import nelieo_nsp as nsp
async def email_responder():
"""Read all unread emails and send a canned reply to each."""
async with nsp.NSPClient(api_key="sk_nsp_...") 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 emails...")
for i in range(min(unread, 10)): # process at most 10
await session.refresh()
if session.get_bool(f"inbox.emails[{i}].is_read"):
continue
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")
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 will get back to you shortly.",
},
verify=True,
verify_expression="inbox.sent_count > 0",
)
if result.success:
print(f" ✅ Sent in {result.latency_ms}ms")
else:
print(f" ❌ Failed: {result.error_message}")
asyncio.run(email_responder())
```