mcp-chaining by parcadei/continuous-claude-v3
npx skills add https://github.com/parcadei/continuous-claude-v3 --skill mcp-chaining一个从研究到实现的管道,串联 5 个 MCP 工具以完成端到端工作流。
一个串联以下工具的管道:
| 步骤 | 服务器 | 工具 ID | 用途 |
|---|---|---|---|
| 1 | nia | nia__search | 搜索库文档 |
| 2 | ast-grep | ast-grep__find_code | 查找 AST 代码模式 |
| 3 | morph | morph__warpgrep_codebase_search |
广告位招租
在这里展示您的产品或服务
触达数万 AI 开发者,精准高效
| 快速代码库搜索 |
| 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 "async error handling 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 not available")
nia__search - 通用文档搜索
nia__nia_research - 带来源的研究
nia__nia_grep - Grep 风格文档搜索
nia__nia_explore - 探索包结构
ast-grep__find_code - 通过 AST 模式查找代码
ast-grep__find_code_by_rule - 通过 YAML 规则查找
ast-grep__scan_code - 使用多个模式扫描
morph__warpgrep_codebase_search - 快 20 倍的 grep
morph__edit_file - 智能文件编辑
qlty__qlty_check - 运行质量检查
qlty__qlty_fmt - 自动格式化代码
qlty__qlty_metrics - 获取代码指标
qlty__smells - 检测代码异味
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() 进行聚合每周安装量
196
仓库
GitHub 星标
3.6K
首次出现
2026 年 1 月 22 日
安全审计
安装于
opencode188
codex186
gemini-cli184
cursor183
github-copilot181
amp178
A research-to-implement pipeline that chains 5 MCP tools for end-to-end workflows.
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 |
scripts/research_implement_pipeline.py - Main pipeline implementationscripts/test_research_pipeline.py - Test harness with isolated sandboxworkspace/pipeline-test/sample_code.py - Test sample code# 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
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.
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")
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__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__warpgrep_codebase_search - 20x faster grep
morph__edit_file - Smart file editing
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_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) |
+----------------+
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))
scripts/research_implement_pipeline.pycheck_tool_available() for graceful degradationPipelineContextprint_summary()Weekly Installs
196
Repository
GitHub Stars
3.6K
First Seen
Jan 22, 2026
Security Audits
Gen Agent Trust HubPassSocketPassSnykPass
Installed on
opencode188
codex186
gemini-cli184
cursor183
github-copilot181
amp178
通过 LiteLLM 代理让 Claude Code 对接 GitHub Copilot 运行 | 高级变通方案指南
31,600 周安装