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