はじめに

MCPをAIエージェントに接続するとは、モデルの外部にある機能をエージェントが制御された方法で発見し、利用できるようにすることです。モデルがデータベースクエリを直接実行したり、非公開ファイルを読み取ったり、業務APIを呼び出したりするわけではありません。代わりに、エージェント内のMCPクライアントがMCPサーバーに接続してツール定義を取得し、それらを利用可能な機能としてモデルに渡し、モデルが選択したツール呼び出しを実行します。

この分離によって、重要な境界が生まれます。モデルはユーザーのリクエストを解釈し、ツール名と構造化された引数を提案する役割を担います。MCPクライアントは、プロトコルセッションの維持、選択されたツールの呼び出し、結果の返却、エラーの報告を担当します。そのため、信頼性の高いワークフローは、発見、選択、呼び出しという3つの明確な段階で構成されます。同じパターンは、モデルが商用API、自社ホスト型モデル、社内エージェントフレームワークのいずれで提供されていても利用できます。

MCPを導入しても、アプリケーションレベルのセキュリティが不要になるわけではありません。エージェントは信頼できるサーバーにのみ接続し、必要に応じて認証を適用し、ユーザー権限を検証するとともに、危険な操作を自動的に承認しないようにする必要があります。ツールの説明はモデルによる機能選択に役立ちますが、セキュリティポリシーの代わりにはなりません。

エージェントのワークフロー

起動時に、MCPクライアントはサーバーへのトランスポート接続を確立し、MCPセッションを初期化します。続いて、クライアントはプロトコルのツール発見操作を呼び出します。応答には、ツール名、説明、入力スキーマが含まれます。これらのスキーマは各ツールが受け付ける引数を定義しているため、エージェントは曖昧な文章ではなく、構造化された情報をモデルに提供できます。

モデルは、発見した定義とユーザーのリクエストを組み合わせて使用します。たとえば、サーバーがsearch_documentsとcreate_ticketというツールを公開している場合、インシデント報告書を探すリクエストに対しては、通常、create_ticketではなくsearch_documentsの呼び出しが生成されます。モデルの出力は、名前と引数を含むツール呼び出しの提案として扱う必要があります。実行する前に、ホストアプリケーションがその名前が発見済みであること、および引数がローカルポリシーを満たすことを確認しなければなりません。

検証後、MCPクライアントは選択したツールを呼び出し、構造化コンテンツ、テキスト、またはエラーを受け取ります。ホストアプリケーションは、その結果をモデルに返し、最終的な自然言語の応答を生成させることができます。複数ステップのエージェントでは、最初のツールの結果から2つ目のツールに必要な情報が得られた場合、このサイクルを繰り返します。制御不能なループを防ぐため、アプリケーションではステップ数、実行時間、機密性の高い操作に制限を設ける必要があります。

最小限のMCPクライアント

次のPythonプログラムは、Streamable HTTPを使用したディスカバリー、選択、呼び出しの流れを示します。クライアントセッションとトランスポートには、公式のMCP Pythonクライアントインターフェースを使用しています。エージェントを実行する環境にMCP Pythonパッケージをインストールし、URLを、すでに起動してこのトランスポートを受け入れるよう設定されたMCPサーバーのものに置き換えてください。

import argparse
import asyncio
import json

from mcp import ClientSession
from mcp.client.streamable_http import streamable_http_client


async def run(url: str, tool_name: str, arguments: dict) -> None:
    async with streamable_http_client(url) as streams:
        read_stream, write_stream, _ = streams

        async with ClientSession(read_stream, write_stream) as session:
            await session.initialize()
            discovered = await session.list_tools()

            print("Discovered tools:")
            for tool in discovered.tools:
                description = tool.description or "No description"
                print(f"- {tool.name}: {description}")

            selected = next(
                (tool for tool in discovered.tools if tool.name == tool_name),
                None,
            )
            if selected is None:
                available = ", ".join(tool.name for tool in discovered.tools)
                raise ValueError(
                    f"Unknown tool '{tool_name}'. Available tools: {available}"
                )

            print(f"Invoking {selected.name} with {json.dumps(arguments)}")
            result = await session.call_tool(
                selected.name,
                arguments=arguments,
            )
            print("Tool result:")
            if hasattr(result, "model_dump_json"):
                print(result.model_dump_json(indent=2))
            else:
                print(result)


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("url", help="MCP Streamable HTTP endpoint")
    parser.add_argument("tool", help="Discovered tool name to invoke")
    parser.add_argument(
        "--arguments",
        default="{}",
        help="JSON object containing tool arguments",
    )
    args = parser.parse_args()

    try:
        arguments = json.loads(args.arguments)
    except json.JSONDecodeError as error:
        raise SystemExit(f"Arguments must be valid JSON: {error}") from error

    if not isinstance(arguments, dict):
        raise SystemExit("Arguments must be a JSON object")

    try:
        asyncio.run(run(args.url, args.tool, arguments))
    except Exception as error:
        raise SystemExit(f"MCP request failed: {error}") from error


