Meta

• License: MIT License (MIT License)
• Author: 10xScale
• Maintainer: Shudipto Trafder
• Tags agent , agents , ai-agent , ai-agents , multi-agent , multi-agent-systems , agentic , agentic-ai , agentic-workflow , agent-framework , agent-orchestration , orchestration , workflow , workflow-engine , state-machine , stategraph , graph , llm , llm-agent , llm-framework , llm-orchestration , llmops , genai , generative-ai , ai , artificial-intelligence , openai , anthropic , claude , gemini , google-genai , mcp , model-context-protocol , tool-use , function-calling , rag , memory , vector-store , qdrant , mem0 , langgraph , langchain , crewai , autogen , pydantic-ai , fastapi , react-agent , reasoning , chatbot , assistant , a2a
• Requires: Python >=3.12
• Provides-Extra: google-genai , realtime , openai , pg-checkpoint , mcp , images , cloud-storage , redis , kafka , rabbitmq , qdrant , mem0 , a2a-sdk , otel , all-publishers

Classifiers

• Development Status
  • 5 - Production/Stable
• Intended Audience
  • Developers
• License
  • OSI Approved :: MIT License
• Operating System
  • OS Independent
• Programming Language
  • Python :: 3
  • Python :: 3.12
  • Python :: 3.13
• Topic
  • Software Development :: Libraries :: Application Frameworks
  • Software Development :: Libraries :: Python Modules

10xScale Agentflow

👁 CI
👁 Release
👁 CodeQL
👁 codecov

👁 PyPI
👁 Python
👁 License
👁 Coverage
👁 Tests
👁 Status
👁 Code style: ruff

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.

---

✨ 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 Unique

Agentflow stands out with powerful features designed for production-grade AI applications:

🏗️ Architecture & Scalability

1. 💾 Checkpointer with Caching Design Intelligent state persistence with built-in caching layer to scale efficiently. PostgreSQL + Redis implementation ensures high performance in production environments.

2. 🧠 3-Layer Memory System

   • Short-term memory: Current conversation context
   • Conversational memory: Session-based chat history
   • Long-term memory: Persistent knowledge across sessions

🔧 Advanced Tooling Ecosystem

3. 🔌 Remote Tool Calls Execute tools remotely using our TypeScript SDK for distributed agent architectures.

4. 🛠️ Comprehensive Tool Integration

   • Local tools (Python functions)
   • Remote tools (via TypeScript SDK)
   • Agent handoff tools (multi-agent collaboration)
   • MCP (Model Context Protocol)

🎯 Intelligent Context Management

5. 📏 Dedicated Context Manager
   • Automatically controls context size to prevent token overflow
   • Called at iteration end to avoid mid-execution context loss
   • Fully extensible with custom implementations

⚙️ Dependency Injection & Control

6. 💉 First-Class Dependency Injection Powered by InjectQ library for clean, testable, and maintainable code patterns.

7. 🎛️ Custom ID Generation Control Choose between string, int, or bigint IDs. Smaller IDs save significant space in databases and indexes compared to standard 128-bit UUIDs.

📊 Observability & Events

8. 📡 Internal Event Publishing Emit execution events to any publisher:
   • Kafka
   • RabbitMQ
   • Redis Pub/Sub
   • OpenTelemetry
   • Custom publishers

🔄 Advanced Execution Features

9. ⏰ Background Task Manager Built-in manager for running tasks asynchronously:

   • Prefetching data
   • Memory persistence
   • Cleanup operations
   • Custom background jobs

10. 🚦 Human-in-the-Loop with Interrupts Pause execution at any point for human approval, then seamlessly resume with full state preservation.

11. 🧭 Flexible Agent Navigation

    • Condition-based routing between agents
    • Command-based jumps to specific agents
    • Agent handoff tools for smooth transitions

🛡️ Security & Validation

