10xScale Agentflow

10xScale Agentflow is a lightweight Python framework for building intelligent agents and orchestrating multi-agent workflows. It's an LLM-agnostic orchestration tool that works with native SDKs from OpenAI, Google Gemini, Anthropic Claude, or any other provider. You choose your LLM library; 10xScale Agentflow provides the workflow orchestration.

This package is the core engine. It ships as part of a complete, end-to-end framework: an API server and CLI, a typed TypeScript/React client, a visual playground, and a full documentation site. Start here, then pick up the rest of the stack as you need it.

🧩 The Agentflow Ecosystem

Agentflow is not just a Python library. It is a full stack for taking a multi-agent system from a prototype to production.

The 60-second path from install to a running service:

pip install 10xscale-agentflow 10xscale-agentflow-cli
agentflow init my-agent && cd my-agent   # scaffold a project
agentflow api                            # REST + WebSocket API on :8000
agentflow play                           # server + visual playground

✨ Key Features

• ⚡ Agent Class - Build complete agents in 10-30 lines of code (new in v0.5.3!)
• 🎯 LLM-Agnostic Orchestration - Works with any LLM provider (OpenAI, Gemini, Claude, native SDKs)
• 🤖 Multi-Agent Workflows - Build complex agent systems with your choice of orchestration patterns
• 📊 Structured Responses - Get content, optional thinking, and usage in a standardized format
• 🌊 Streaming Support - Real-time incremental responses with delta updates
• 🎙️ Realtime Audio Agents - Live audio-to-audio sessions over Gemini Live with barge-in, transcripts, tool calling, and automatic reconnect (AudioAgent)
• 🔧 Tool Integration - Native support for function calling and MCP tools with parallel execution
• 🔀 LangGraph-Inspired Engine - Flexible graph orchestration with nodes, conditional edges, and control flow
• 💾 State Management - Built-in persistence with in-memory and PostgreSQL+Redis checkpointers
• 🔄 Human-in-the-Loop - Pause/resume execution for approval workflows and debugging
• 🚀 Production-Ready - Event publishing (Console, Redis, Kafka, RabbitMQ), metrics, and observability
• 🧩 Dependency Injection - Clean parameter injection for tools and nodes
• 📦 Prebuilt Patterns - React, RAG, Swarm, Router, MapReduce, SupervisorTeam, and more

🌟 What Makes Agentflow Different

Architecture and scale

• Cached checkpointing — PostgreSQL for durability, Redis as the hot layer, so state persistence stays fast under load.
• 3-layer memory — short-term working context, session conversation history, and long-term vector memory (Qdrant / Mem0).
• Custom ID generation — string, int, or bigint IDs instead of 128-bit UUIDs, which meaningfully shrinks database indexes.

Tools and context

• Parallel tool execution by default, with local Python tools, remote tools over the TypeScript SDK, agent handoff tools, and MCP servers all in the same registry.
• Dedicated context manager — trims context at iteration boundaries, never mid-execution, and is fully replaceable.
• First-class dependency injection via InjectQ, so tools and nodes stay testable.

Control and safety

• Human-in-the-loop interrupts — pause anywhere, resume with full state intact.
• Callback system at every execution stage for logging, validation, and prompt-injection defence.
• Flexible navigation — conditional routing, Command jumps, and handoff tools between agents.

Operations

• Event publishing to Kafka, RabbitMQ, Redis Pub/Sub, OpenTelemetry, or your own publisher.
• Background task manager for prefetching, memory persistence, and cleanup outside the request path.
• Pydantic-first — state, messages, and tool calls serialize, validate, and store without glue code.

📦 Installation

pip install 10xscale-agentflow          # or: uv pip install 10xscale-agentflow

Provider SDKs and infrastructure integrations are optional extras — install only what you use:

pip install "10xscale-agentflow[google-genai,openai,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

🚀 Quick Start

Prebuilt agents are the fastest way in. A complete tool-calling agent:

from agentflow.core.state import Message
from agentflow.prebuilt.agent import ReactAgent


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


app = ReactAgent(
    model="gemini/gemini-2.5-flash",
    system_prompt=[{"role": "system", "content": "You are a helpful assistant."}],
    tools=[get_weather],
).compile()

result = app.invoke(
    {"messages": [Message.text_message("What's the weather in Tokyo?")]},
    config={"thread_id": "1"},
)

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

That is the whole program. Message conversion, the tool loop, and parallel tool execution are handled for you. Swap ReactAgent for RAGAgent, SwarmAgent, SupervisorTeamAgent, or PlanActReflectAgent and the shape stays the same.

Stream it — same agent, incremental output:

async for chunk in app.astream(
    {"messages": [Message.text_message("What's the weather in Tokyo?")]},
    config={"thread_id": "1"},
):
    print(chunk.model_dump())

Add MCP tools — pass a fastmcp client; remote tools join your local ones:

from fastmcp import Client

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

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

Persist conversations — pass a checkpointer at compile time and reuse the thread_id:

from agentflow.storage.checkpointer import InMemoryCheckpointer  # PgCheckpointer in production

app = ReactAgent(model="gemini/gemini-2.5-flash", tools=[get_weather]).compile(
    checkpointer=InMemoryCheckpointer(),
)

🏗️ Building Your Own Graph

Prebuilt agents are graphs. When you need custom control flow, build one directly with the same primitives — nodes, conditional edges, and an entry point:

