Lightweight MCP orchestrator for AI agents — parallel tool calls, local filtering, token trimming, and a safe sandbox for LLM-generated control flow.
pip install mcp-flow
# or
uv add mcp-flowSTANDARD TOOL-CALLING (sequential, slow, expensive)
──────────────────────────────────────────────────
LLM ──► tool A ──► wait ──► LLM ──► tool B ──► wait ──► LLM ──► tool C
│ │ │
└──── full JSON into context every hop ─┘
latency ≈ sum(A+B+C) tokens ≈ 3× raw payloads
mcp-flow (parallel, local filter, token-bounded)
────────────────────────────────────────────────
LLM ──► Flow.parallel(A, B, C) ──► local filter/trim ──► LLM
│ │ │
└──asyncio.gather──┘
latency ≈ max(A,B,C) tokens ≤ limit_tokens(N)
| Sequential MCP | mcp-flow | |
|---|---|---|
| Latency (3 tools @ 200ms) | ~600ms | ~200ms |
| Round-trips to LLM | 3–4 | 1 |
| Context tokens | full raw JSON | capped (limit_tokens) |
| Dependencies | frameworks | mcp + pydantic only |
| LLM-written scripts | unsafe exec |
AST sandbox |
import asyncio
from mcp_flow import Flow, MCPClientPool
async def main():
async with MCPClientPool() as pool:
await pool.add_stdio(
"fs", "npx",
["-y", "@modelcontextprotocol/server-filesystem", "."],
)
await pool.add_http("api", "http://127.0.0.1:8000/mcp")
results = await (
Flow(pool)
.parallel(
("fs", "read_file", {"path": "README.md"}),
("api", "search", {"q": "mcp"}),
)
.filter(lambda r: r.ok)
.limit_tokens(1500)
.run()
)
for r in results:
print(r.server, r.tool, r.data)
asyncio.run(main())Flow(pool)
.parallel(*tool_calls) # asyncio.gather across servers
.chain(*steps) # sequential; use "$prev" in args
.fallback(primary, *backups) # automatic failover
.race(*tool_calls) # first success wins
.each(items, tool_call) # bounded concurrent map
.filter(lambda r: ...) # local Python predicate
.map(lambda r: ...) # project results
.select(".items[*].id") # jq-like path (no jq dep)
.when(pred, step) # conditional branch
.limit_tokens(2000) # structural JSON trim
.run() # -> list[ToolResult] | valueTool calls are intentionally noisy-free for codegen:
("server", "tool", {"arg": 1})
ToolCall(server="s", tool="t", arguments={...})
{"server": "s", "tool": "t", "arguments": {...}}Large MCP responses (directory listings, search hits, DB rows) blow up context windows. mcp-flow trims structurally — no tokenizer download required:
from mcp_flow import trim_payload, TrimConfig, estimate_tokens
cfg = TrimConfig(max_tokens=800, max_array_items=10, drop_nulls=True)
small = trim_payload(huge_json, cfg)
print(estimate_tokens(small))Strategy: drop nulls → cap arrays/strings → keep priority keys (id, name, error, …) → hard preview as last resort.
Let the model emit a short script; evaluate it without os, open, imports, or dunder escapes:
from mcp_flow import FlowSandbox, Flow
script = """
await flow.parallel(
("fs", "read_file", {"path": "a.txt"}),
("web", "search", {"q": "mcp"}),
).filter(lambda r: r.ok).limit_tokens(1000).run()
"""
sandbox = FlowSandbox(flow=Flow(pool), timeout=30)
results = await sandbox.run(script)Rejected: import, eval/exec, open, __class__, function/class definitions, etc.
Allowed: flow, ToolCall, lambdas, list ops, trimmer helpers.
| Transport | Register | Typical use |
|---|---|---|
| stdio | pool.add_stdio(name, command, args) |
Local MCP servers (npx, uvx, binaries) |
| SSE | pool.add_sse(name, url) |
Legacy remote HTTP+SSE |
| HTTP | pool.add_http(name, url) |
Streamable HTTP (MCP 2025-03-26+) |
Auto-reconnect (configurable), call timeouts, and async with graceful shutdown are built in.
Paste into your agent instructions:
You orchestrate Model Context Protocol tools via the `mcp-flow` Python library.
Do NOT call tools one-by-one with intermediate reasoning when they are independent.
Instead, output a single Python snippet using only the injected `flow` object:
Rules:
1. Use flow.parallel(...) for independent tool calls.
2. Use flow.chain(...) when step N needs output of step N-1 (placeholder "$prev").
3. Use flow.fallback(primary, backup) for resilient calls.
4. Always .filter(lambda r: r.ok) and .limit_tokens(1500) before returning.
5. Tool call form: ("server_name", "tool_name", {..args..})
6. No imports, no files, no network except via flow — code runs in a sandbox.
7. End with .run() (or assign to `results`).
Example:
results = await (
flow.parallel(
("fs", "read_file", {"path": "README.md"}),
("gh", "search_issues", {"q": "bug label:p0"}),
)
.filter(lambda r: r.ok)
.limit_tokens(1500)
.run()
)
Then:
results = await FlowSandbox(flow=Flow(pool)).run(llm_code)Measured pattern on localhost mocks (3 tools × ~50ms each, 50-hit search payload):
| Mode | Wall time | Est. tokens to LLM |
|---|---|---|
| Sequential tool→LLM→tool | ~180ms + 3 LLM RTTs | ~12 000 |
| Naive parallel, no trim | ~55ms + 1 LLM RTT | ~12 000 |
| mcp-flow parallel + limit_tokens(1500) | ~55ms + 1 LLM RTT | ≤1500 |
Exact numbers depend on servers and models; the architecture guarantees O(max latency) tool wait and bounded context.
pip install mcp-flow
# from source
git clone https://github.com/mano7onam/mcp-flow.git
cd mcp-flow
pip install -e ".[dev]"
pytestRequirements: Python 3.10+ · dependencies: mcp, pydantic (plus their transitive runtime stack).
src/mcp_flow/
client.py # MCPClientPool — stdio / SSE / HTTP
flow.py # Fluent Flow engine
trimmer.py # Token / JSON optimizer
sandbox.py # AST-safe script runner
exceptions.py # Error types
Those are full agent frameworks. mcp-flow is a scalpel:
- zero graph DSL, zero multi-agent runtime
- optimized for tool fan-out + context hygiene
- safe eval surface for codegen agents
- works next to any agent stack (or raw LLM loop)
MIT © mano7onam