Embedding API

Embedding API

Nodus is designed to function as an embedded automation runtime inside larger Python applications. The NodusRuntime class is the stable embedding interface since v1.0.

Installation

pip install nodus-lang

Basic Usage

from nodus import NodusRuntime          # canonical import path (v1.0+)
# also works:
# from nodus.runtime.embedding import NodusRuntime

rt = NodusRuntime()
result = rt.run_source('print("hello from Nodus")')
print(result["stdout"])    # "hello from Nodus\n"

Important — an embedded runtime denies subprocess, network and environment access by default. This changed in v5.0.0 and is the one breaking change in that major. A bare NodusRuntime() cannot shell out, open a socket, or read the process environment; grant what you need explicitly:

rt = NodusRuntime(allow_subprocess=True, allow_network=True)

nodus run and the rest of the CLI are deliberately unaffected — the domain deny-by-default protects is work you did not fully author, and a developer running a script they just wrote is not that.

allowed_paths was already a CWD jail before this and still is: unset means the current working directory, not unrestricted.

The old 200 ms default timeout is gone. timeout_ms defaults to None, so no wall-clock deadline is applied unless you ask for one. Advice to pass timeout_ms=None predates that fix and is now a no-op.

NodusRuntime Constructor

The constructor takes 27 keyword arguments. The ones most embedders touch:

NodusRuntime(
    # limits
    max_steps=10_000_000,
    timeout_ms=None,
    max_stdout_chars=20_000,
    max_frames=None,
    max_memory_mb=None,
    coroutine_timeout_ms=None,

    # capabilities -- all three deny by default since v5.0.0
    allow_subprocess=False,
    allow_network=False,
    allow_env=False,
    allow_input=False,
    allowed_commands=None,
    allowed_hosts=None,

    # filesystem
    project_root=None,
    allowed_paths=...,      # unset = a jail on the current working directory
    writable_paths=None,

    # per-runtime state and services
    share_process_state=False,
    memory_store=None,
    agent_registry=None,
    agent_timeout_ms=None,
    workflow_runner=None,
    worker_dispatcher=None,
    persist_workflow_source=True,

    # policy and observability
    capability_policy=None,
    approval_channel=None,
    extensions=None,
    event_sinks=None,
    on_error=None,
)

Do not add **kwargs to this signature, and keep the confinement flags keyword-only. Both are load-bearing and pinned by test: with a catch-all, a renamed flag is silently swallowed and the guest runs unconfined while every mock-based test on the embedder's side stays green.

Parameter	Default	Description
max_steps	10_000_000	Max VM instructions per execution. Raises RuntimeLimitExceeded. None = unlimited.
timeout_ms	None	Wall-clock deadline in ms. None since the EMBED-001 fix — no deadline unless asked for.
max_stdout_chars	20_000	Max captured stdout. Output beyond it is silently truncated.
max_frames	None	Max call stack depth. None still applies the 10,000-frame backstop, so runaway recursion raises Call stack overflow rather than growing until the OS kills the process.
max_memory_mb	None	Memory ceiling for the run.
coroutine_timeout_ms	None	Per-coroutine deadline.
allow_subprocess	False	Shell out. Denied by default since v5.0.0.
allow_network	False	Open sockets. Denied by default since v5.0.0.
allow_env	False	Read the process environment. Denied by default since v5.0.0.
allow_input	False	Whether input() is permitted.
allowed_commands	None	Narrows allow_subprocess to a named set.
allowed_hosts	None	Narrows allow_network to a named set.
project_root	None	Absolute path the module loader resolves non-relative imports against.
allowed_paths	(unset)	Directories the script may read. Unset is a jail on the current working directory, not unrestricted.
writable_paths	None	Directories the script may write, if narrower than allowed_paths.
share_process_state	False	Share the process-global memory store. Off since v5.0.3: two runtimes in one process no longer share memory, because a guest writes it and a shared store was a channel between tenants.
memory_store	None	Inject a specific memory store.
agent_registry	None	Scope the agent registry per tenant. Agents are deliberately not isolated by default — a guest cannot register one, so a shared registry only holds what the host put there.
agent_timeout_ms	None	Bounds a host agent handler. Stops the wait, not the handler: arbitrary Python cannot be preempted, so the handler is abandoned on a daemon thread and recorded.
workflow_runner	None	Give the runtime its own runner instead of the process-global one.
worker_dispatcher	None	Custom worker dispatch.
persist_workflow_source	True	Persist a run's source so a resume can rebuild it.
capability_policy	None	Optional three-valued allow | ask | deny policy consulted at the host boundary. ask with no approval_channel denies.
approval_channel	None	Where an ask decision goes.
extensions	None	Extension names to load.
event_sinks	None	Runtime event sinks. A plain callable works.
on_error	None	Error hook.

