Mesh Agent Graph Orchestration
Build agent and multi-agent workflows as executable graphs with token-by-token streaming, state management, and seamless integration with Vel and OpenAI Agents SDK.
Get started now View on GitHub
Features
- Graph-Based Workflows
- Build agent workflows as directed graphs with controlled cycles, using programmatic or declarative interfaces
- Dual API
- LangGraph-style programmatic builder or React Flow JSON (Flowise-compatible)
- Token-by-Token Streaming
- Real-time streaming with provider-agnostic events via AsyncIterator
- Event Translation
- Use Vel’s standardized events or provider-native events with opt-in flag
- Multi-Provider Support
- OpenAI, Anthropic, Google via Vel, plus native OpenAI Agents SDK integration
- 7 Core Node Types
- Start, End, Agent, LLM, Tool, Condition, Loop
- State Persistence
- Pluggable backends (SQLite, in-memory, custom)
- Variable Resolution
- Template variables for dynamic workflows (
,, etc.) - Production Ready
- Error handling, retries, structured logging, and SSE streaming
Quick Start
Installation
# Basic installation
pip install mesh
# With Vel SDK support
pip install mesh[vel]
# With OpenAI Agents SDK
pip install mesh[agents]
# All features
pip install mesh[all]
Basic Usage
import asyncio
from mesh import StateGraph, Executor, ExecutionContext, MemoryBackend
async def main():
# Build graph
graph = StateGraph()
graph.add_node("llm", None, node_type="llm", model="gpt-4")
graph.add_edge("START", "llm")
graph.set_entry_point("llm")
# Compile and execute
compiled = graph.compile()
executor = Executor(compiled, MemoryBackend())
context = ExecutionContext(
graph_id="my-graph",
session_id="session-1",
chat_history=[],
variables={},
state={}
)
# Stream results
async for event in executor.execute("What is 2+2?", context):
if event.type == "token":
print(event.content, end="", flush=True)
asyncio.run(main())
Documentation
Getting Started
Installation and configuration guide
Quick Start
Build your first agent graph in minutes
Graphs
Understanding graph structure and execution
Nodes
7 core node types and how to use them
Events
Provider-agnostic event streaming
Streaming
Token-by-token streaming patterns
Variables
Template variable resolution
Vel Integration
Using Vel agents with Mesh
OpenAI Agents SDK
OpenAI Agents SDK integration
Event Translation
Vel vs native event handling
Examples
Complete working examples
API Reference
Complete API documentation
Architecture
┌─────────────────────────────────────────────┐
│ User Application │
│ (FastAPI/Flask/Django) │
└────────────────┬────────────────────────────┘
│
▼
┌─────────────────────────────────────────────┐
│ MESH LIBRARY │
├─────────────────────────────────────────────┤
│ ┌──────────────┐ ┌──────────────┐ │
│ │ Parsers │ │ Builders │ │
│ │ - ReactFlow │ │ - StateGraph│ │
│ └──────┬───────┘ └──────┬───────┘ │
│ └──────────┬─────────┘ │
│ ▼ │
│ ┌──────────────────┐ │
│ │ Graph Compiler │ │
│ └────────┬─────────┘ │
│ ▼ │
│ ┌──────────────────┐ │
│ │ Execution Engine │ │
│ │ - Queue-based │ │
│ │ - Streaming │ │
│ └────────┬─────────┘ │
│ ▼ │
│ ┌───────────────────────┐ │
│ │ Node Implementations│ │
│ │ Agent│LLM│Tool│... │ │
│ └───────────────────────┘ │
└─────────────────────────────────────────────┘
Key Concepts
StateGraph Builder
LangGraph-style programmatic API for building graphs:
from mesh import StateGraph
graph = StateGraph()
graph.add_node("agent", my_agent, node_type="agent")
graph.add_node("tool", my_function, node_type="tool")
graph.add_edge("START", "agent")
graph.add_edge("agent", "tool")
graph.set_entry_point("agent")
compiled = graph.compile()
Event Streaming
Provider-agnostic events via Vel translation by default:
async for event in executor.execute(input, context):
if event.type == "token":
print(event.content, end="", flush=True)
elif event.type == "message_complete":
print("\n[Done]")
Or use native provider events:
graph.add_node("agent", agent, node_type="agent", use_native_events=True)
State Management
Persistent state across executions:
from mesh.backends import SQLiteBackend
backend = SQLiteBackend("mesh_state.db")
executor = Executor(compiled, backend)
# State automatically persists
Why Mesh?
- Zero Framework Lock-in: No LangChain dependency, use any agent SDK
- Flowise Compatible: Parse and execute React Flow JSON workflows
- Production Ready: State persistence, error handling, retries
- Streaming First: Token-by-token streaming with AsyncIterator
- Provider Agnostic: Consistent events across OpenAI, Anthropic, Google
- Flexible: Programmatic or declarative, Vel or native events
Examples
See the examples directory for:
- Simple LLM streaming
- Vel agent integration
- OpenAI Agents SDK integration
- Event translation comparison
- React Flow JSON parsing
- FastAPI server with SSE streaming
License
Mesh is distributed under the MIT License.
Credits
Inspired by: