mcp-chaining

MCP Chaining Pipeline

エンドツーエンドのワークフローを実現する、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	コードベースを高速検索する
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 - テスト用サンプルコード

使用例

# 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

重要な修正：環境変数

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

ツール名リファレンス

nia（ドキュメント検索）

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（構造的なコード検索）

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（高速テキスト検索と編集）

morph__warpgrep_codebase_search  - 20x faster grep
morph__edit_file                 - Smart file editing

qlty（コード品質）

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

エラー処理

パイプライン全体を失敗させることなく、エラーを捕捉します。

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

独自のパイプラインを作成する

1. scripts/research_implement_pipeline.py のパターンをコピーします
2. 各手順を非同期関数として定義します
3. 段階的な機能縮退には check_tool_available() を使用します
4. PipelineContext を通じて結果を連鎖させます
5. print_summary() で結果を集計します