Graph

The graph module provides the core building blocks for constructing and executing agent workflows. StateGraph builds workflows, CompiledGraph runs them, and supporting classes like Node, Edge, BaseAgent, and RemoteToolConfig configure execution. Every 10xGraph application starts with a graph.

Import paths

from tenxgraph.core.graph import (
    Agent, BaseAgent, CompiledGraph, Edge, Node, RemoteToolConfig, StateGraph, ToolNode
)
from tenxgraph.core.state import AgentState, Message, StreamChunk
from tenxgraph.utils import END, START, ResponseGranularity

Constants

The two sentinel names that mark where a graph starts and ends.

StateGraph[StateT]

The builder class. Construct a workflow by adding nodes and edges, then call compile() to get an executable CompiledGraph. For the concept, see State graph.

Constructor

The constructor takes an optional state, plus optional context manager, publisher, ID generator and DI container.

from tenxgraph.core.graph import StateGraph
from tenxgraph.core.state import AgentState

graph = StateGraph()                          # default AgentState
graph = StateGraph(MyCustomState())           # custom state instance
graph = StateGraph(MyCustomState)             # custom AgentState subclass (instantiated automatically)

Parameters:

Methods

add_node

Registers a node. Pass a function, or a name plus a function, ToolNode or Agent.

graph.add_node(func)                          # name inferred from function.__name__
graph.add_node("my_node", func)              # explicit name
graph.add_node("agent", agent_instance)      # Agent or ToolNode instance

Returns StateGraph for chaining. Raises ValueError if the arguments match none of the patterns below.

add_edge

Adds a static edge between two nodes.

graph.add_edge(START, "entry")   # also sets the entry point
graph.add_edge("entry", END)
graph.add_edge("node_a", "node_b")

The graph always follows this route. An edge from START also sets the entry point. Returns StateGraph.

add_conditional_edges

Routes from a node using a function evaluated at run time. The condition receives only the current state (not config) and must be side-effect free.

def route(state: AgentState) -> str:
    # Return a node name (or END) directly
    if state.context and state.context[-1].text() == "done":
        return END
    return "process"

graph.add_conditional_edges("check", route)

# With a path map: the return value is looked up in the map
graph.add_conditional_edges(
    "classify",
    lambda state: "urgent" if "urgent" in state.context[-1].text() else "normal",
    path_map={
        "urgent": "urgent_handler",
        "normal": "normal_handler",
    },
)

Parameters:

set_entry_point

Sets the first node to run.

graph.set_entry_point("my_node")
# equivalent to graph.add_edge(START, "my_node")

override_node

Replaces an existing node with a new function, Agent, or ToolNode. Raises KeyError if the node does not exist. Useful for swapping production nodes with test doubles before compilation.

graph.override_node("MAIN", test_agent)

compile

Validates the graph and returns an executable CompiledGraph.

app: CompiledGraph = graph.compile()

app = graph.compile(
    checkpointer=my_checkpointer,
    store=my_store,
    media_store=my_media_store,
    interrupt_before=["review_node"],
    interrupt_after=["tool_node"],
    shutdown_timeout=30.0,
)

Parameters:

Raises: GraphError with GRAPH_002 if no entry point is set, GRAPH_003 if a node is orphaned (no edge touches it), or GRAPH_004 if an edge target or an interrupt node does not exist.

CompiledGraph[StateT]

The executable graph produced by StateGraph.compile(). Do not instantiate directly.

invoke

Runs the graph synchronously and returns the result.

result = app.invoke(
    {"messages": [Message.text_message("Hello")]},
    config={"thread_id": "session-1", "user_id": "alice"},
    response_granularity=ResponseGranularity.LOW,
)

Blocks until the graph finishes. It uses asyncio.run() internally, so do not call it from an async context. Use ainvoke() instead.

ainvoke

The asynchronous form of invoke. It starts fresh or resumes an interrupted run.

