🧩 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