12. 🎣 Comprehensive Callback System Hook into various execution stages for:
    • Logging and monitoring
    • Custom behavior injection
    • Prompt injection attack prevention
    • Input/output validation

📦 Ready-to-Use Components

13. 🤖 Prebuilt Agent Patterns Production-ready implementations:
    • React agents
    • RAG (Retrieval-Augmented Generation)
    • Swarm architectures
    • Router agents
    • MapReduce patterns
    • Supervisor teams

📐 Developer Experience

14. 📋 Pydantic-First Design All core classes (State, Message, ToolCalls) are Pydantic models:
    • Automatic JSON serialization
    • Type safety
    • Easy debugging and logging
    • Seamless database storage

---

Installation

Basic installation with uv (recommended):

uvpipinstall10xscale-agentflow

Or with pip:

pipinstall10xscale-agentflow

Optional Dependencies:

10xScale Agentflow supports optional dependencies for specific functionality:

# PostgreSQL + Redis checkpointing
pipinstall10xscale-agentflow[pg_checkpoint]

# MCP (Model Context Protocol) support
pipinstall10xscale-agentflow[mcp]

# Google GenAI adapter (google-genai SDK)
pipinstall10xscale-agentflow[google-genai]

# OpenAI adapter (openai SDK)
pipinstall10xscale-agentflow[openai]

# Realtime audio-to-audio agents (Gemini Live)
pipinstall10xscale-agentflow[realtime]

# Vector / long-term memory stores
pipinstall10xscale-agentflow[qdrant]# Qdrant store
pipinstall10xscale-agentflow[mem0]# Mem0 store

# Individual publishers
pipinstall10xscale-agentflow[redis]# Redis publisher
pipinstall10xscale-agentflow[kafka]# Kafka publisher
pipinstall10xscale-agentflow[rabbitmq]# RabbitMQ publisher
pipinstall10xscale-agentflow[otel]# OpenTelemetry tracing

# Multiple extras
pipinstall10xscale-agentflow[pg_checkpoint,mcp,google-genai,openai]

Environment Setup

Set your LLM provider API key:

exportOPENAI_API_KEY=sk-...# for OpenAI models
# or
exportGEMINI_API_KEY=...# for Google Gemini
# or
exportANTHROPIC_API_KEY=...# for Anthropic Claude

If you have a .env file, it will be auto-loaded (via python-dotenv).

---

🎯 Two Ways to Build Agents

10xScale Agentflow offers two approaches—choose based on your needs:

Approach	Best For	Lines of Code
Agent Class ⭐	Most use cases, rapid development	10-30 lines
Custom Functions	Complex custom logic, custom SDK integrations	50-150 lines

Recommendation: Start with the Agent class. It handles 90% of use cases with minimal code.

---

💡 Simple Example with Agent Class

Here's a complete tool-calling agent in under 30 lines:

fromagentflow.core.graphimport Agent, StateGraph, ToolNode
fromagentflow.core.stateimport AgentState, Message
fromagentflow.utils.constantsimport END


# 1. Define your tool
defget_weather(location: str) -> str:
"""Get weather for a location."""
 return f"The weather in {location} is sunny, 72°F"


# 2. Build the graph with Agent class
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]))


