mcp-chaining

Compare original and translation side by side

🇺🇸

Original

English

🇨🇳

Translation

Chinese

MCP Chaining Pipeline

MCP串联工作流管道

A research-to-implement pipeline that chains 5 MCP tools for end-to-end workflows.

这是一个串联5个MCP工具的从研究到实现的工作流管道，支持端到端工作流。

When to Use

适用场景

Building multi-tool MCP pipelines
Understanding how to chain MCP calls with graceful degradation
Debugging MCP environment variable issues
Learning the tool naming conventions for different MCP servers

构建多工具MCP工作流管道
了解如何通过优雅降级来串联MCP调用
调试MCP环境变量问题
学习不同MCP服务器的工具命名规范

What We Built

我们构建的内容

A pipeline that chains these tools:


nia__search
ast-grep__find_code
morph__warpgrep_codebase_search
qlty__qlty_check
git__git_status

Step	Server	Tool ID	Purpose
1	nia	`nia__search`	Search library documentation
2	ast-grep	`ast-grep__find_code`	Find AST code patterns
3	morph	`morph__warpgrep_codebase_search`	Fast codebase search
4	qlty	`qlty__qlty_check`	Code quality validation
5	git	`git__git_status`	Git operations

一个串联以下工具的工作流管道：


nia__search
ast-grep__find_code
morph__warpgrep_codebase_search
qlty__qlty_check
git__git_status

步骤	服务器	工具ID	用途
1	nia	`nia__search`	搜索库文档
2	ast-grep	`ast-grep__find_code`	查找AST代码模式
3	morph	`morph__warpgrep_codebase_search`	快速代码库搜索
4	qlty	`qlty__qlty_check`	代码质量验证
5	git	`git__git_status`	Git操作

Key Files

关键文件

```
scripts/research_implement_pipeline.py
```
- Main pipeline implementation
```
scripts/test_research_pipeline.py
```
- Test harness with isolated sandbox
```
workspace/pipeline-test/sample_code.py
```
- Test sample code

```
scripts/research_implement_pipeline.py
```
- 主管道实现文件
```
scripts/test_research_pipeline.py
```
- 带隔离沙箱的测试工具
```
workspace/pipeline-test/sample_code.py
```
- 测试示例代码

Usage Examples

使用示例

bash

undefined

bash

undefined

Dry-run pipeline (preview plan without changes)

试运行管道（预览计划但不执行变更）

uv run python -m runtime.harness scripts/research_implement_pipeline.py
--topic "async error handling python"
--target-dir "./workspace/pipeline-test"
--dry-run --verbose

Run tests

运行测试

uv run python -m runtime.harness scripts/test_research_pipeline.py --test all

View the pipeline script

查看管道脚本

cat scripts/research_implement_pipeline.py

undefined

cat scripts/research_implement_pipeline.py

undefined

Critical Fix: Environment Variables

关键修复：环境变量

The MCP SDK's

get_default_environment()

only includes basic vars (PATH, HOME, etc.), NOT

os.environ

. We fixed

src/runtime/mcp_client.py

to pass full environment:

python

undefined

MCP SDK的

get_default_environment()

仅包含基础变量（PATH、HOME等），不包含

os.environ

。我们修复了

src/runtime/mcp_client.py

以传递完整环境变量：

python

undefined

In _connect_stdio method:

在_connect_stdio方法中：

full_env = {**os.environ, **(resolved_env or {})}


This ensures API keys from `~/.claude/.env` reach subprocesses.

full_env = {**os.environ, **(resolved_env or {})}


这确保`~/.claude/.env`中的API密钥能传递给子进程。

Graceful Degradation Pattern

优雅降级模式

Each tool is optional. If unavailable (disabled, no API key, etc.), the pipeline continues:

python

async def check_tool_available(tool_id: str) -> bool:
    """Check if an MCP tool is available."""
    server_name = tool_id.split("__")[0]
    server_config = manager._config.get_server(server_name)
    if not server_config or server_config.disabled:
        return False
    return True

每个工具都是可选的。如果工具不可用（已禁用、无API密钥等），工作流管道会继续执行：

python

async def check_tool_available(tool_id: str) -> bool:
    """检查MCP工具是否可用。"""
    server_name = tool_id.split("__")[0]
    server_config = manager._config.get_server(server_name)
    if not server_config or server_config.disabled:
        return False
    return True

In step function:

在步骤函数中：

if not await check_tool_available("nia__search"): return StepResult(status=StepStatus.SKIPPED, message="Nia not available")

undefined

if not await check_tool_available("nia__search"): return StepResult(status=StepStatus.SKIPPED, message="Nia不可用")

