mcp-chaining

用途

MCP Chaining 是一个“研究到实现”（research-to-implement）的流水线，将 5 个 MCP 工具串联成端到端工作流。它主要用于：构建多工具 MCP 流水线；理解如何以优雅降级的方式串联 MCP 调用；调试 MCP 环境变量问题；学习不同 MCP 服务器的工具命名约定。

主要能力

流水线按顺序串联以下工具：

nia（nia__search）：搜索库文档
ast-grep（ast-grep__find_code）：查找 AST 代码模式
morph（morph__warpgrep_codebase_search）：快速代码库搜索
qlty（qlty__qlty_check）：代码质量校验
git（git__git_status）：Git 操作

各服务器提供的工具包括：

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（暂存文件）

使用方式

关键文件：

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

创建自己的流水线：

复制 scripts/research_implement_pipeline.py 中的模式
将步骤定义为 async 函数
使用 check_tool_available() 实现优雅降级
通过 PipelineContext 串联结果
使用 print_summary() 聚合输出

关键修复：环境变量

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

流水线架构

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

适用场景

构建多工具 MCP 流水线
理解如何以优雅降级方式串联 MCP 调用
调试 MCP 环境变量问题
学习不同 MCP 服务器的工具命名约定

限制

资料未提供关于收费、价格、平台支持范围、性能指标或具体效果保证的信息。