from agentflow.core.graph import Agent, StateGraph, ToolNode
from agentflow.core.state import AgentState, Message
from agentflow.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 helpful assistant."}],
    tool_node="TOOL",
))
graph.add_node("TOOL", ToolNode([get_weather]))


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 and keep full control over the request. See examples/react/ for that variant.

🎙️ 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 agentflow.prebuilt.agent import AudioAgent
from agentflow.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 10xscale-agentflow[realtime].

⚡ Parallel Tool Execution

When an LLM requests several tools at once, Agentflow runs them concurrently. No configuration, no code changes — it applies to every agent and graph.

get_weather("NYC") 1.0s + get_news("tech") 1.5s + get_stock("AAPL") 0.8s

Sequential: 1.0 + 1.5 + 0.8 = 3.3s
Agentflow:  max(1.0, 1.5, 0.8) = 1.5s

See the parallel tool execution docs.

📚 More Examples

Every pattern below is a runnable script in examples/:

Run one:

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

🎯 Use Cases & Patterns

10xScale Agentflow includes prebuilt agent patterns for common scenarios:

🤖 Agent Types

• React Agent - Reasoning and acting with tool calls
• RAG Agent - Retrieval-augmented generation
• Guarded Agent - Input/output validation and safety
• Plan-Act-Reflect - Multi-step reasoning
• Audio Agent - Realtime audio-to-audio sessions (Gemini Live) with barge-in and tool calling

🔀 Orchestration Patterns

• Router Agent - Route queries to specialized agents
• Swarm - Dynamic multi-agent collaboration
• SupervisorTeam - Hierarchical agent coordination
• MapReduce - Parallel processing and aggregation
• Sequential - Linear workflow chains
• Branch-Join - Parallel branches with synchronization

🔬 Advanced Patterns

• Deep Research - Multi-level research and synthesis
• Network - Complex agent networks

See the documentation for complete examples.

🔧 Development

For Library Users

Install 10xScale Agentflow as shown above. The pyproject.toml contains all runtime dependencies.

For Contributors

# Clone the monorepo and enter the core framework
git clone https://github.com/10xHub/agentflow.git
cd agentflow/agentflow

# Create the environment and install dev dependencies
uv sync --dev
uv run pre-commit install          # optional, mirrors CI

# Run tests (coverage gate: >= 80%)
uv run pytest --cov --cov-branch
uv run pytest tests/graph          # one area

# Lint, format, type-check
uv run ruff check . && uv run ruff format .
uv run mypy agentflow/

# Run an example
uv run python examples/react/react_sync.py

Development Tools

The project uses:

• pytest for testing (async support, coverage gate at 80%)
• ruff for linting and formatting (line length 100, target py312)
• mypy for type checking, applied in phases across the codebase
• bandit for security checks
• pre-commit to run all of the above before each commit
• Docusaurus for the documentation site (agentflow-docs/)

Tool configuration lives in pyproject.toml.

🗺️ Roadmap

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

🤝 Contributing

Agentflow is built in the open, and we are actively looking for contributors.

The framework is production-stable at its core, but it is wide: a graph engine, an API server, a TypeScript SDK, a React playground, and a documentation site. Every one of those surfaces has room for people who want to harden it, extend it, or make it easier to learn. If you have ever wanted to work on agent infrastructure rather than just use it, this is a good place to start.

Where you can contribute

Contributions are not limited to this package. The whole framework lives in one monorepo, so pick whichever part matches your skills:

Where we most need help right now

• Stabilization — edge cases in long-running graphs, streaming, and reconnect behaviour
• Persistence backends — Redis, DynamoDB, and other checkpointers beyond Postgres
• Provider coverage — more LLM adapters and better parity across providers
• Typing — the codebase is on phased mypy adoption; removing a module from the ignore list is a genuinely welcome PR
• Docs and examples — the fastest way to help other people adopt the framework
• Bug reports — even an issue with a clean reproduction moves the project forward

Getting started

git clone https://github.com/10xHub/agentflow.git
cd agentflow/agentflow
uv sync --dev                      # environment + dev tools
uv run pre-commit install          # optional, matches CI
uv run pytest --cov --cov-branch   # tests + coverage gate

Read CONTRIBUTING.md for the full workflow, and the Code of Conduct before you participate. Issues labelled good first issue are a deliberate on-ramp — if none are open, say hello in Discussions and we will scope one with you.

🌟 Contributors

Every feature in this framework exists because someone decided to build it. Thank you to everyone who has contributed code, docs, examples, issues, and reviews.

Your avatar belongs here. Open a pull request on any package in the monorepo and this wall updates itself.

🔐 Security

Found a vulnerability? Please do not open a public issue. Follow the disclosure process in SECURITY.md.

📄 License

MIT License. See LICENSE.

🔗 Links & Resources

Documentation

• Documentation home — tutorials, how-to guides, concepts, API reference
• Examples directory — runnable code for every major pattern

Packages

• PyPI: 10xscale-agentflow (core) · 10xscale-agentflow-cli (API + CLI)
• npm: @10xscale/agentflow-client (TypeScript SDK)

Project

• GitHub repository — source for all packages
• Issues — bug reports and feature requests
• Discussions — questions, ideas, and help
• Changelog — release history

Ready to build? Start with the documentation, or jump straight into the examples. If Agentflow is useful to you, a ⭐ on the repository helps other developers find it.