Nodes

Write graph steps as Python functions

Each node receives a NodexState object and must return a non-empty dictionary of state updates. The next parameter tells Nodex which node to run after this one.

from nodex import Agent

app = Agent(name="my-agent")

@app.node(next="writer", retry=2)
def research(state):
    query = state.get("query", "LangGraph")
    return {"notes": f"Research notes about {query}"}

@app.node(next="end")
def writer(state):
    return {"final": state.get("notes", "").upper()}

State

Pass data through the graph

Nodex wraps your initial data in a NodexState object. It exposes get, set, and update while keeping internal execution fields (_trace, _retry_count, _current_node) hidden from your node logic.

# Read a value (with optional default)
query = state.get("query", "fallback")

# Set a single value
state.set("answer", "42")

# Merge multiple values at once
state.update({
    "answer": "42",
    "confidence": 0.9,
})

# Protected internal keys
# — do NOT read or write these
# directly from node functions:
#
#   state._current_node
#   state._trace
#   state._retry_count
#
# Nodex manages them automatically.

The return value of each node is automatically merged into state before the next node runs. You do not need to call state.update() yourself — just return a dictionary.

Middleware

Wrap every node with a shared chain

Middleware is the right place for logging, auth checks, metrics, and shared policy. Each middleware receives the current state and a next_node callable — call it to continue execution.

@app.middleware
def logger(state, next_node):
    print(f"[start] {state._current_node}")
    result = next_node(state)
    print(f"[done]  {state._current_node}")
    return result

Multiple middleware functions run in registration order. See the Middleware guide for advanced patterns including auth, metrics, and chaining.

Retries

Keep retry behavior near the node

Use retry for transient failures and on_fail to name a fallback node. When retries are exhausted and no fallback is set, Nodex raises NodexNodeError.

@app.node(next="writer", retry=2, on_fail="fallback_research")
def research(state):
    return {"notes": call_model(state.get("query"))}

@app.node(next="writer")
def fallback_research(state):
    return {"notes": "Could not reach model — using cached notes."}

Routing

Branch the graph based on state

Use @app.route when the next step depends on the output of the current node. The condition is evaluated after the node returns.

@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 here", "confidence": 0.92}

@app.node(next="end")
def publish(state):
    return {"status": "published"}

@app.node(next="end")
def review(state):
    return {"status": "needs_review"}

Tracing

Read what ran after each execution

app.run() always prints a coloured terminal summary and returns an ExecutionTrace object you can inspect programmatically.

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

print(trace.success)        # True
print(trace.total_duration) # 0.48 (seconds)
print(trace.total_tokens)   # 237
print(trace.total_cost)     # 0.0005

# Inspect per-node results
for r in trace.results:
    print(r.node_name, r.status, r.duration)

The API reference lists all fields on ExecutionTrace and NodeResult.

Human review

Pause before continuing

Set human_review=True on a node to ask for terminal approval before the graph moves to the next node. The agent will print the current state and wait for y / n.

@app.node(next="publish", human_review=True)
def review(state):
    # Nodex will pause here and ask for approval in the terminal
    return {"approved_content": state.get("draft")}