10xGraph

Formerly Agentflow. 10xGraph by 10xScale: graph engineering for production AI agents.

10xGraph is an open-source Python framework for building multi-agent AI systems and running them in production. You write the agent as a graph of nodes and tools. 10xGraph keeps tool calls from running twice after a crash, guards state writes, enforces timeouts, and, with the CLI package, generates the API server around the graph.

This repository is the core engine (PyPI 10xgraph, import tenxgraph). The docs live at 10xgraph.com.

What it gives you

1. Correct under failure (this package)

• Replay-safe tools. The run loop persists the current node before running it, so a process killed mid-node re-runs that node on resume. Before calling a tool, 10xGraph checks the checkpointer's tool ledger and records each completed call as soon as it returns. A tool that already ran is not executed again: no double charge, no duplicate email. Requires a checkpointer.
• Versioned state writes. Durable writes use optimistic compare-and-swap, so two runs on one thread cannot overwrite each other. The Redis cache write is version-guarded too.
• Real node and tool timeouts. Set node_timeout and tool_timeout in the run config (defaults 900 s and 300 s), so a hung tool cannot hold a worker forever.
• Human approval inside a tool. interrupt() pauses the run and saves the thread; resume with the decision.

2. The production server ships in the box (CLI package, MIT)

The agentflow CLI turns a compiled graph into a service: REST, SSE streaming, WebSocket and realtime-audio endpoints; JWT or custom auth; scoped authorization on every endpoint; thread ownership isolation; rate limiting; and Docker Compose and Kubernetes files. See the ecosystem table.

Also included (these are standard for agent frameworks, listed as facts): graph orchestration and the ReAct tool-calling loop, OpenAI, Google Gemini and Anthropic support, MCP tools, streaming, and checkpointing to a database.

Install

pip install 10xgraph

Provider SDKs and infrastructure integrations are optional extras. Install only what you use:

pip install "10xgraph[google-genai,openai,anthropic,mcp,pg_checkpoint]"

pip install "10xgraph[google-genai,openai,anthropic,mcp,pg_checkpoint]"

Then set your provider key. A .env file in the working directory is loaded automatically.

export GEMINI_API_KEY=...        # Google Gemini
export OPENAI_API_KEY=sk-...     # OpenAI, or any OpenAI-compatible endpoint
export ANTHROPIC_API_KEY=sk-...  # Anthropic Claude

Requires Python 3.12 or newer.

Quick start: a support agent with an approval step

lookup_order reads data. refund_order moves money, so it pauses for a human decision with interrupt() before it acts. With a checkpointer, a crash or retry after the refund does not issue it twice.

from tenxgraph.core.state import Message
from tenxgraph.prebuilt.agent import ReactAgent
from tenxgraph.storage.checkpointer import InMemoryCheckpointer
from tenxgraph.utils import interrupt


def lookup_order(order_id: str) -> dict:
    """Look up an order by id."""
    return {"order_id": order_id, "status": "delivered", "total": 42.00}


def refund_order(order_id: str, amount: float) -> str:
    """Refund an order. Requires approval."""
    decision = interrupt({"amount": amount}, message=f"Refund ${amount}?")
    if not decision.get("approved"):
        return "Refund declined by reviewer."
    return f"Refunded ${amount} for order {order_id}."


app = ReactAgent(
    model="gemini/gemini-2.5-flash",
    system_prompt=[{"role": "system", "content": "You are a support agent for an online store."}],
    tools=[lookup_order, refund_order],
).compile(checkpointer=InMemoryCheckpointer())

config = {"thread_id": "1"}
result = app.invoke(
    {"messages": [Message.text_message("Order A-1001 arrived broken, please refund it.")]},
    config=config,
)

# The run pauses inside refund_order. After a reviewer approves:
result = await app.ainvoke({"resume": {"approved": True}}, config)

Swap ReactAgent for RAGAgent, SwarmAgent, SupervisorTeamAgent or PlanActReflectAgent and the shape stays the same. Use PgCheckpointer (Postgres plus Redis) in production, or SqliteCheckpointer for local work.

Stream it:

async for chunk in app.astream(
    {"messages": [Message.text_message("Where is order A-1001?")]},
    config={"thread_id": "2"},
):
    print(chunk.model_dump())

Add MCP tools by passing a fastmcp client; remote tools join your local ones:

from fastmcp import Client

mcp_client = Client({
    "mcpServers": {
        "orders": {"url": "http://127.0.0.1:8000/mcp", "transport": "streamable-http"},
    }
})

app = ReactAgent(
    model="gemini/gemini-2.5-flash",
    tools=[lookup_order],
    client=mcp_client,
).compile()

Building your own graph

Prebuilt agents are graphs. When you need custom control flow, build one directly:

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

graph = StateGraph()
graph.add_node("MAIN", Agent(
    model="gemini/gemini-2.5-flash",
    system_prompt=[{"role": "system", "content": "You are a support agent."}],
    tool_node="TOOL",
))
graph.add_node("TOOL", ToolNode([lookup_order, refund_order]))


def route(state: AgentState) -> str:
    if state.context and state.context[-1].tools_calls:
        return "TOOL"
    return END


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

app = graph.compile()

Nodes can be plain functions too, so you can call a provider SDK directly. See examples/react/.

