What imports do I need?

Import `StateGraph`, `Agent`, `ToolNode` from `tenxgraph.core.graph`, `AgentState` and `Message` from `tenxgraph.core.state`, and `START`, `END` from `tenxgraph.utils`.

How do I route between nodes?

Use `add_edge()` for static routes and `add_conditional_edges()` with a function that returns a node name or looks up the name in a map.

Do I need a checkpointer?

No; graphs default to in-memory state. Add a checkpointer to `compile()` for persistence across runs.

Build a graph | 10xGraph

StateGraph is the core orchestration primitive in 10xGraph. You construct a workflow by adding nodes (functions, Agent instances, or ToolNode instances), connecting them with edges, and compiling to get a runnable CompiledGraph. This guide walks you through building, wiring, and running a graph end to end.

Prerequisites

Install the core library with a provider extra:

Terminal

pip install "10xgraph[google-genai]"

Set your provider API key:

Terminal

export GOOGLE_API_KEY=...         # for Google Gemini
export OPENAI_API_KEY=sk-...      # for OpenAI
export ANTHROPIC_API_KEY=...      # for Anthropic

Import the essentials

Python

from tenxgraph.core.graph import StateGraph, Agent, ToolNode
from tenxgraph.core.state import AgentState, Message
from tenxgraph.utils import START, END

Define your tools

Tools are plain Python functions. The LLM sees the function name, docstring, and type-annotated parameters. This schema determines what tools the LLM can call.

Python

def get_weather(city: str) -> str:
    """Return current weather for a city."""
    return f"Weather in {city}: 22°C, partly cloudy."

def calculate(expression: str) -> str:
    """Evaluate a math expression safely."""
    try:
        return str(eval(expression, {"__builtins__": {}}, {}))
    except Exception as e:
        return f"Error: {e}"

Create a ToolNode

Group tools in a ToolNode. Each function’s __name__ becomes the tool name the LLM uses. The node runs all tools in parallel when the LLM requests several at once.

Python

tool_node = ToolNode([get_weather, calculate])

Create an Agent node

Agent wraps the LLM as a graph node. It calls the model, receives tool calls if the LLM requests them, and returns a message. You pass tools by name or as a ToolNode instance.

Python

agent = Agent(
    model="gemini-2.5-flash",
    system_prompt=[{"role": "system", "content": "You are a helpful assistant."}],
    tool_node=tool_node,
)

Other key options:

provider: explicit provider choice ("google", "openai", "anthropic"). Usually inferred from the model name.
temperature, max_tokens, top_p: model parameters.
output_schema: Pydantic model for structured output.

Build and wire the graph

A graph is built by adding nodes and edges. Edges define the flow: static edges (always route one way) or conditional edges (route based on the state).

Python

graph = StateGraph()

# Add two nodes: the agent and the tool executor
graph.add_node("MAIN", agent)
graph.add_node("TOOL", tool_node)

# Conditional routing: if the agent produced tool calls, go to TOOL, else END
def should_use_tools(state: AgentState) -> str:
    last = state.context[-1] if state.context else None
    if last and last.role == "assistant" and getattr(last, "tools_calls", None):
        return "TOOL"
    return END

graph.add_conditional_edges("MAIN", should_use_tools, {"TOOL": "TOOL", END: END})

# After tools are done, loop back to the agent
graph.add_edge("TOOL", "MAIN")

# Set the entry point (START → MAIN)
graph.set_entry_point("MAIN")

Key graph methods

Method	Purpose
`add_node(name, func)`	Add a node. The name is a string; `func` is a callable, `Agent`, or `ToolNode`.
`add_edge(from_node, to_node)`	Add a static route: always go from `from_node` to `to_node`.
`add_conditional_edges(from_node, condition, path_map)`	Add dynamic routing. `condition(state)` returns a key; `path_map` maps keys to node names. Or omit `path_map` and have `condition` return the node name directly.
`set_entry_point(node_name)`	Set the starting node. Shorthand for `add_edge(START, node_name)`.
`compile(checkpointer=None, store=None, ...)`	Build the executable graph. Returns a `CompiledGraph`.

Compile

Once the graph is wired, compile it to get a runnable CompiledGraph:

Python

app = graph.compile()

Compilation validates the graph structure and prepares it for execution. Without a checkpointer, the graph uses an InMemoryCheckpointer: threads keep their state while the process runs, and everything is lost on restart. For state that survives restarts, pass a durable checkpointer (the SQLite one needs pip install "10xgraph[sqlite_checkpoint]"):

Python

from tenxgraph.storage.checkpointer import SqliteCheckpointer

app = graph.compile(
    checkpointer=SqliteCheckpointer(db_path="./db.sqlite"),
)

See set up checkpointing for persistent state options.

Run the graph

Call invoke() to run the graph synchronously:

Python

result = app.invoke(
    {"messages": [Message.text_message("What is the weather in Paris?")]},
    config={"thread_id": "session-1", "user_id": "user-42"},
)

for msg in result["messages"]:
    print(f"{msg.role}: {msg.text()}")

Or use ainvoke() for async execution:

Python

import asyncio

async def main():
    result = await app.ainvoke(
        {"messages": [Message.text_message("Calculate 123 * 456")]},
        config={"thread_id": "session-2"},
    )
    for msg in result["messages"]:
        print(f"{msg.role}: {msg.text()}")

