Metadata-Version: 2.4
Name: mcp-flow
Version: 0.1.0
Summary: Lightweight MCP orchestrator: parallel tool calls, local filtering, token trimming, and safe sandboxed flow scripts for AI agents.
Project-URL: Homepage, https://github.com/mano7onam/mcp-flow
Project-URL: Repository, https://github.com/mano7onam/mcp-flow
Project-URL: Issues, https://github.com/mano7onam/mcp-flow/issues
Project-URL: Changelog, https://github.com/mano7onam/mcp-flow/releases
Author-email: mano7onam <mano7onamm@gmail.com>
License: MIT
License-File: LICENSE
Keywords: ai-agents,asyncio,llm,mcp,model-context-protocol,orchestration,token-trimmer,tool-calling
Classifier: Development Status :: 4 - Beta
Classifier: Framework :: AsyncIO
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3 :: Only
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Typing :: Typed
Requires-Python: >=3.10
Requires-Dist: mcp>=1.0.0
Requires-Dist: pydantic>=2.0
Provides-Extra: dev
Requires-Dist: build; extra == 'dev'
Requires-Dist: mypy>=1.8; extra == 'dev'
Requires-Dist: pytest-asyncio>=0.23; extra == 'dev'
Requires-Dist: pytest>=7; extra == 'dev'
Requires-Dist: twine; extra == 'dev'
Description-Content-Type: text/markdown

# 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](https://img.shields.io/pypi/v/mcp-flow.svg)](https://pypi.org/project/mcp-flow/)
[![Python](https://img.shields.io/pypi/pyversions/mcp-flow.svg)](https://pypi.org/project/mcp-flow/)
[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](LICENSE)
[![CI](https://github.com/mano7onam/mcp-flow/actions/workflows/test.yml/badge.svg)](https://github.com/mano7onam/mcp-flow/actions/workflows/test.yml)

```bash
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)

```python
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

```python
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:

```python
("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:

```python
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:

```python
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:

```text
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:

```python
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

```bash
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

```text
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](https://github.com/mano7onam)
