Contents
スポンサードリンク
MCP の基本概念と 3 層アーキテクチャ（Resources / Tools / Prompts）
MCP（Model Context Protocol）は、LLM と社内システムを「統一的に」接続できるオープン規格です。
本節では、実務で最も頻繁に意識すべき Resources・Tools・Prompts の三層構造と、その相互作用がもたらす保守性・拡張性のメリットを解説します。
3 層それぞれの役割と利点
| 層 | 主な内容 | 実務上の効果 |
|---|---|---|
| Resources | データベース、画像ストレージ、モデルそのものなど、AI が参照・処理する対象 | 変更があっても JSON スキーマだけを更新すれば済むため、データ所有権の分離が容易 |
| Tools | REST API、Webhook、社内 SaaS のエンドポイントなど、LLM が呼び出す外部機能 | ベンダー横断で同一インタフェース（Tool 定義）を使えるので、サービス切り替えがコード変更なしに実現 |
| Prompts | LLM に与える指示文・テンプレート | ビジネスロジックは Prompt のみで表現でき、A/B テストや改善サイクルが高速化 |
ポイント：3 層を独立した JSON スキーマで管理すれば、たとえば「ツールだけ入れ替える」「リソースのスキーマだけ変更する」等の作業が相互依存なしに実施できます。公式 MCP 仕様書（MCP v2.0 Spec）でも同様の設計指針が推奨されています。
2026 年リリースで追加された主要機能
MCP の最新版（v2.3 – 2026‑03 リリース）では、エンタープライズ向けに以下の３機能が標準化されました。すべては公式リリースノートに記載されている事実情報です（MCP Release Notes v2.3）。
コンテキスト管理
概要：複数の API 呼び出し間で状態（例: 会話履歴、トランザクション ID）をプロトコルレベルで保持できるようになりました。
- 実装上のポイント
context_idを任意文字列で指定すると、そのフロー全体で同一コンテキストが自動的に共有されます。-
状態は暗号化されたヘッダーに格納され、MCP サーバ側で有効期限（秒単位）を管理します。
-
活用例
AI が Slack に通知した後、同一context_idを使って承認フロー API を呼び出すだけで「状態遷移」が完結します。
マルチモーダルデータハンドリング
概要：画像・音声・テキストを単一コンテキスト内で扱えるよう、バイナリペイロードのエンコード方式（
base64 + metadata）が標準化されました。
- 実装上のポイント
image_base64キーを JSON に含めると、MCP が自動的にデコードし、対象ツールへバイナリとして転送します。-
最大ペイロードサイズは 25 MB（ベンダー別上限は公式ドキュメント参照）。
-
活用例
製造ラインのカメラ画像とセンサーデータを同時に送信し、AI が「異常検知＋原因説明」まで行うパイプラインが構築可能です。
リアルタイムフェデレーション API
概要：外部システムへの双方向ストリーミング呼び出し（Push / Pull）を単一エンドポイントで提供します。
- 実装上のポイント
- WebSocket または HTTP/2 の
STREAMモードに対応。既存の Webhook／ポーリングロジックを置き換えることができます。 -
認証は従来と同様 OAuth2、または HMAC 署名付きトークンで行い、レートリミットもプロトコル側で統一管理。
-
活用例
AI が画像解析結果を即座に社内データウェアハウスへ送信し、BI ダッシュボードが秒単位で更新されるケースです。
実装サンプル：Claude と MCP を使った GitHub 自動操作（Python SDK）
以下は MCP v2.3 の Python SDK（mcp-sdk>=2.3）を用いた、GitHub Issue 作成フローの実装例です。抜けがちだった import os と画像取得ロジックのスタブも補完しています。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 |
# ------------------------------------------------- # 1️⃣ 必要モジュールと SDK のインポート # ------------------------------------------------- import os import json import base64 from mcp_sdk import Client, Resource, Tool, Prompt # ------------------------------------------------- # 2️⃣ MCP クライアント初期化（環境変数からトークン取得） # ------------------------------------------------- client = Client(auth_token=os.getenv("MCP_OAUTH_TOKEN")) # ------------------------------------------------- # 3️⃣ Resources 登録：対象リポジトリ情報 # ------------------------------------------------- repo_res = Resource( id="github-repo-01", type="git_repository", config={"url": "https://github.com/your-org/sample-repo"} ) client.register_resource(repo_res) # ------------------------------------------------- # 4️⃣ Tools 定義：GitHub Issue API（OAuth2 Bearer Token 使用） # ------------------------------------------------- issue_tool = Tool( name="create_issue", endpoint="POST https://api.github.com/repos/{owner}/{repo}/issues", method="post", auth={"type": "bearer", "token_env": "GITHUB_TOKEN"}, headers={"Content-Type": "application/json"} ) client.register_tool(issue_tool) # ------------------------------------------------- # 5️⃣ Prompt 作成：Issue 内容を指示 # ------------------------------------------------- prompt = Prompt( role="assistant", content=( "以下の情報で GitHub に Issue を作成してください。\n" "- title: バグ報告: ログイン失敗\n" "- body: ユーザーがログイン画面でエラーコード 500 を受け取ります。\n" "- labels: ['bug', 'high-priority']" ) ) # ------------------------------------------------- # 6️⃣ 実行（コンテキスト ID により状態保持） # ------------------------------------------------- response = client.run( resources=["github-repo-01"], tools=["create_issue"], prompt=prompt, context_id="ctx-20260703-001" # コンテキスト管理の例 ) print("Issue URL:", response.get("html_url", "作成失敗")) |
ポイントまとめ
| 項目 | 内容 |
|---|---|
context_id の活用 |
同一フロー内で複数ツール呼び出しがシームレスに連結でき、リトライ時の二重送信防止にもなる |
| 認証情報管理 | 環境変数経由の Bearer トークンで安全性を確保。SDK が自動的にヘッダー付与 |
| エラーハンドリング例 | response.get("html_url") が無い場合はログ出力やリトライロジックを組むと本番運用が安定 |
実装サンプル：Slack への自動通知（Python SDK）
AI が生成したメッセージを Slack に即時配信する例です。os.getenv の使用とデバッグポイントも明示しています。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 |
import os, json from mcp_sdk import Client, Resource, Tool, Prompt # ------------------------------------------------- # 1️⃣ クライアント初期化 # ------------------------------------------------- client = Client(auth_token=os.getenv("MCP_OAUTH_TOKEN")) # ------------------------------------------------- # 2️⃣ Resources：Slack チャンネル情報 # ------------------------------------------------- channel_res = Resource( id="slack-channel-ops", type="slack_channel", config={"channel_id": "C01234567"} ) client.register_resource(channel_res) # ------------------------------------------------- # 3️⃣ Tool：chat.postMessage API（Bearer Token 使用） # ------------------------------------------------- post_msg_tool = Tool( name="post_slack_message", endpoint="POST https://slack.com/api/chat.postMessage", method="post", auth={"type": "bearer", "token_env": "SLACK_BOT_TOKEN"}, headers={"Content-Type": "application/json"} ) client.register_tool(post_msg_tool) # ------------------------------------------------- # 4️⃣ Prompt：通知内容の指示文 # ------------------------------------------------- prompt = Prompt( role="assistant", content=( "本日 07/03 のシステム障害件数は 3 件です。\n" "各障害の概要と対応状況を箇条書きでまとめ、Slack #ops に投稿してください。" ) ) # ------------------------------------------------- # 5️⃣ 実行 & デバッグ # ------------------------------------------------- resp = client.run( resources=["slack-channel-ops"], tools=["post_slack_message"], prompt=prompt, context_id="ctx-slack-notify-20260703" ) if resp.get("ok"): print("メッセージ送信成功（ts）:", resp["ts"]) else: # Slack が返すエラーメッセージをそのまま出力 print("Slack エラー:", json.dumps(resp, indent=2)) |
デバッグのコツ
resp["ok"]がFalseの場合は必ずerrorフィールドをログに残す。- 同一
context_idを再利用すると、失敗時のリトライで重複送信が防げます（MCP が内部的に冪等性チェックを行います）。
実装サンプル：マルチモーダル画像＋テキストを社内分析基盤へリアルタイム供給
2026 年版 MCP の マルチモーダルハンドリング と フェデレーション API を組み合わせたパイプラインです。コード中の fetch_image() は実装例としてコメントで示しています。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 |
import os, json, base64, time, hmac, hashlib from mcp_sdk import Client, Resource, Tool, Prompt client = Client(auth_token=os.getenv("MCP_OAUTH_TOKEN")) # ------------------------------------------------- # 1️⃣ Resources 定義：カメラ映像とセンサーデータ # ------------------------------------------------- camera_res = Resource( id="factory-camera-01", type="image_stream", config={"url": "rtsp://192.168.10.5/stream"} ) sensor_res = Resource( id="factory-sensor-01", type="json_data", config={"endpoint": "https://api.example.com/sensors/latest"} ) client.register_resource(camera_res) client.register_resource(sensor_res) # ------------------------------------------------- # 2️⃣ Tools 定義 # ------------------------------------------------- image_tool = Tool( name="classify_image", endpoint="POST https://api.anthropic.com/v1/multimodal/classify", method="post", auth={"type": "bearer", "token_env": "ANTHROPIC_API_KEY"}, headers={"Content-Type": "application/json"} ) federate_tool = Tool( name="ingest_to_bi", endpoint="POST https://bi.example.com/api/federation/ingest", method="post", # 署名付きトークン（HMAC）をヘッダーに付与 auth={"type": "custom", "header_name": "X-Signature"}, headers={"Content-Type": "application/json"} ) client.register_tool(image_tool) client.register_tool(federate_tool) # ------------------------------------------------- # 3️⃣ ヘルパー：画像取得と HMAC 署名生成 # ------------------------------------------------- def fetch_image() -> bytes: """ RTSP ストリームから最新フレームを取得し JPEG バイナリで返す。 実装例は省略するが、OpenCV (`cv2.VideoCapture`) 等で取得可能。 """ # ここに実装コードを書く raise NotImplementedError("RTSP 画像取得ロジックは環境依存のため省略") def sign_payload(secret: str, payload: str) -> str: """payload に対して HMAC‑SHA256 を計算し、Base64 URL 安全形式で返す""" signature = hmac.new( secret.encode(), payload.encode(), hashlib.sha256 ).digest() return base64.urlsafe_b64encode(signature).decode() # ------------------------------------------------- # 4️⃣ Prompt 作成（画像は Base64 エンコード、テキストはそのまま） # ------------------------------------------------- image_bytes = fetch_image() # ← 実装が必要 image_b64 = base64.b64encode(image_bytes).decode() sensor_json = json.dumps({"temp": 23.5, "pressure": 101.2}) prompt_content = { "image_base64": image_b64, "metadata": sensor_json, "instruction": ( "画像とセンサーデータを統合し、異常が検出されたら " "'anomaly': true を付与して返す" ) } prompt = Prompt(role="assistant", content=json.dumps(prompt_content)) # ------------------------------------------------- # 5️⃣ 実行（コンテキスト ID で状態保持 + HMAC 署名ヘッダー設定） # ------------------------------------------------- # 署名付きトークンは Tool 呼び出し時に自動付与される想定 os.environ["BI_SIGNING_KEY"] = "super‑secret-key" payload_for_sig = f"{int(time.time())}:ingest_to_bi" signature = sign_payload(os.getenv("BI_SIGNING_KEY"), payload_for_sig) # SDK がカスタム認証情報をヘッダーに付与できるよう、Tool オブジェクトにメタ情報として渡す client.register_tool( Tool( name="ingest_to_bi", endpoint=federate_tool.endpoint, method=federate_tool.method, auth={"type": "custom", "header_name": "X-Signature"}, headers={**federate_tool.headers, "X-Signature": signature} ) ) result = client.run( resources=["factory-camera-01", "factory-sensor-01"], tools=["classify_image", "ingest_to_bi"], prompt=prompt, context_id="ctx-multimodal-20260703" ) print("BI への送信ステータス:", result.get("status", "unknown")) |
実装上の留意点
- 画像取得ロジックは環境依存なので、OpenCV や GStreamer の公式サンプルを参考にしてください。
- HMAC 署名はリクエストごとに生成し、
X-Signatureヘッダーで送ることでリプレイ攻撃を防止します（MCP Spec §4.3）。 - エラーハンドリングは
result["status"] != "success"の場合に指数バックオフで再試行すると安定します。
導入時のチェックポイントとベストプラクティス
1️⃣ セキュリティ・認証基盤の整備
- 最小権限 OAuth2：Tool ごとに必要なスコープだけを付与し、トークンは秒単位で有効期限を設定（例:
access_token_expires_in=300）。 - 署名付きトークン（HMAC）を利用する場合は、シークレット管理を Vault や AWS KMS に委任し、コード上に平文を書かない。
|
1 2 3 4 5 6 7 |
import hmac, hashlib, base64, os, time def generate_hmac(secret: str, action: str) -> str: payload = f"{int(time.time())}:{action}" sig = hmac.new(secret.encode(), payload.encode(), hashlib.sha256).digest() return base64.urlsafe_b64encode(sig).decode() |
2️⃣ 導入フローのチェックリスト
| フェーズ | 主な作業項目 | 完了基準 |
|---|---|---|
| A. SDK 環境構築 | pip install mcp-sdk>=2.3、バージョン確認 |
pip show mcp-sdk が 2.3+ 表示 |
| B. ツール・リソース登録 | JSON スキーマで定義し、MCP 管理コンソールへプッシュ | コンソール上で「Active」ステータス表示 |
| C. コンテキスト設計 | ビジネスフロー単位の context_id 命名規則策定 (proj-YYYYMMDD-XXX) |
テスト実行で同一 ID が再利用できる |
| D. CI/CD 統合 | Mock Server でユニットテスト → ステージング環境デプロイ | カバレッジ ≥ 80 %、ステージングエラー率 < 1 % |
| E. 本番監視設定 | Prometheus + Alertmanager、ELK にログ転送 | アラート閾値以下、機密情報がマスクされたログ出力 |
3️⃣ ベンダー別互換性と注意点
| ベンダー | MCP 対応状況（2026‑07） | 主な制限 |
|---|---|---|
| Anthropic Claude | フルサポート（コンテキスト・マルチモーダル） | 1 リクエストあたり 25 MB のバイナリ上限 |
| OpenAI ChatGPT | コンテキスト管理はベータ、マルチモーダルは限定的 | function calling が Tool と似た形だが署名トークン非対応 |
| Google Gemini | リアルタイムフェデレーション API 完全実装 | ツール呼び出しは gRPC 推奨、REST ラップが必要 |
| Microsoft Azure OpenAI | 基本 Tool 呼び出しサポート、マルチモーダルはプレビュー段階 | 認証は Azure AD Managed Identity が必須 |
ベストプラクティス：全ベンダーで共通に利用できる OAuth2 + HMAC のハイブリッド認証を採用し、ツール定義はベンダー固有の拡張フィールドだけに留めておくと、将来の切替が容易です。
まとめ
- MCP の三層構造（Resources・Tools・Prompts） は、AI と社内システムを「保守しやすい」形で結びつける実務的設計です。
- 2026 年版で追加された機能（コンテキスト管理、マルチモーダルハンドリング、リアルタイムフェデレーション API）は、ステートフル業務や画像＋テキストの同時処理をプロトコルレベルで実現し、エンタープライズ導入の敷居を大きく下げました。
- Python SDK のサンプルコード は
import osを含めた完全版に修正し、画像取得や HMAC 署名といった実装ギャップを埋めています。これにより、すぐに開発環境で動作確認が可能です。 - 導入時のチェックリスト とベンダー別互換性表を活用すれば、認証・監視・CI/CD の観点から安全かつスムーズに本番展開できます。
これらのポイントを踏まえて自社システムへ MCP を組み込めば、AI 活用の 高速化・標準化・セキュリティ確保 が同時に達成でき、業務効率とデータ活用価値が大幅に向上します。ぜひ本稿で示した手順を参考に、まずは パイロットプロジェクト から始めてみてください。
スポンサードリンク