使用说明

mcp管道

MCP服务器的状态管道框架。

类型安全状态。声明式工具链。更少的工具，更少的代币。

英语· 韩语

问题

MCP服务器存在令牌问题。每次LLM调用工具时 工具描述的完整列表 在系统提示中发送。更多的工具=更多的代币=更多的成本=更差的准确性。

以下是当您将一些MCP服务器连接到Claude时实际发生的情况：

设置	工具描述	200k上下文的百分比
仅GitHub MCP（93个工具）	~55000个代币	27.5%
7台MCP服务器平均	~67300个令牌	33.7%
5台服务器×每台30个工具	~60000个令牌	30.0%

你的上下文窗口的三分之一——在对话开始之前就消失了。 更糟糕的是：每一次往返都会重新发送这一切。

但代币浪费并不仅限于工具描述。工具也不能共享状态：

# FastMCP — each tool is isolated
@server.tool()
async def search_products(query: str) -> dict:
    results = await db.search(query)
    return {"products": results}  # 500 tokens of product data

@server.tool()
async def buy_product(product_id: str) -> dict:
    # Where is this product_id from? The LLM has to relay it.
    # That means the 500 tokens of search results live in the LLM context
    # and get re-sent on this call.
    ...

LLM成为数据中继。它接收搜索结果（500个令牌），将其存储在上下文中，然后将相关比特传递给下一个工具。 每一条数据都流经LLM——即使不需要。

解决方案

两个想法：

工具更少 --将5个小工具合并为一个管道工具，在内部执行多项任务
服务器端状态 --让服务器在工具调用之间缓存数据，这样LLM就不必中继它

from mcp_pipeline import PipelineMCP, State
from typing import Any

# 1. Define typed state
class ShopState(State):
    search_results: dict[str, Any] = {}
    cart: list[dict] = []

server = PipelineMCP("shop", state=ShopState)

# 2. This tool stores its results in state
@server.tool(stores="search_results")
async def search(query: str, state: ShopState) -> dict:
    """Search products. Results are cached server-side."""
    products = await db.search(query)
    # state.search_results is auto-populated with the return value
    return {
        "products": [
            {"id": p.id, "name": p.name, "price": p.price}
            for p in products[:5]  # Only send top 5 to LLM (compressed)
        ]
    }

# 3. This tool requires previous results
@server.tool(requires="search_results")
async def buy(product_id: str, state: ShopState) -> dict:
    """Buy a product from search results. Requires search first."""
    product = state.search_results[product_id]  # From server cache, not LLM context
    order = await db.create_order(product)
    return {"order_id": order.id, "status": "confirmed"}

发生了什么变化：

LLM不存储500个产品数据令牌。服务器确实如此。
buy 直接从服务器状态访问搜索结果。
如果有人打电话来 buy 没有 search，他们会得到一个有用的错误，而不是崩溃。

之前和之后

之前：传统MCP（14个工具，9次往返）

Round 1: LLM → list_platforms() → LLM         # "which platforms exist?"
Round 2: LLM → get_trending("reddit") → LLM   # "what's trending?"
Round 3: LLM → search("MCP server") → LLM     # "find relevant posts"
Round 4: LLM → analyze_post(id) → LLM         # "is this post good?"
Round 5: LLM → get_post(id) → LLM             # "get full content"
Round 6: LLM → get_comments(id) → LLM         # "get the comments"
Round 7: LLM → check_rate_limit() → LLM       # "can I still post?"
Round 8: LLM → preview(content) → LLM         # "dry run"
Round 9: LLM → write_comment(id, text) → LLM  # "actually post"

Each round: resends 14 tool descriptions (~3,000 tokens) + full conversation history
Total tool description overhead: 14 × 9 = ~27,000 tokens

之后：管道MCP（3个工具，3次往返）

Round 1: LLM → scout("MCP server", ["reddit"]) → LLM
         Server internally: trending + search + analyze + rank + filter
         Returns: top 3 opportunities (compressed, ~200 tokens)

Round 2: LLM → draft("opp_1") → LLM
         Server internally: get_post + get_comments + analyze_tone
         Uses cached opportunity from scout (server state)
         Returns: context summary (~300 tokens)
         LLM generates natural content

Round 3: LLM → strike("opp_1", "comment", "content...") → LLM
         Server internally: write_comment + record_history
         Uses cached context from draft (server state)
         Returns: {"url": "...", "status": "posted"}