asyncio.run(main())

Run config: thread_id, user_id, and more

The config dict controls runtime behavior. Most keys are optional:

Key	Default	Purpose
`thread_id`	UUID (auto-generated, with a logged warning)	Conversation thread ID. Used by the checkpointer to save and load state. Pass one explicitly for any run you may resume or stop.
`user_id`	`"anonymous"`	User identifier. Passed to tools and event publishers.
`recursion_limit`	25	Max node execution steps. Prevents infinite loops; raises `GraphRecursionError` if exceeded.

Other reserved keys are set by the runtime and should not be passed:

run_id: unique execution ID.
is_stream: true if this is a streaming run.
timestamp: run start time.

If you enable JWT auth on the API server, two additional keys are injected:

user_id: authenticated user.
user: the auth object or claims.

Extra keys in config are kept and are readable by nodes that take config:

Python

result = app.invoke(
    {"messages": [Message.text_message("What is the weather in Paris?")]},
    config={"thread_id": "session-1", "user_id": "user-42", "metadata": {"source": "api"}},
)

To set fields on a custom state at invoke time, pass them under a "state" key in the input, for example {"messages": [...], "state": {"priority": "high"}}. See use custom state.

Handle tool errors

When a tool raises an exception, the error is caught and returned as a ToolResultBlock with the error message. The agent sees the error and can retry or report it.

Python

def divide(a: float, b: float) -> float:
    """Divide two numbers."""
    if b == 0:
        raise ValueError("Cannot divide by zero.")
    return a / b

When the agent calls divide(10, 0), the error is turned into a tool result and sent back to the agent in the next message. The agent can decide to inform the user or try a different approach.

Conditional routing patterns

Direct routing (condition returns node name)

The condition function returns the node name directly:

Python

class TicketState(AgentState):
    priority: str = "normal"
    category: str = "default"

def route_to_next(state: TicketState) -> str:
    priority = state.priority
    return "urgent_queue" if priority == "high" else "normal_queue"

graph.add_conditional_edges("classifier", route_to_next)

Mapped routing (condition result mapped to nodes)

The condition returns a key, and a map translates it to a node name:

Python

def get_category(state: TicketState) -> str:
    return state.category

category_map = {
    "finance": "finance_processor",
    "legal": "legal_processor",
    "default": "general_processor",
}
graph.add_conditional_edges("categorizer", get_category, category_map)

Tool call detection (common pattern)

Check if the last message from the agent contains tool calls:

Python

def should_use_tools(state: AgentState) -> str:
    last_msg = state.context[-1] if state.context else None
    if last_msg and last_msg.role == "assistant" and getattr(last_msg, "tools_calls", None):
        return "tools"
    return "end"

graph.add_conditional_edges("agent", should_use_tools, {"tools": "tool_node", "end": END})

Complete example

Here is a working agent that can check weather and do math:

Python

from tenxgraph.core.graph import StateGraph, Agent, ToolNode
from tenxgraph.core.state import AgentState, Message
from tenxgraph.utils import END

def get_weather(city: str) -> str:
    """Get the weather for a city."""
    return f"Weather in {city}: 22°C, sunny."

def calculate(expression: str) -> str:
    """Evaluate a math expression."""
    try:
        return str(eval(expression, {"__builtins__": {}}, {}))
    except Exception as e:
        return f"Error: {e}"

tool_node = ToolNode([get_weather, calculate])

agent = Agent(
    model="gemini-2.5-flash",
    system_prompt=[{"role": "system", "content": "You are a helpful assistant."}],
    tool_node=tool_node,
)

graph = StateGraph()
graph.add_node("MAIN", agent)
graph.add_node("TOOL", tool_node)

def should_use_tools(state: AgentState) -> str:
    last = state.context[-1] if state.context else None
    if last and last.role == "assistant" and getattr(last, "tools_calls", None):
        return "TOOL"
    return END

graph.add_conditional_edges("MAIN", should_use_tools, {"TOOL": "TOOL", END: END})
graph.add_edge("TOOL", "MAIN")
graph.set_entry_point("MAIN")

app = graph.compile()

result = app.invoke(
    {"messages": [Message.text_message("What is the weather in Tokyo and what is 100 * 50?")]},
    config={"thread_id": "demo-1", "user_id": "user-1"},
)

for msg in result["messages"]:
    print(f"{msg.role}: {msg.text()}")

The output includes the assistant’s tool call turn, the tool results (Weather in Tokyo: 22°C, sunny. and 5000) and a final assistant answer. Exact wording depends on the model.

Next steps

Configure an Agent: dive into model selection, system prompts, and structured output.
Set up checkpointing: add persistence for multi-turn conversations.
Stream responses: emit tokens and intermediate results as they happen.
Custom state: extend AgentState with domain-specific fields.
Use custom nodes: write function nodes that are not agents or tools.

Prerequisites

Import the essentials

Define your tools

Create a ToolNode

Create an Agent node

Build and wire the graph

Key graph methods

Compile

Run the graph

Run config: thread_id, user_id, and more

Handle tool errors

Conditional routing patterns

Direct routing (condition returns node name)

Mapped routing (condition result mapped to nodes)

Tool call detection (common pattern)

Complete example

Next steps

Frequently asked questions