name: mcp-chaining description: 研究到实现管道，链式连接5个MCP工具，具有优雅降级功能 allowed-tools: [Bash, Read] user-invocable: false

MCP链式管道

一个研究到实现的管道，链式连接5个MCP工具，用于端到端工作流。

何时使用

构建多工具MCP管道
理解如何链式调用MCP工具并实现优雅降级
调试MCP环境变量问题
学习不同MCP服务器的工具命名约定

我们构建了什么

一个链式连接这些工具的管道：

步骤	服务器	工具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操作

关键文件

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

使用示例

# 干运行管道（预览计划而不做更改）
uv run python -m runtime.harness scripts/research_implement_pipeline.py \
    --topic "异步错误处理 python" \
    --target-dir "./workspace/pipeline-test" \
    --dry-run --verbose

# 运行测试
uv run python -m runtime.harness scripts/test_research_pipeline.py --test all

# 查看管道脚本
cat scripts/research_implement_pipeline.py

关键修复：环境变量

MCP SDK的get_default_environment()只包含基本变量（PATH、HOME等），不包括os.environ。我们修复了src/runtime/mcp_client.py以传递完整环境：

# 在 _connect_stdio 方法中：
full_env = {**os.environ, **(resolved_env or {})}

这确保来自~/.claude/.env的API密钥可以到达子进程。

优雅降级模式

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

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

# 在步骤函数中：
if not await check_tool_available("nia__search"):
    return StepResult(status=StepStatus.SKIPPED, message="Nia 不可用")

工具名称参考

nia（文档搜索）

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

ast-grep（结构代码搜索）

ast-grep__find_code      - 通过AST模式查找代码
ast-grep__find_code_by_rule - 通过YAML规则查找
ast-grep__scan_code      - 使用多个模式扫描

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

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

qlty（代码质量）

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

git（版本控制）

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

管道架构

                    +----------------+
                    |   CLI参数      |
                    | (主题, 目录)   |
                    +-------+--------+
                            |
                    +-------v--------+
                    | PipelineContext|
                    | (共享状态)     |
                    +-------+--------+
                            |
    +-------+-------+-------+-------+-------+
    |       |       |       |       |       |
+---v---+---v---+---v---+---v---+---v---+
| nia   |ast-grp| morph | qlty  | git   |
|搜索   |模式   |搜索   |检查   |状态   |
+---+---+---+---+---+---+---+---+---+---+
    |       |       |       |       |
    +-------v-------v-------v-------+
                    |
            +-------v--------+
            | StepResult[]   |
            | (聚合)         |
            +----------------+

错误处理

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

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

创建您自己的管道

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