Realtime Audio Agents

Live audio-to-audio sessions over Gemini Live. The provider owns the turn loop, so the session runs
through arealtime: you push into an input queue and consume normalized events.

from tenxgraph.prebuilt.agent import AudioAgent
from tenxgraph.core.realtime import LiveInputQueue, RealtimeConfig

app = AudioAgent(
    "gemini-live-2.5-flash-preview",
    realtime_config=RealtimeConfig(model="gemini-live-2.5-flash-preview", voice="Puck"),
    tools=[get_weather],
).compile()

queue = LiveInputQueue()
queue.send_audio(pcm16_bytes)   # non-blocking, safe to call from an audio callback

async for event in app.arealtime(queue, {"thread_id": "t1"}):
    ...                         # AudioDeltaEvent / transcripts / ToolCallEvent / ...

queue.close()

Barge-in, persisted transcripts (raw audio is never stored), automatic reconnect with session
resumption, and image/video frame input are handled for you. system_prompt, skills, and memory
work as they do on any other agent. Needs pip install "10xgraph[realtime]".

Parallel tool execution

When an LLM requests several tools at once, 10xGraph runs them concurrently. No configuration is needed; it applies to every agent and graph.

Prebuilt agents and patterns

• Agents: React, RAG, Guarded, Plan-Act-Reflect, Audio, Swarm, SupervisorTeam, StructuredOutput
• Orchestration: Router, MapReduce, Sequential, Branch-Join
• Other: dependency injection through InjectQ, skills (Agent Skills spec), 3-layer memory (working state, checkpointer, vector stores such as Qdrant and Mem0), publishers (Console, Redis, Kafka, RabbitMQ, OpenTelemetry), evaluation and testing helpers

Moving from Agentflow

10xGraph is the new name of Agentflow. The framework, license and maintainers are the same. 10xscale-agentflow 0.10.1 is its last release.

pip uninstall 10xscale-agentflow
pip install 10xgraph

Uninstall first: both distributions provide the agentflow module and must not be installed side by side.

# before
from agentflow.core.graph import StateGraph
# after
from tenxgraph import StateGraph

The import name is tenxgraph because a Python identifier cannot start with a digit. All canonical paths are the old ones with agentflow replaced by tenxgraph (for example tenxgraph.core.graph, tenxgraph.storage.checkpointer, tenxgraph.prebuilt.agent). import agentflow keeps working as a deprecated alias until 2.0, with one DeprecationWarning.

Other renamed identifiers (old values still work where noted):

Ecosystem

The CLI and TypeScript client get their 10xGraph names in a later release. Until then they keep their current names.

From install to a running service:

pip install 10xgraph 10xscale-agentflow-cli
agentflow init --path my-agent && cd my-agent
agentflow api                    # REST and WebSocket API on :8000
agentflow play                   # server plus playground
agentflow build --docker-compose --k8s

A production scaffold with JWT auth and Redis rate limiting:

agentflow init --path my-agent --yes --template production --auth jwt --rate-limit redis

Examples

Runnable scripts in examples/:

Run one:

export GEMINI_API_KEY=...   # or OPENAI_API_KEY
python examples/react/react_single_class.py

Some examples still use pre-rename import paths; the canonical paths are listed in CLAUDE.md.

Limitations

• Smaller community and fewer integrations than LangGraph or CrewAI.
• Pre-1.0 (current release line 0.10.x): pin versions and read the changelog.
• The rename from Agentflow resets brand recognition; "10xGraph" has no search history yet.
• LangGraph has stronger visual tooling (Studio, LangSmith observability).
• Requires Python 3.12 or newer. Code-first, not a no-code builder.
• Automatic per-tool permissions (user A may call refund, user B may not) are not built in. Tools can check the caller's verified scopes with tenxgraph.core.authz.has_scope.

Roadmap

• Done: Core graph engine with nodes and edges
• Done: State management and checkpointing
• Done: Tool integration (MCP, custom tools, parallel execution)
• Done: Parallel tool execution
• Done: Streaming and event publishing
• Done: Human-in-the-loop support
• Done: Prebuilt agent patterns
• Done: Agent-to-Agent (A2A) communication protocols
• Done: Observability and tracing (OpenTelemetry)
• Done: Realtime audio-to-audio agents (Gemini Live)
• Planned: Remote node execution for distributed processing
• Planned: More persistence backends (Redis, DynamoDB)
• Planned: Parallel/branching strategies
• Planned: Visual graph editor

Contributing

10xGraph is built in the open and contributions are welcome: bug reports with a clean reproduction, docs and examples, provider coverage, persistence backends, and typing (removing a module from the mypy ignore list is a welcome pull request).

git clone https://github.com/10xGraph/10xGraph.git
cd 10xGraph
uv sync --dev
uv run pytest
uv run ruff check .
uv run mypy tenxgraph/

Read CONTRIBUTING.md for the full workflow and the Code of Conduct. Questions and ideas go to Discussions.

Security

Found a vulnerability? Do not open a public issue. Follow the process in SECURITY.md.

License

10xGraph is MIT licensed and made by 10xScale. Copyright 10xScale. Contributions are accepted under the same license.

Links

• Documentation: 10xgraph.com
• PyPI: 10xgraph
• Issues and Discussions
• Changelog
• Examples