Skip to main content

mcp-flow

Lightweight MCP orchestrator for AI agents — parallel tool calls, local filtering, token trimming, and a safe sandbox for LLM-generated control flow.

PyPI Python License: MIT CI

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

Metadata

Release files for mcp-flow 0.1.1

For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.

Source distribution (sdist)

Source distribution for mcp-flow 0.1.1
File Size Uploaded
mcp_flow-0.1.1.tar.gz 28.8 kB Details

Built distribution (wheel)

Table of built distributions (wheels) for mcp-flow 0.1.1
File Interpreter ABI Platform
mcp_flow-0.1.1-py3-none-any.whl Python 3 none any Details

Total release size: 52.1 kB

Release files / mcp_flow-0.1.1.tar.gz

Download URL mcp_flow-0.1.1.tar.gz
Size 28.8 kB
Tags Source
SHA-256 checksum
How to use checksums
43098c82606246e00027e01ed86b7d0be4d4ab2c377f21ba21270c45b40c0d55
BLAKE2b-256 checksum
How to use checksums
c29934098b9b943492e2bced8c97b59ede6a476beb5e58969ca9ead688e06d9b
Upload date
Uploaded using Trusted Publishing?
What is trusted publishing?
No
Uploaded via twine/6.2.0 CPython/3.12.13

Release files / mcp_flow-0.1.1-py3-none-any.whl

Download URL mcp_flow-0.1.1-py3-none-any.whl
Size 23.3 kB
Tags Python 3
SHA-256 checksum
How to use checksums
fbadd275a8b788fa4f6eeb1fc8dd4bab36abe888d7645950248deb53ee3a018d
BLAKE2b-256 checksum
How to use checksums
5ca379d3287ea5da9554ed089d9d58eeab9eb1c9f5b0fec2b30ec58430160f08
Upload date
Uploaded using Trusted Publishing?
What is trusted publishing?
No
Uploaded via twine/6.2.0 CPython/3.12.13

Release history Release notifications | RSS feed

This release

0.1.1 This release

2 release files

0.1.0

2 release files

Anthropic, PBC Visionary sponsor Bloomberg Visionary sponsor Hudson River Trading Visionary sponsor Meta Visionary sponsor NVIDIA Visionary sponsor Microsoft Sustainability sponsor Depot Continuous Integration AWS Cloud computing and Security Sponsor Datadog Monitoring Fastly CDN Google Download Analytics Sentry Error logging StatusPage Status page