Middleware

Wrap every node with one reusable chain

Middleware is the right place for logging, auth checks, metrics, tracing context, and shared policy. Register it once — it runs around every node execution automatically.

@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

Each middleware receives the current NodexState and a next_node callable. You must call next_node(state) to continue execution and return its result.

Order

Middleware runs in registration order

The first registered middleware wraps the second, and the final callable is the node itself. This is the same onion-model used by Express, Koa, and ASGI middleware stacks.

@app.middleware
def first(state, next_node):
    print("first: before")
    result = next_node(state)   # calls second, which calls the node
    print("first: after")
    return result

@app.middleware
def second(state, next_node):
    print("second: before")
    result = next_node(state)   # calls the actual node
    print("second: after")
    return result

# Output order:
# first: before
# second: before
# [node runs]
# second: after
# first: after

Errors

Middleware errors are attributed

If middleware raises, Nodex wraps it in NodexMiddlewareError with the middleware function name included in the message so you can identify which layer failed.

from nodex import NodexMiddlewareError

try:
    trace = app.run({"query": "test"})
except NodexMiddlewareError as exc:
    print(exc)  # "Middleware 'auth_check' failed: Missing api_key"

Example

Auth check middleware

Reject execution early if a required credential is missing.

from nodex import NodexAuthError

@app.middleware
def auth_check(state, next_node):
    if not state.get("api_key"):
        raise NodexAuthError("Missing api_key in state")
    return next_node(state)

Example

Metrics and timing middleware

Measure per-node latency or record custom counters without touching individual node functions.

import time

@app.middleware
def metrics(state, next_node):
    node = state._current_node
    start = time.perf_counter()
    result = next_node(state)
    elapsed = time.perf_counter() - start
    # Send to your monitoring system:
    record_metric(f"nodex.node.{node}.duration_ms", elapsed * 1000)
    return result

Example

Retry-aware middleware

Access state._retry_count to change behavior on retried executions — for example, switching to a cheaper model on a second attempt.

@app.middleware
def retry_logger(state, next_node):
    if state._retry_count > 0:
        print(f"Retry #{state._retry_count} for {state._current_node}")
    return next_node(state)

Read-only internal fields You may read state._current_node and state._retry_count in middleware for observability, but do not modify them.