The Floor (DEFAULT_FLOOR) sits under all of this and cannot be overridden — a policy that allows everything still cannot let a program write into .nodus/, the runtime's own state. Reads are untouched.

Methods

run_source(code, filename=None)

Execute a Nodus source string. Returns a result dict.

result = rt.run_source("""
let x = 42
print(x * 2)
""")

print(result["ok"])      # True or False
print(result["stdout"])  # "84\n"
print(result["stderr"])  # "" or error message

On error, result["error"] contains a structured error record.

run_file(path)

Execute a Nodus source file by path.

result = rt.run_file("/path/to/script.nd")

register_function(name, fn, arity=None)

Register a Python callable as a builtin callable from Nodus scripts. Registered functions are available in every subsequent run_source / run_file call.

def send_email(addr, body):
    print(f"[host] sending to {addr}")
    return True

rt.register_function("send_email", send_email, arity=2)

Arity is inferred from the function signature when not provided explicitly.

reset()

Reset the runtime state (clears globals, memory, etc.) while retaining registered functions and configuration.

Host Function Registration Example

from nodus import NodusRuntime

rt = NodusRuntime(
    max_steps=100_000,
    timeout_ms=5000,
)

# Register host functions
def db_query(sql):
    # connect to your database
    return [{"id": 1, "name": "Alice"}]

def log_event(event_name, payload):
    print(f"[event] {event_name}: {payload}")
    return True

rt.register_function("db_query", db_query, arity=1)
rt.register_function("log_event", log_event, arity=2)

# Run a script that calls the host functions
result = rt.run_source("""
let rows = db_query("SELECT * FROM users")
log_event("query_complete", len(rows))
print("Got " + str(len(rows)) + " rows")
""")

print(result["stdout"])

Sandbox Configuration

For untrusted code, configure all relevant limits:

rt = NodusRuntime(
    max_steps=50_000,           # cap instruction count
    timeout_ms=2000,            # 2-second wall clock limit
    max_stdout_chars=10_000,    # cap output size
    allowed_paths=["/data/safe"],  # restrict filesystem access
    allow_input=False,          # disallow input()
    max_frames=100,             # cap call stack depth
)

When a limit is exceeded, the VM raises RuntimeLimitExceeded (surfaced as err.kind = "sandbox" in the result).

Filesystem Sandboxing

allowed_paths restricts these builtins: read_file, write_file, append_file, mkdir, list_dir, exists. Paths outside the allowlist raise a sandbox error.

Result Dict Structure

{
    "ok": True,          # bool — did the script complete without error?
    "stdout": "...",     # captured stdout string
    "stderr": "...",     # empty string on success
    "error": {           # present only when ok=False
        "message": "...",
        "kind": "name",   # or "type", "sandbox", etc.
        "line": 5,
        "column": 3,
        "path": "script.nd"
    }
}

Complete Working Example

from nodus import NodusRuntime

# Configure the runtime
rt = NodusRuntime(
    project_root="/my/project",
    max_steps=200_000,
    timeout_ms=10_000,
    allowed_paths=["/my/project/data"],
)

# Register host capabilities
def fetch_config(key):
    configs = {"env": "production", "version": "1.0"}
    return configs.get(key)

rt.register_function("fetch_config", fetch_config, arity=1)

# Execute a workflow script
result = rt.run_file("/my/project/workflows/deploy.nd")

if result["ok"]:
    print("Workflow completed")
    print(result["stdout"])
else:
    print("Workflow failed:", result["error"]["message"])

Workflow and Task Graph Integration

The embedding API supports running scripts that contain workflow and goal DSL constructs. The NodusRuntime handles the full pipeline internally — tokenize → parse → compile → optimize → execute.

result = rt.run_source("""
workflow build {
    step compile {
        return fetch_config("version")
    }
    step test after compile {
        return compile + "-tested"
    }
}

let r = run_workflow(build)
print(r["steps"]["test"])
""")

Event Integration

Runtime events (tool calls, agent calls, workflow steps, memory operations) are emitted during execution. In the current API, events are accessible from inside scripts via std:runtime.events(). Structured event sink subscriptions from the host are planned for a future release.

API Stability

The NodusRuntime embedding API is stable as of v1.0:

• Constructor parameters are stable
• run_source(), run_file(), register_function(), reset() are stable
• from nodus import NodusRuntime is the canonical import path
• set_trace_id(id) — set trace ID propagated into all runtime events (v4.0)
• tool_registry property — Python-side access to the tool registry (v4.0)

Low-Level API

nodus.tooling.loader.run_source() is available but does not provide sandbox controls or host function registration. Use NodusRuntime for all embedding scenarios.

See Also

• Getting Started — CLI usage
• Workflows and Task Graphs — workflow DSL reference
• Standard Library — available stdlib modules