nxnodex
Search docsv0.1.0GitHub

API

API reference

The reference is organized around the objects developers import and call directly.

Class

Agent

The main entry point for a Nodex graph. Create one agent, register nodes with decorators, then call run().

Agent(name: str = "agent", debug: bool = False)
nameName shown in terminal output and execution traces.
debugPrints richer failure details when available.
from nodex import Agent

app = Agent(name="research-agent", debug=True)

trace = app.run({"query": "AI trends"})

Decorator

app.node()

Registers a Python function as a graph node. The function receives state and must return a non-empty dictionary.

@app.node(next="end", retry=0, on_fail="raise", timeout=None, human_review=False)
nextNext node name, or "end" to finish the graph.
retryNumber of retry attempts before failure handling kicks in.
on_failFailure behavior: "raise" (default), "skip", or a fallback node name.
timeoutMaximum seconds allowed for the node function. Raises NodexTimeoutError when exceeded.
human_reviewIf True, Nodex prompts for terminal approval before continuing to the next node.
@app.node(next="writer", retry=2, on_fail="fallback", timeout=10.0)
def research(state):
    return {"notes": call_model(state.get("query"))}

Decorator

app.middleware

Registers a middleware function that wraps every node call. Middleware receives the current NodexState and a next_node callable. Call next_node(state) to continue execution.

@app.middleware
def my_middleware(state, next_node): ...

See the Middleware guide for full examples.

Decorator

app.route()

Adds conditional routing to a node. Use it when the next step depends on state produced by earlier nodes.

@app.route(condition, if_true="publish", if_false="review")
conditionA callable that receives NodexState and returns bool.
if_trueNode name to route to when the condition is truthy.
if_falseNode name to route to when the condition is falsy.
@app.node()
@app.route(
    condition=lambda state: state.get("confidence", 0) > 0.8,
    if_true="publish",
    if_false="review",
)
def writer(state):
    return {"draft": "content", "confidence": 0.92}

State

NodexState

A small wrapper around user data. It keeps internal execution fields separate from the dictionary your nodes read and update.

get(key, default)Read a value from the current state. Returns default if the key is missing.
set(key, value)Set one state value.
update(dict)Merge multiple values without overwriting internal fields.
Protected keys Do not read or write _current_node, _trace, or _retry_count from node code. Nodex manages these internally.

Return value

ExecutionTrace

Returned by app.run(). Captures the full result of one graph execution.

agent_nameThe name passed to Agent(name=...).
resultsList of NodeResult objects, one per executed node.
total_durationWall-clock seconds for the full graph run.
total_costSummed cost in USD across all nodes.
total_tokensSummed token count across all nodes.
successTrue if all nodes completed without a fatal error.
trace = app.run({"query": "AI trends"})

if trace.success:
    print(f"Done in {trace.total_duration:.2f}s, {trace.total_tokens} tokens")
else:
    for r in trace.results:
        if r.status != "success":
            print(f"Failed: {r.node_name} — {r.error}")

Result

NodeResult

One result per executed node, available as trace.results.

node_nameThe name of the node function.
status"success", "failed", "skipped", "retrying", or "cancelled".
outputThe dictionary returned by the node function.
durationWall-clock seconds for this node.
errorException string if the node failed, otherwise None.
retriesNumber of retry attempts made.
tokensToken usage reported by the node (0 if not set).
costCost in USD reported by the node (0.0 if not set).

Testing

Testing utilities

Import from nodex.testing to test individual nodes in isolation without running the full graph.

from nodex.testing import assert_node_output, make_state, test_node

# Run a node and inspect the result
result = test_node(research, {"query": "AI trends"})
assert result.success
assert "notes" in result.output

# Assert specific keys are present in the output
assert_node_output(research, {"query": "AI trends"}, ["notes"])

# Build a NodexState manually for more control
state = make_state({"query": "test", "confidence": 0.9})
result = test_node(writer, state)
test_node(fn, input)Runs a single node function with the given input dict or NodexState. Returns a NodeTestResult.
assert_node_output(fn, input, keys)Asserts that all keys are present in the node output. Raises AssertionError on failure.
make_state(dict)Creates a NodexState from a plain dictionary, useful when you need direct state control in tests.

Errors

Exceptions

All Nodex exceptions inherit from NodexError.

NodexErrorBase class for all Nodex exceptions. Safe to catch as a catch-all.
NodexNodeErrorRaised when a node fails after exhausting retries and failure handling.
NodexStateErrorRaised when a node returns invalid state output (empty or non-dict).
NodexMiddlewareErrorRaised when middleware raises. Includes the middleware function name.
NodexRouteErrorRaised when a route condition raises or returns a non-boolean value.
NodexConfigErrorRaised for invalid agent or graph configuration, such as a missing node reference.
NodexAuthErrorRaised by auth middleware when a request is rejected. Extend this for custom auth logic.
NodexTimeoutErrorRaised when a node exceeds its timeout value.