Each round: resends 3 tool descriptions (~500 tokens) + conversation history
Total tool description overhead: 3 × 3 = ~4,500 tokens

代币节省：减少约83% 仅在工具描述开销方面。此外，LLM不会在工具之间中继中间数据。

运作原理

1.类型安全状态

from mcp_pipeline import State

class MyState(State):
    search_results: dict[str, Any] = {}     # Tool A stores here
    processed_data: dict[str, Any] = {}     # Tool B stores here
    action_history: list[dict] = []         # Tool C appends here

与FastMCP的内置状态相比：

# FastMCP — string keys, no type safety, known bugs (#2098)
await ctx.set_state("search_results", data)
results = await ctx.get_state("search_results")  # Returns Any, no autocomplete

# mcp-pipeline — typed fields, IDE autocomplete, compile-time checks
state.search_results = data                        # Type-checked
results = state.search_results                     # IDE knows the type

2. `stores` 和 `requires` --声明性依赖关系

@server.tool(stores="search_results")
async def scout(query: str, state: MyState) -> dict:
    """This tool's return value is automatically saved to state.search_results."""
    ...

@server.tool(requires="search_results")
async def act(result_id: str, state: MyState) -> dict:
    """This tool needs scout to have run first."""
    ...

当 requires 未满足：

# LLM calls act() before scout()
# Instead of crashing or returning garbage:
{
    "error": "Required state 'search_results' is empty.",
    "hint": "Call scout(query=...) first to populate search results.",
    "required_by": "act"
}
# The LLM reads this and knows to call scout first.
# No wasted API call. No stack trace. Just guidance.

3.自动生成 `_status` 工具

每 PipelineMCP 服务器自动获得 _status 工具：

# LLM calls _status()
{
    "state": {
        "search_results": {"populated": true, "count": 5},
        "processed_data": {"populated": false},
        "action_history": {"populated": true, "count": 2}
    },
    "tools": {
        "available": ["scout", "act"],       # requires are met
        "blocked": ["process"]               # waiting on dependencies
    }
}

LLM总是知道：存在什么数据，可以运行什么工具，什么是阻塞。

4.管道压缩

核心设计原则： 将编排从LLM移动到服务器。

Traditional: LLM orchestrates
┌─────┐      ┌─────┐      ┌─────┐      ┌─────┐
│ LLM │─────▶│Tool1│─────▶│ LLM │─────▶│Tool2│───▶ ...
│     │◀─────│     │◀─────│relay│◀─────│     │◀──
└─────┘      └─────┘      └─────┘      └─────┘
  LLM shuttles data between tools (expensive)

Pipeline: Server orchestrates
┌─────┐      ┌──────────────────────────────┐
│ LLM │─────▶│ PipelineTool                 │
│     │◀─────│  internally: Tool1 → Tool2   │
└─────┘      │  state cached server-side    │
             └──────────────────────────────┘
  One call, server handles the rest (cheap)

完整示例：社交媒体代理

from mcp_pipeline import PipelineMCP, State
from typing import Any

class SocialState(State):
    opportunities: dict[str, Any] = {}
    contexts: dict[str, Any] = {}
    history: list[dict] = []

server = PipelineMCP("social-agent", state=SocialState)

@server.tool(stores="opportunities")
async def scout(topic: str, platforms: list[str] | None = None) -> dict:
    """Scan developer communities for relevant discussions.
    Internally runs: trending + search + score + filter across all platforms.
    Returns top opportunities with IDs for follow-up."""
    all_posts = []
    for platform in get_active_platforms(platforms):
        trending = await platform.get_trending()
        searched = await platform.search(topic)
        all_posts.extend(trending + searched)

    scored = score_relevance(all_posts, topic)
    top = scored[:5]

    return {
        "opportunities": [
            {
                "id": f"opp_{i}",
                "platform": opp.platform,
                "title": opp.title[:80],
                "relevance": round(opp.score, 2),
                "comments": opp.comment_count,
                "reason": opp.reason,
            }
            for i, opp in enumerate(top)
        ],
        "total_scanned": len(all_posts),
        "summary": f"{len(top)} opportunities across {len(set(o.platform for o in top))} platforms",
    }