# 3. Define routing
defroute(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")

# 4. Run it!
app = graph.compile()
result = app.invoke({
 "messages": [Message.text_message("What's the weather in NYC?")]
}, config={"thread_id": "1"})

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

That's it! The Agent class handles message conversion, LLM calls, and tool integration automatically.

---

How to run the example locally

1. Install dependencies (recommended in a virtualenv):

pipinstall-rrequirements.txt
# or if you use uv
uvpipinstall-rrequirements.txt

2. Set your LLM provider API key (for example OpenAI):

exportOPENAI_API_KEY="sk-..."
# or create a .env with the key and the script will load it automatically

3. Run the example script:

pythonexamples/react/react_weather_agent.py

Notes:

• The example uses the OpenAI async client. Set OPENAI_API_KEY and choose a model available in your account.
• InMemoryCheckpointer is for demo/testing only. Replace with a persistent checkpointer for production.

---

Example: MCP Integration

10xScale Agentflow supports integration with Model Context Protocol (MCP) servers, allowing you to connect external tools and services. The example in examples/react-mcp/ demonstrates how to integrate MCP tools with your agent.

First, create an MCP server (see examples/react-mcp/server.py):

fromfastmcpimport FastMCP

mcp = FastMCP("My MCP Server")

@mcp.tool(
 description="Get the weather for a specific location",
)
defget_weather(location: str) -> dict:
 return {
 "location": location,
 "temperature": "22°C",
 "description": "Sunny",
 }

if __name__ == "__main__":
 mcp.run(transport="streamable-http")

Then, integrate MCP tools into your agent (from examples/react-mcp/react-mcp.py):

fromtypingimport Any

fromdotenvimport load_dotenv
fromfastmcpimport Client
fromopenaiimport AsyncOpenAI

fromagentflow.core.graphimport StateGraph, ToolNode
fromagentflow.core.stateimport AgentState, Message
fromagentflow.storage.checkpointerimport InMemoryCheckpointer
fromagentflow.utilsimport convert_messages
fromagentflow.utils.constantsimport END

load_dotenv()
client = AsyncOpenAI()

checkpointer = InMemoryCheckpointer()

config = {
 "mcpServers": {
 "weather": {
 "url": "http://127.0.0.1:8000/mcp",
 "transport": "streamable-http",
 },
 }
}

client_http = Client(config)

# Initialize ToolNode with MCP client
tool_node = ToolNode([], client=client_http)


async defmain_agent(state: AgentState):
 prompts = "You are a helpful assistant."

 messages = convert_messages(
 system_prompts=[{"role": "system", "content": prompts}],
 state=state,
 )

 # Get all available tools (including MCP tools)
 tools = await tool_node.all_tools()

 response = await client.chat.completions.create(
 model="gpt-4o-mini",
 messages=messages,
 tools=tools,
 )
 return response


defshould_use_tools(state: AgentState) -> str:
"""Determine if we should use tools or end the conversation."""
 if not state.context or len(state.context) == 0:
 return "TOOL"

 last_message = state.context[-1]

 if (
 hasattr(last_message, "tools_calls")
 and last_message.tools_calls
 and len(last_message.tools_calls) > 0
 ):
 return "TOOL"

 if last_message.role == "tool" and last_message.tool_call_id is not None:
 return END

 return END


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

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(checkpointer=checkpointer)

# Run the agent
inp = {"messages": [Message.text_message("Please call the get_weather function for New York City")]}
config = {"thread_id": "12345", "recursion_limit": 10}

res = app.invoke(inp, config=config)

for i in res["messages"]:
 print(i)

How to run the MCP example:

1. Install MCP dependencies:

pipinstall10xscale-agentflow[mcp]
# or
uvpipinstall10xscale-agentflow[mcp]

2. Start the MCP server in one terminal:

cdexamples/react-mcp
pythonserver.py

3. Run the MCP-integrated agent in another terminal:

pythonexamples/react-mcp/react-mcp.py

---

Example: Streaming Agent

10xScale Agentflow supports streaming responses for real-time interaction. The example in examples/react_stream/stream_react_agent.py demonstrates different streaming modes and configurations.

importasyncio
importlogging

fromdotenvimport load_dotenv
fromopenaiimport AsyncOpenAI

fromagentflow.core.graphimport StateGraph, ToolNode
fromagentflow.core.stateimport AgentState, Message
fromagentflow.storage.checkpointerimport InMemoryCheckpointer
fromagentflow.utilsimport ResponseGranularity, convert_messages
fromagentflow.utils.constantsimport END

load_dotenv()
client = AsyncOpenAI()
checkpointer = InMemoryCheckpointer()


defget_weather(
 location: str,
 tool_call_id: str,
 state: AgentState,
) -> Message:
"""Get weather with injectable parameters."""
 res = f"The weather in {location} is sunny."
 return Message.tool_message(
 content=res,
 tool_call_id=tool_call_id,
 )


tool_node = ToolNode([get_weather])


async defmain_agent(state: AgentState, config: dict):
 prompts = "You are a helpful assistant. Answer conversationally. Use tools when needed."

 messages = convert_messages(
 system_prompts=[{"role": "system", "content": prompts}],
 state=state,
 )

 is_stream = config.get("is_stream", False)

 if (
 state.context
 and len(state.context) > 0
 and state.context[-1].role == "tool"
 ):
 response = await client.chat.completions.create(
 model="gpt-4o-mini",
 messages=messages,
 stream=is_stream,
 )
 else:
 tools = await tool_node.all_tools()
 response = await client.chat.completions.create(
 model="gpt-4o-mini",
 messages=messages,
 tools=tools,
 stream=is_stream,
 )

 return response


defshould_use_tools(state: AgentState) -> str:
 if not state.context or len(state.context) == 0:
 return "TOOL"

 last_message = state.context[-1]

 if (
 hasattr(last_message, "tools_calls")
 and last_message.tools_calls
 and len(last_message.tools_calls) > 0
 ):
 return "TOOL"

 if last_message.role == "tool" and last_message.tool_call_id is not None:
 return END

 return END


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

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(checkpointer=checkpointer)


async defrun_stream_test():
 inp = {"messages": [Message.text_message("Call get_weather for Tokyo, then reply.")]}
 config = {"thread_id": "stream-1", "recursion_limit": 10}

 logging.info("--- streaming start ---")
 stream_gen = app.astream(
 inp,
 config=config,
 response_granularity=ResponseGranularity.LOW,
 )
 async for chunk in stream_gen:
 print(chunk.model_dump(), end="\n", flush=True)


if __name__ == "__main__":
 asyncio.run(run_stream_test())

Run the streaming example:

pythonexamples/react_stream/stream_react_agent.py

---

🎙️ Example: Realtime Audio Agent

Build a live, audio-to-audio agent over Gemini Live. The session is driven by a separate runtime (arealtime) because the provider owns the turn loop — you feed an input queue and consume normalized events (audio, transcripts, tool calls, barge-in).

importasyncio
fromagentflow.prebuilt.agentimport AudioAgent
fromagentflow.core.realtimeimport LiveInputQueue, RealtimeConfig

defget_weather(city: str) -> str:
"""Look up the weather for a city."""
 return f"It's sunny in {city}."

app = AudioAgent(
 "gemini-live-2.5-flash-preview",
 realtime_config=RealtimeConfig(model="gemini-live-2.5-flash-preview", voice="Puck"),
 system_prompt=[{"role": "system", "content": "You are a concise voice assistant."}],
 tools=[get_weather], # advertised to the model; runs React-style (incl. barge-in)
).compile()

async defmain():
 queue = LiveInputQueue()
 queue.send_audio(pcm16_bytes) # non-blocking; safe from an audio callback
 # queue.send_image(jpeg_bytes) # optional: send a still image / video frame
 async for event in app.arealtime(queue, {"thread_id": "t1"}):
 ... # AudioDeltaEvent / transcripts / ToolCallEvent / ...
 queue.close() # ends the session once the provider goes idle

asyncio.run(main())

Install the extra and set your key:

pipinstall10xscale-agentflow[realtime]
exportGEMINI_API_KEY=...

Highlights: barge-in, persisted transcripts (raw audio is never stored), automatic reconnect with session resumption, image/video frame input, and system_prompt / skills / memory working like a normal agent. See examples/realtime/.

---

⚡ Parallel Tool Execution

10xScale Agentflow automatically executes multiple tool calls in parallel when an LLM requests multiple tools simultaneously. This dramatically improves performance for I/O-bound operations.

Benefits

• Faster Response Times: Multiple API calls execute concurrently
• Better Resource Utilization: Don't wait for one tool to finish before starting the next
• Seamless Integration: Works automatically with existing code - no changes needed

Example Performance

# LLM requests 3 tools simultaneously:
# - get_weather("NYC") # Takes 1.0s
# - get_news("tech") # Takes 1.5s
# - get_stock("AAPL") # Takes 0.8s

# Sequential execution: 1.0 + 1.5 + 0.8 = 3.3 seconds
# Parallel execution: max(1.0, 1.5, 0.8) = 1.5 seconds ⚡

See the parallel tool execution documentation for more details.

---

🎯 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 repository
gitclonehttps://github.com/10xhub/10xScaleAgentflow.git
cd10xScaleAgentflow

# Create virtual environment
python-mvenv.venv
source.venv/bin/activate# On Windows: .venv\Scripts\activate

# Install dev dependencies
pipinstall-rrequirements-dev.txt
# or
uvpipinstall-rrequirements-dev.txt

# Run tests
maketest
# or
pytest-q

# Build docs
makedocs-serve# Serves at http://127.0.0.1:8000

# Run examples
cdexamples/react
pythonreact_sync.py

Development Tools

The project uses:

• pytest for testing (with async support)
• ruff for linting and formatting
• mypy for type checking
• mkdocs with Material theme for documentation
• coverage for test coverage reports

See pyproject.dev.toml for complete tool configurations.

---

🗺️ 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

---

📄 License

MIT License - see LICENSE for details.

---

🔗 Links & Resources

• Documentation - Full documentation with tutorials and API reference
• GitHub Repository - Source code and issues
• PyPI Project - Package releases
• Examples Directory - Runnable code samples

---

🙏 Contributing

Contributions are welcome! Please see our GitHub repository for:

• Issue reporting and feature requests
• Pull request guidelines
• Development setup instructions
• Code style and testing requirements

---

💬 Support

• Documentation: https://10xhub.github.io/Agentflow/
• Examples: Check the examples directory
• Issues: Report bugs on GitHub Issues
• Discussions: Ask questions in GitHub Discussions

---

Ready to build intelligent agents? Check out the documentation to get started!

Verified details

These details have been verified by PyPI

Maintainers

👁 Avatar for iamsdt from gravatar.com
iamsdt

Unverified details

These details have not been verified by PyPI

Project links

• Documentation
• Homepage
• Issues
• Repository

• Download URL: 10xscale_agentflow-0.8.0.tar.gz
• Upload date: Jun 15, 2026
• Size: 465.2 kB
• Tags: Source
• Uploaded using Trusted Publishing? No
• Uploaded via: twine/6.2.0 CPython/3.13.13

Hashes for 10xscale_agentflow-0.8.0.tar.gz

Algorithm	Hash digest
SHA256	02299595bcca4d258c7e8ac3e3aa71642bb601be340e6d02549d6bcf58fab98b
MD5	a3a35011338c165ceb1d73e4c73244d9
BLAKE2b-256	fc714429c31248afd982d8d916832e563b3c064bf28251f525473690c13fc23f

See more details on using hashes here.

• Download URL: 10xscale_agentflow-0.8.0-py3-none-any.whl
• Upload date: Jun 15, 2026
• Size: 504.8 kB
• Tags: Python 3
• Uploaded using Trusted Publishing? No
• Uploaded via: twine/6.2.0 CPython/3.13.13

Hashes for 10xscale_agentflow-0.8.0-py3-none-any.whl

Algorithm	Hash digest
SHA256	3f95c1b2cc8a4bfe3fdc239a0e3f5ef4e058f21334f3af19df42b72d85e9b569
MD5	fe0c10135c62a116b453e3583a9b01aa
BLAKE2b-256	10595ed8d309a55e589ba1399d8a611652bccdf8a44890dcb3df245316655cb2

See more details on using hashes here.