if __name__ == "__main__":
    main()

入力は、MCPエンドポイント、ツール名、引数のJSONオブジェクトで構成されます。プログラムはまず、ディスカバリーで返されたツールを表示し、次に要求された名前が実際に存在することを確認します。この確認は重要です。モデルがツール名を創作したり、以前のセッションで使われた古い名前を保持していたりする可能性があるためです。出力にはサーバーの結果が含まれ、周囲のエージェントはそれをモデルに返して解釈させることができます。

ディスカバリーとモデル選択の接続

本番環境のエージェントでは、ディスカバリーと呼び出しの間にモデルアダプターを配置します。list_toolsの戻り後、ホストは各ツールの名前、説明、入力スキーマをモデルプロバイダーのツール呼び出し形式に変換します。正確なリクエスト形式はモデルAPIによって異なるため、MCPクライアントはすべてのプロバイダーが同じメッセージ構造を使用すると想定してはいけません。エージェントはモデルの応答を受け取り、提案された名前とJSON引数を抽出したうえで、session.call_toolを呼び出す前にローカルで検証します。

ホストは、現在のディスカバリー応答に名前が存在しない場合、引数がオブジェクトではない場合、必須フィールドが欠けている場合、または値がアプリケーションポリシーに違反している場合、ツール呼び出しを拒否する必要があります。スキーマ検証は有用ですが、認可チェックは別途実施しなければなりません。たとえば、形式的に有効なdelete_record呼び出しであっても、読み取り専用ユーザーには許可できない場合があります。取り消し不能な操作、金融処理、外部メッセージ、本番データの変更については、確認を求めることも適切です。

呼び出し後、結果をツール結果メッセージとしてモデルとの会話に追加できます。その後、モデルは結果を要約したり、別の許可されたツールが必要かどうかを判断したりできます。ログ記録用に生の結果を保持しつつ、サードパーティのモデルにコンテンツを送信する前に、認証情報、個人データ、内部診断情報をマスキングしてください。

よくある失敗と信頼性を高める実践方法

接続に失敗する場合、通常はエンドポイントが間違っている、サーバーが利用できない、トランスポートが一致していない、または認証情報が不足していることを意味します。サーバーが選択したトランスポートをサポートしていること、またネットワーク、プロキシ、TLSの設定が接続を許可していることを確認してください。不明なプロトコルへ暗黙的にフォールバックしてトランスポートの問題を解決しようとしてはいけません。設定されたトランスポートを明示し、元のエラーを報告してください。

不明なツールエラーは、エージェントが現在のディスカバリー手順で返されなかった名前を呼び出したときに発生します。セッションやサーバーの機能が変わる可能性がある場合はディスカバリーを更新し、要求された名前を必ず現在のツールセットと比較してください。引数が無効になる一般的な原因は、必須プロパティの欠落、型の誤り、またはモデルがオブジェクトではなく文字列としてエンコードされたJSONを返すことです。呼び出し前に引数を検証して正規化し、デバッグに役立つサーバーのエラーメッセージは保持してください。

もう1つのよくある誤りは、接続されているすべてのサーバーのツールを、すべてのユーザーに公開してしまうことです。ツール名や説明が重複することがあり、悪意のあるサーバーや設定の不備があるサーバーが、安全でない機能を提示する可能性もあります。明示的なサーバー許可リストを使用し、認証情報を分離し、ツールごとの認可を適用し、タイムアウトを設定し、繰り返し呼び出しの回数を制限してください。また、どのユーザー、モデルの判断、サーバー、ツールが関与したかを記録します。これらの制御により、エージェントの動作を可視化し、レビューしやすくなります。

まとめ

MCP対応エージェントは、2つのコンポーネントが連携するホストアプリケーションとして理解するのが最も適切です。モデルはユーザーの意図を推論して構造化された呼び出しを提案し、MCPクライアントは機能を検出して、プロトコルに準拠した実行を行います。モデルが言及したからといってツールが存在するとクライアントが決めつけてはなりません。現在利用可能なツールを検出し、選択内容と引数を検証したうえで、呼び出し前にローカルポリシーを適用する必要があります。

実際のワークフローは、初期化、ツール一覧の取得、スキーマのモデルへの提示、モデルが提案した呼び出しの検証、選択したツールの呼び出し、そして最終推論のための結果の返却です。このループが1つのサーバーで機能すれば、コアとなる境界を変更することなく、複数サーバーや複数ステップのタスクへ拡張できます。最も重要な実装上の選択は、明示的なトランスポート設定、厳格な検証、安全な認可、実行範囲の制限、そして有用なエラー報告です。