result = await app.ainvoke(
    {"messages": [Message.text_message("Hello")]},
    config={"thread_id": "session-1"},
    response_granularity=ResponseGranularity.FULL,
)

Parameters (both invoke and ainvoke):

ResponseGranularity values:

Returns: dict with keys depending on granularity. A graph that contains a LiveAgent raises RuntimeError; drive it with arealtime(). For streaming and interrupts, see Stream a graph.

stream / astream

Run the graph and yield StreamChunk objects as it executes.

# Sync
for chunk in app.stream({"messages": [msg]}, config={"thread_id": "t1"}):
    if chunk.event == "message":
        print(chunk.message.content)

# Async
async for chunk in app.astream({"messages": [msg]}, config={"thread_id": "t1"}):
    if chunk.event == "message":
        print(chunk.message.content)

StreamChunk fields:

stop / astop

Request that a running thread stop.

# Async: call from within an async route handler
resp = await app.astop(config={"thread_id": "session-1"})
# {"ok": True, "running": True}

# Sync
resp = app.stop(config={"thread_id": "session-1"})

Sets a stop flag on the running execution. The graph checks this flag between node transitions and halts cleanly. config must carry the thread_id of the run. Without a checkpointer it returns {"ok": False, "reason": "no-checkpointer"}.

Returns: dict with ok, running, and optional reason keys.

override_node (on CompiledGraph)

compiled.override_node("MAIN", test_agent)

Same semantics as StateGraph.override_node() but on an already-compiled graph. Useful when testing a compiled production graph without rebuilding it.

Async context manager

Use async with on a compiled graph so resources are released on exit. CompiledGraph implements the async context manager protocol. Use async with to ensure aclose() runs automatically on exit, even when an exception is raised:

async def main():
    graph = graph_builder.compile()
    async with graph:
        result = await graph.ainvoke({"messages": [msg]})

# aclose() has been called here, even if ainvoke raised

When your graph is built by an async factory function, await the factory first, then use async with:

async def main():
    async with await build_and_compile_graph() as graph:
        result = await graph.ainvoke(input_data)

aclose

Releases the graph’s resources.

await app.aclose()

It stops background tasks, closes the checkpointer, closes the publisher, and releases the store. Always call this when shutting down a long-lived application.

aclose() is idempotent: calling it a second time is a no-op and returns {"status": "already_closed"} rather than raising. This means calling aclose() explicitly inside an async with block, then having __aexit__ call it again, is safe.

generate_graph

Returns a dictionary describing the graph structure.

graph_info = app.generate_graph()

Returns: dict with keys nodes (list of node objects with id and name), edges (list of edge objects with id, source, target), and info (metadata including node and edge counts, checkpointer and store presence, interrupt lists, context manager type, ID generator type, state type and state fields). Node and edge id values are random UUIDs generated on each call.

attach_remote_tools

Attaches client-executed tools to a ToolNode.

from tenxgraph.core.graph import RemoteToolConfig

tools = [
    RemoteToolConfig(
        node="tool_node",
        name="read_clipboard",
        description="Read the current clipboard text",
        parameters={
            "type": "object",
            "properties": {},
            "required": [],
        },
    )
]
app.attach_remote_tools(tools)

# Or pass OpenAI-compatible schemas with a node name
openai_schemas = [{
    "type": "function",
    "function": {
        "name": "read_clipboard",
        "description": "Read the current clipboard text",
        "parameters": {"type": "object", "properties": {}, "required": []},
    },
}]
app.attach_remote_tools(openai_schemas, node_name="tool_node")

Remote tools run on the client, not the server. The graph advertises their schemas to the model and emits a RemoteToolCallBlock when one is called.

Parameters:

Raises: GraphError (GRAPH_004) if the node does not exist, GraphError (GRAPH_005) if it is not a ToolNode, and ValueError for duplicate tool names or a tool whose node differs from node_name.

is_realtime

Reports whether the graph must be driven in realtime mode.

if app.is_realtime():
    # arealtime(input_queue, config=None, state=None) yields realtime events
    async for event in app.arealtime(input_queue):
        print(event)
