mcp-chaining

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