mcp-flow 0.1.0

Lightweight MCP orchestrator: parallel tool calls, local filtering, token trimming, and safe sandboxed flow scripts for AI agents.

mcp-flow

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-flow

---

The problem vs the solution

STANDARD 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

---

Quickstart (≈10 lines)

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

Fluent API surface

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] | value

Tool calls are intentionally noisy-free for codegen:

("server", "tool", {"arg": 1})
ToolCall(server="s", tool="t", arguments={...})
{"server": "s", "tool": "t", "arguments": {...}}

---

Token trimmer

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.

---

Safe sandbox for LLM-generated flows

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.

---

Transports

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.

---

LLM system prompt template

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)

---

Benchmark (illustrative)

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.

---

Install & develop

pip install mcp-flow

# from source
git clone https://github.com/mano7onam/mcp-flow.git
cd mcp-flow
pip install -e ".[dev]"
pytest

Requirements: Python 3.10+ · dependencies: mcp, pydantic (plus their transitive runtime stack).

---

Project layout

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

---

Why not LangGraph / CrewAI / …

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)

---

License

MIT © mano7onam