mcp-chaining

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 - 测试示例代码

使用示例

# 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

关键修复：环境变量

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

# In _connect_stdio method:
full_env = {**os.environ, **(resolved_env or {})}

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

优雅降级模式

每个工具都是可选的。如果工具不可用（已禁用、没有 API 密钥等），流水线会继续运行：

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

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

工具名称参考

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

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

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

morph__warpgrep_codebase_search  - 20x faster grep
morph__edit_file                 - Smart file editing

qlty（代码质量）

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

git（版本控制）

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

流水线架构

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

错误处理

流水线会捕获错误，但不会导致整个运行失败：

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

创建自己的流水线

1. 复制 scripts/research_implement_pipeline.py 中的模式
2. 将各个步骤定义为异步函数
3. 使用 check_tool_available() 实现优雅降级
4. 通过 PipelineContext 传递结果
5. 使用 print_summary() 汇总结果