@server.tool(stores="contexts", requires="opportunities")
async def draft(opportunity_id: str, state: SocialState) -> dict:
    """Gather full context for a specific opportunity.
    Reads the post, its comment tree, and analyzes the conversation tone.
    Returns a context summary for the LLM to craft a response."""
    opp = state.opportunities[opportunity_id]
    post = await get_post(opp)
    comments = await get_comments(opp)

    return {
        "title": post.title,
        "body_summary": post.body[:500],
        "comment_count": len(comments),
        "top_comments": [c.body[:200] for c in comments[:5]],
        "tone": analyze_tone(comments),
        "suggested_approach": suggest_approach(post, comments),
    }

@server.tool(requires="contexts")
async def strike(
    opportunity_id: str,
    action: str,
    content: str,
    state: SocialState,
) -> dict:
    """Execute: write a comment, reply, or new post.
    Uses cached context from draft. Records action in history."""
    ctx = state.contexts[opportunity_id]
    result = await write_to_platform(ctx, action, content)

    state.history.append({
        "opportunity": opportunity_id,
        "action": action,
        "url": result.url,
        "timestamp": now(),
    })

    return {"url": result.url, "status": "posted"}

3个工具。3次往返。从发现到行动的完整工作流程。

比较

与原始FastMCP

	FastMCP	mcp管道
州API	`ctx.set_state("key", val)` --字符串键， `Any` 类型	`state.field = val` --键入，IDE自动补全
状态错误	已知问题使用get/set	基于简单的数据类构建——可预测
工具依赖关系	无--所有工具都是独立的	`stores`/`requires` --声明性链接
缺少依赖关系	无声失败或崩溃	引导错误：“先调用X”
状态自省	自己构建	自动生成 `_status` 工具
设计理念	通用工具包	令牌高效管道设计

vs mcp工作流程

	mcp工作流程	mcp管道
语言	TypeScript	python
方法	基于步骤的工作流引擎	声明性状态依赖关系
国家	`WorkflowSessionManager` （地图）	类型州类
复杂性	工作流DSL、分支、暂停/恢复	最小API： `stores`/`requires`
集成	自己的工作流引擎	包装FastMCP

vs mcp代理

	mcp代理	mcp管道
焦点	代理编排（客户端）	MCP服务器设计（服务器端）
问题	“如何协调多个代理？”	“如何减少每台服务器的令牌？”
状态	LLM包装器中的对话历史	服务器端键入状态
输出	多代理工作流	更少、更智能的工具

安装

pip install mcp-pipeline

仅取决于 mcp[cli] （FastMCP）。没有其他依赖关系。

建筑

mcp_pipeline/
├── __init__.py      # PipelineMCP, State exports
├── server.py        # PipelineMCP (wraps FastMCP, adds state injection)
├── state.py         # State base class (dataclass-style, field introspection)
├── decorators.py    # stores/requires logic (validation, auto-save, error guidance)
└── status.py        # Auto-generated _status tool

┌───────────────────────────────────────────────┐
│  LLM Agent                                    │
│  Sees: 3-4 tools (not 14)                     │
│  Does: judgment, content generation            │
└──────────────┬────────────────────────────────┘
               │ minimal round trips
┌──────────────▼────────────────────────────────┐
│  PipelineMCP Server                           │
│                                               │
│  ┌──────────┐  ┌──────────┐  ┌────────────┐  │
│  │  Tool A  │─▶│  State   │◀─│   Tool B   │  │
│  │ stores=X │  │  (typed) │  │ requires=X │  │
│  └──────────┘  └──────────┘  └────────────┘  │
│                                               │
│  Server handles: data fetching, analysis,     │
│  filtering, caching, orchestration            │
└───────────────────────────────────────────────┘

发展

git clone https://github.com/SonAIengine/mcp-pipeline.git
cd mcp-pipeline
pip install -e ".[dev]"

pytest
mypy mcp_pipeline/
ruff check mcp_pipeline/

许可证

麻省理工学院

使用说明

mcp管道

问题

解决方案

之前和之后

之前：传统MCP（14个工具，9次往返）

之后：管道MCP（3个工具，3次往返）

运作原理

1.类型安全状态

2. stores 和 requires --声明性依赖关系

3.自动生成 _status 工具

4.管道压缩

完整示例：社交媒体代理

比较

与原始FastMCP

vs mcp工作流程

vs mcp代理

安装

建筑

发展

许可证

相关 MCP

浏览器自动化代理

国际象棋状态管理服务器

模型上下文协议服务器

任务编排服务器

RSS MCP服务器

AI代理休息服务器

2. `stores` 和 `requires` --声明性依赖关系

3.自动生成 `_status` 工具