MCP链式管道：5个MCP工具串联实现端到端研发工作流，支持优雅降级

mcp-chaining by parcadei/continuous-claude-v3

196 周安装量

3,600 GitHub Stars

GitHub

安装命令

npx skills add https://github.com/parcadei/continuous-claude-v3 --skill 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`

广告位招租

在这里展示您的产品或服务

触达数万 AI 开发者，精准高效

联系我们

关键修复：环境变量

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

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--------+
                    | 管道上下文     |
                    | (共享状态)     |
                    +-------+--------+
                            |
    +-------+-------+-------+-------+-------+
    |       |       |       |       |       |
+---v---+---v---+---v---+---v---+---v---+
| nia   |ast-grp| morph | qlty  | git   |
|搜索   |模式   |搜索   |检查   |状态   |
+---+---+---+---+---+---+---+---+---+---+
    |       |       |       |       |
    +-------v-------v-------v-------+
                    |
            +-------v--------+
            | 步骤结果[]     |
            | (聚合)         |
            +----------------+

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

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() 进行聚合

2026 年 1 月 22 日

🇺🇸English

MCP Chaining Pipeline

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

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

What We Built

A pipeline that chains these tools:

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

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

Usage Examples

# 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

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:

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

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

Graceful Degradation Pattern

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

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

Tool Name Reference

nia (Documentation Search)

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 (Structural Code Search)

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 (Fast Text Search + Edit)

morph__warpgrep_codebase_search  - 20x faster grep
morph__edit_file                 - Smart file editing

qlty (Code Quality)

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

git (Version Control)

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

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

Error Handling

The pipeline captures errors without failing the entire run:

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

Weekly Installs

196

Repository

parcadei/contin…laude-v3

GitHub Stars

3.6K

First Seen

Jan 22, 2026

Security Audits

Gen Agent Trust HubPass SocketPass SnykPass

Installed on

opencode188

codex186

gemini-cli184

cursor183

github-copilot181

amp178

通过 LiteLLM 代理让 Claude Code 对接 GitHub Copilot 运行 | 高级变通方案指南

31,600 周安装