else:
    result = app.invoke({"messages": [msg]})

Returns True if the graph contains at least one LiveAgent, meaning it must be driven via arealtime() or realtime() rather than invoke() or stream(). See Use realtime audio.

Returns: bool

Node

Represents an executable unit within a graph. It wraps a function, ToolNode, or Agent that processes state and returns output.

Constructor:

from tenxgraph.core.graph import Node

node = Node("processor", my_function)

Nodes are typically added to a graph via StateGraph.add_node() rather than instantiated directly.

Edge

Represents a connection between two nodes in the graph, defining routing logic and execution flow.

Constructor:

from tenxgraph.core.graph import Edge

# Static edge
edge = Edge("process", "approval")

# Conditional edge: the condition receives only the state
def needs_approval(state):
    return bool(state.context) and "approve" in state.context[-1].text()

conditional_edge = Edge("process", "approval", condition=needs_approval)

Edges are typically added to a graph via StateGraph.add_edge() or StateGraph.add_conditional_edges() rather than instantiated directly.

BaseAgent

Abstract base class for all agents (production and test). Subclasses must implement execute() and _call_llm(). Agent is the concrete production subclass; its own constructor accepts more options (see Agent).

Constructor:

from tenxgraph.core.graph import Agent

# Agent is a concrete subclass of BaseAgent
agent = Agent(
    model="gpt-4o",
    provider="openai",
    system_prompt=[{"role": "system", "content": "You are helpful."}],
)

BaseAgent parameters:

Methods:

RemoteToolConfig

Validated Pydantic model for a tool executed by the client rather than the server. Unknown fields are rejected, so typos fail at startup.

Constructor:

from tenxgraph.core.graph import RemoteToolConfig

config = RemoteToolConfig(
    node="tool_node",
    name="read_clipboard",
    description="Read the current clipboard text",
    parameters={
        "type": "object",
        "properties": {
            "full": {
                "type": "boolean",
                "description": "Return full history if true",
            }
        },
        "required": ["full"],
    },
)

Methods:

Config dictionary keys

The config dict passed to invoke, ainvoke, stream, astream and stop accepts these keys. Missing identifiers are generated for you.

Execution deadlines

Every node and tool call has a deadline, so a hung call cannot block a run forever.

Nothing else bounds a node or a tool. The provider SDK’s own timeout does not cover custom tools, MCP calls, or condition functions, so without a deadline a hung call blocks the run forever: the loop never advances a step, the recursion limit never trips, and the between-nodes stop check is never reached.

Both defaults are deliberately generous. They are a backstop against hangs, not a latency budget. The node default (900s) sits above the LLM client timeout (600s) so a slow LLM call fails with its own error rather than being masked by the node deadline.

result = app.invoke(
    input_data,
    config={
        "thread_id": "t-1",
        "node_timeout": 120.0,   # tighten the node deadline
        "tool_timeout": 30.0,    # tighten the tool deadline
    },
)

# Disable a deadline entirely: pass None, or any non-positive number
config = {"thread_id": "t-1", "node_timeout": None}

An absent key uses the framework default. An explicit None disables the deadline, which is different from the key being absent. A non-positive value (0) also disables it. A value that cannot be parsed as a float logs a warning and falls back to the default.

When a deadline expires the node task is cancelled and a NodeTimeoutError is raised, which the execution loop persists and reports as a normal node error.

Node function signature

Node functions receive the current state and config as their first two positional arguments. Additional keyword arguments are resolved from the dependency injection container (see Dependency injection):

from tenxgraph.core.state import AgentState, Message
from tenxgraph.storage.store import BaseStore

def my_node(state: AgentState, config: dict, store: BaseStore) -> list:
    # Return a list of Message objects to append to state.context
    return [Message.text_message("response", role="assistant")]

Alternatively, functions can return an updated state dict:

def my_node(state: AgentState, config: dict) -> dict:
    return {"context_summary": "Updated summary"}

Common errors