undefined

Tool Name Reference

工具名称参考

nia (Documentation Search)

nia（文档搜索）

nia__search              - Universal documentation search
nia__nia_research        - Research with sources
nia__nia_grep            - Grep-style doc search
nia__nia_explore         - Explore package structure

nia__search              - 通用文档搜索
nia__nia_research        - 带来源的研究
nia__nia_grep            - Grep风格的文档搜索
nia__nia_explore         - 探索包结构

ast-grep (Structural Code Search)

ast-grep（结构化代码搜索）

ast-grep__find_code      - Find code by AST pattern
ast-grep__find_code_by_rule - Find by YAML rule
ast-grep__scan_code      - Scan with multiple patterns

ast-grep__find_code      - 按AST模式查找代码
ast-grep__find_code_by_rule - 按YAML规则查找
ast-grep__scan_code      - 多模式扫描

morph (Fast Text Search + Edit)

morph（快速文本搜索+编辑）

morph__warpgrep_codebase_search  - 20x faster grep
morph__edit_file                 - Smart file editing

morph__warpgrep_codebase_search  - 快20倍的grep
morph__edit_file                 - 智能文件编辑

qlty (Code Quality)

qlty（代码质量）

qlty__qlty_check         - Run quality checks
qlty__qlty_fmt           - Auto-format code
qlty__qlty_metrics       - Get code metrics
qlty__smells             - Detect code smells

qlty__qlty_check         - 运行质量检查
qlty__qlty_fmt           - 自动格式化代码
qlty__qlty_metrics       - 获取代码指标
qlty__smells             - 检测代码异味

git (Version Control)

git（版本控制）

git__git_status          - Get repo status
git__git_diff            - Show differences
git__git_log             - View commit history
git__git_add             - Stage files

git__git_status          - 获取仓库状态
git__git_diff            - 显示差异
git__git_log             - 查看提交历史
git__git_add             - 暂存文件

Pipeline Architecture

管道架构

                    +----------------+
                    |   CLI Args     |
                    | (topic, dir)   |
                    +-------+--------+
                            |
                    +-------v--------+
                    | PipelineContext|
                    | (shared state) |
                    +-------+--------+
                            |
    +-------+-------+-------+-------+-------+
    |       |       |       |       |       |
+---v---+---v---+---v---+---v---+---v---+
| nia   |ast-grp| morph | qlty  | git   |
|search |pattern|search |check  |status |
+---+---+---+---+---+---+---+---+---+---+
    |       |       |       |       |
    +-------v-------v-------v-------+
                    |
            +-------v--------+
            | StepResult[]   |
            | (aggregated)   |
            +----------------+

                    +----------------+
                    |   CLI 参数     |
                    | (主题, 目录)   |
                    +-------+--------+
                            |
                    +-------v--------+
                    | PipelineContext|
                    | (共享状态)     |
                    +-------+--------+
                            |
    +-------+-------+-------+-------+-------+
    |       |       |       |       |       |
+---v---+---v---+---v---+---v---+---v---+
| nia   |ast-grp| morph | qlty  | git   |
|search |pattern|search |check  |status |
+---+---+---+---+---+---+---+---+---+---+
    |       |       |       |       |
    +-------v-------v-------v-------+
                    |
            +-------v--------+
            | StepResult[]   |
            | (聚合结果)      |
            +----------------+

Error Handling

错误处理

The pipeline captures errors without failing the entire run:

python

try:
    result = await call_mcp_tool("nia__search", {"query": topic})
    return StepResult(status=StepStatus.SUCCESS, data=result)
except Exception as e:
    ctx.errors.append(f"nia: {e}")
    return StepResult(status=StepStatus.FAILED, error=str(e))

管道会捕获错误而不会导致整个运行失败：

python

try:
    result = await call_mcp_tool("nia__search", {"query": topic})
    return StepResult(status=StepStatus.SUCCESS, data=result)
except Exception as e:
    ctx.errors.append(f"nia: {e}")
    return StepResult(status=StepStatus.FAILED, error=str(e))

Creating Your Own Pipeline

创建你自己的管道

Copy the pattern from
```
scripts/research_implement_pipeline.py
```
Define your steps as async functions
Use
```
check_tool_available()
```
for graceful degradation
Chain results through
```
PipelineContext
```
Aggregate with
```
print_summary()
```

从
```
scripts/research_implement_pipeline.py
```
复制模式
将步骤定义为异步函数
使用
```
check_tool_available()
```
实现优雅降级
通过
```
PipelineContext
```
传递结果
使用
```
print_summary()
```
聚合结果