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)
| name | Name shown in terminal output and execution traces. |
|---|---|
| debug | Prints 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)
| next | Next node name, or "end" to finish the graph. |
|---|---|
| retry | Number of retry attempts before failure handling kicks in. |
| on_fail | Failure behavior: "raise" (default), "skip", or a fallback node name. |
| timeout | Maximum seconds allowed for the node function. Raises NodexTimeoutError when exceeded. |
| human_review | If 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")
| condition | A callable that receives NodexState and returns bool. |
|---|---|
| if_true | Node name to route to when the condition is truthy. |
| if_false | Node 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. |
_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_name | The name passed to Agent(name=...). |
|---|---|
| results | List of NodeResult objects, one per executed node. |
| total_duration | Wall-clock seconds for the full graph run. |
| total_cost | Summed cost in USD across all nodes. |
| total_tokens | Summed token count across all nodes. |
| success | True 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_name | The name of the node function. |
|---|---|
| status | "success", "failed", "skipped", "retrying", or "cancelled". |
| output | The dictionary returned by the node function. |
| duration | Wall-clock seconds for this node. |
| error | Exception string if the node failed, otherwise None. |
| retries | Number of retry attempts made. |
| tokens | Token usage reported by the node (0 if not set). |
| cost | Cost 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.
| NodexError | Base class for all Nodex exceptions. Safe to catch as a catch-all. |
|---|---|
| NodexNodeError | Raised when a node fails after exhausting retries and failure handling. |
| NodexStateError | Raised when a node returns invalid state output (empty or non-dict). |
| NodexMiddlewareError | Raised when middleware raises. Includes the middleware function name. |
| NodexRouteError | Raised when a route condition raises or returns a non-boolean value. |
| NodexConfigError | Raised for invalid agent or graph configuration, such as a missing node reference. |
| NodexAuthError | Raised by auth middleware when a request is rejected. Extend this for custom auth logic. |
| NodexTimeoutError | Raised when a node exceeds its timeout value. |