1️⃣
Part1 : Azure AI Foundry で MCPを使ってみた【深掘りと最新動向調査】

2025/03/30に公開
Azure
Microsoft
Model Context Protocol
Claude
tech
 はじめに：この記事についてこの記事はModel Context Protocol (MCP)という、2024年11月にAnthropicが発表した新しいAIエージェント接続プロトコルについて、その基本概念から実践的な活用方法まで包括的に解説します。特にMicrosoftのAzure AI Foundryとの連携を中心に、具体的なハンズオン手順を通じてMCPの可能性と実用性を探ります。

また、本記事は2部構成となっております。

Part2は以下です。

https://zenn.dev/chips0711/articles/ba72417a1f6c34

 この記事で学べることMCPの基本概念とアーキテクチャを理解できる
クラウドとローカル環境の両方でMCPを実装する方法を学べる
Azure AI Foundryとの連携手順を通じて実践的なスキルを身につけられる
エンタープライズ環境でのMCP活用におけるセキュリティ考察を理解できる
MCP登場から約4ヶ月の2025年3月現在、MCPは急速に業界標準として採用されつつあり、Microsoft、Amazon、GitHub、Cursor、Replitなど多くの企業が対応を進めています。その理由は、MCPがAIエージェントとさまざまなデータソースやツールの連携を劇的に簡素化し、相互運用性を高め、標準化する強力な仕組みを提供するからです。
MCPは単なる「ツール連携」や「データアクセス」の手段ではなく、エンタープライズレベルのAIエコシステムを実現するための基盤技術となりつつあります。この記事では、MCPの基本から実装まで段階的に解説し、誰でも実践できる形でまとめています。

 対象読者AIエージェント開発に興味のあるエンジニアやデベロッパー
LLMと外部ツール・データソースの連携に課題を感じている方
Azure AIサービスを活用したい方
標準プロトコルによるAI開発の効率化に関心のある方
実用的なAIエージェントを構築したいビジネス関係者

 この記事で得られることMCPの概念的理解と技術的詳細（わかりやすい比喩を多用）
基本的なMCPサーバー/クライアントの実装方法（ステップバイステップのコード例）
Azure AI Foundryとの連携手順（具体的なハンズオン）
セキュリティと運用に関する重要な考察
最新の応用事例と今後の展望
さらに、本記事を通じて「AIエージェントがなぜ標準プロトコルを必要としているのか」という根本的な問いにも答えていきます。適切な比喩や例を多用することで、技術的な背景を持たない方にもMCPの価値をわかりやすく伝えることを目指しています。
それでは、AIエージェント開発における「USB-C革命」とも言えるMCPの世界へご案内します。

 1. はじめに：AIエージェント開発に待望の「USB-Cポート」登場！
 AIエージェントと外部連携の現状LLM（大規模言語モデル）を活用したAIエージェントが、私たちの働き方や情報との関わり方を変えつつあります。単なる対話だけでなく、情報収集、データ分析、ソフトウェア操作など、様々なタスクをこなす賢いアシスタントです。
AIエージェントは実際どのような形で私たちの生活に入りつつあるのでしょうか？例えば：
マーケティング担当者が「先月のキャンペーンデータを分析して、主要な成功要因を教えて」と尋ねるだけで、AIがデータベースに接続し、分析を実行して洞察を提供
ソフトウェア開発者が「このバグを修正するには？」と質問すると、AIがGitリポジトリを調査し、コードを分析して解決策を示す
デザイナーが「このロゴを青と緑のグラデーションで修正して」と言うと、AIがデザインツールを操作して編集を行う
財務アナリストが「現在のSQUAREとVISAの株価比較グラフを生成して」と指示すると、AIが金融APIを使って最新データを取得し、視覚化する
カスタマーサポート担当者が「このお客様の過去のケースをすべて確認して、傾向をまとめて」と頼むと、AIがCRMシステムを検索して包括的なサマリーを提供する
しかし、AIエージェントが真価を発揮するには、外部のツールやデータソースとの連携が不可欠。これまでの開発では、LLMごとに仕様が異なったり、ツールごとに個別実装が必要だったりと、多くの「痛み」が伴いました。まるでガジェットごとに違う充電ケーブルが必要だった時代のような、煩雑さが存在したのです。
そんな中、2024年11月26日にAnthropic社によって提唱・公開され、急速に注目を集めているのがModel Context Protocol (MCP) です！(公式ブログ記事)。MCPは、まさに「AI業界のUSB-Cポート」のような存在を目指すオープン標準プロトコルです。

 MCPとは何か？シンプルな例えで理解するMCPを別の角度から説明すると、「AIのためのユニバーサルリモコン」とも言えます。テレビ、スピーカー、照明、さらにはコーヒーマシンまで、すべての機器を1つのリモコンで操作できるイメージです。AIモデルはさまざまな情報源やツールにアクセスする必要がありますが、MCPがなければ、それぞれに個別のリモコン（カスタム連携）が必要になってしまいます。
もう一つの例えとして、「レストランのウェイター」モデルも分かりやすいでしょう：
あなた（AIアシスタント）がテーブルに座って注文します
ウェイター（MCP）がオーダーを受け、キッチン（データベース、API、ツールなど）に伝えます
キッチン（MCPサーバー）が注文を処理し（データ取得や関数実行）、ウェイターに渡します
ウェイターが料理（必要なデータや実行結果）を持ってきます
あなた自身がキッチンに入って料理する（直接APIを操作する）代わりに、ウェイターを通じて必要なものを伝えるだけで、MCPが残りを処理してくれるのです。
また、MCPは「言語サポートデスク」にも例えられます。世界中の様々な言語を話す人々（LLMモデル）が、多様なサービス（データソース、ツール）を利用したいとき、各自が全言語を習得する必要はなく、サポートデスク（MCP）を通じて対話するだけでよいのです。サポートデスクが両者の間で翻訳を行い、コミュニケーションを可能にします。
MCPは、様々なLLM（Claude, GPT-4, Llamaなど）と、多種多様なツール/データソースを、種類やメーカーに関わらず、統一された「お作法」で接続するための「共通言語」を提供します。これにより、これまでサイロ化しがちだったAIコンポーネント間の連携に道を開き、AIエージェント開発における相互運用性を劇的に向上させ、イノベーションを加速させることが期待されています。MCPの登場は、AIエージェント開発が新たな「プロトコル時代」へと突入したことを象徴しています。

 MCPが解決する実際の課題MCPが解決する具体的な問題を、現実世界の例で見てみましょう：
企業のナレッジマネジメントの場合：

MCP導入前: マーケティング部門がAIアシスタントを使用してキャンペーン分析を行いたい場合、IT部門は特定のLLM用にカスタム接続を開発する必要がありました。別部門が異なるLLMで同様の分析を行いたい場合、また一から開発が必要になります。

MCP導入後: 社内データベース用のMCPサーバーを一度開発すれば、どの部門がどのLLMを使っても、同じMCPサーバーを通じてデータにアクセスできます。開発リソースを節約し、全社的なAI活用を加速できます。
ソフトウェア開発の場合：

MCP導入前: コードアシスタントがコードベースを理解するには、各IDE/エディタが独自の連携方法を実装する必要がありました。新しいAIモデルが登場するたびに、全エディタが再実装を強いられます。

MCP導入後: リポジトリ用のMCPサーバーを一度実装すれば、VS Code、JetBrains、Vimなど、MCP対応のエディタならどれでも同じコンテキストにアクセスできます。開発者はお気に入りのツールとAIモデルを自由に選べます。
クラウドとオンプレミスの統合：

MCP導入前: クラウドLLM(Claude、GPT-4など)とローカルLLM(Ollama、Qwenなど)では連携方法が大きく異なり、それぞれに専用の実装が必要でした。

MCP導入後: 同じMCPサーバーをクラウドとローカル両方のLLMから利用できるため、環境に依存せずに一貫した体験を提供できます。組織は必要に応じてLLMを切り替えても、既存のツール連携資産を活用できます。
こうした実用例からもわかるように、MCPは単なる技術的な標準化を超えて、実際のビジネス課題を解決し、AIの実用的な価値を大きく引き上げる可能性を秘めています。
この記事では、Microsoftの Azure AI Foundry プレビュー環境における Azure AI Agent Service と MCP の連携事例を具体的なハンズオンを通して紹介することを主軸に据えつつ、この登場からまだ約4ヶ月（現在2025年3月）ながらも急速に進化するMCPの世界を深く、そして分かりやすく探求していきます。MCPの基本概念、仕組み、最新仕様、実装方法、主要プラットフォームでの活用事例、他の技術との比較、そして忘れてはならないセキュリティに関する重要な考察（第6章）、今後の展望まで、幅広くカバーします。特に、Claude Desktop アプリを使って、Azure上のAIエージェントとMCP経由で対話する手順は、MCPの実際の動きを体験する良い機会になるはずです。

 2. MCPの基本概念とアーキテクチャ
 2.1 MCP登場の背景：AIエージェント連携の課題MCP登場以前、AIエージェントに外部ツール連携を追加するには、多くの「痛み」がありました。
LLMごとの「方言」地獄: LLMごとに連携方法が異なり、開発者はLLMが変わるたびに再実装する必要がありました。これは非効率的で、特定のLLMプラットフォームへのロックインを招き、自由な技術選択を妨げていました。例えば、OpenAIのGPT-4用に開発したツール連携は、AnthropicのClaude用に書き直す必要があり、さらにLlamaベースのモデルに対応するにはまた別の実装が必要でした。
ツールごとの「個別配線」地獄: Web検索API、データベース接続、ファイル操作、各種SaaS API… 連携したいツールが増えるほど、異なるインターフェースや認証方式に対応するための個別実装が必要になり、システムは複雑な「配線」で絡まり合い、メンテナンスも困難な「スパゲッティ状態」に陥りがちでした。
エコシステムの「断片化」問題: 上記の結果、ツール連携機能は特定のLLMや環境に閉じてしまい、再利用性や相互運用性が欠如。コミュニティ全体で知見や資産を共有し、イノベーションを加速させるための共通の「土台」が欠けていました。OpenAIのFunction Calling、AnthropicのTool Use、その他各社のツール呼び出し機能は、基本的なアイデアは似ていてもAPIやパラメータの形式が異なり、統一されていませんでした。
クラウドとローカルLLMの分断: クラウドAPIとして提供されるLLM（GPT-4、Claudeなど）とローカル環境で実行するLLM（Llama、Qwen、Mistralなど）では接続方法が大きく異なり、ツール連携の実装も別々に行う必要がありました。組織がクラウドとローカルの両方の環境を活用しようとすると、同じ機能を二重に実装する無駄が生じていました。
この状況は、かつての独自コネクタが乱立したモバイルデバイスの世界に似ています。そこで登場したのが、接続を標準化したUSB-Cでした。MCPは、これと同様にAIコンポーネント間の「論理的な接続」を標準化する｢AI業界のUSB-C｣を目指しています。

 2.2 MCPの解決策：M×N問題解消と疎結合アーキテクチャMCPが解決する核心的な問題が「M×N問題」です。M種類のLLMとN種類のツールを連携させる場合、従来は最大 M × N 通り の個別実装が必要でした。例えば、3つのLLM（Claude、GPT-4、Llama）と4つのツール（ファイル操作、Web検索、データベース、メール）を連携させる場合、従来は3×4=12の個別統合が必要でした。しかしMCPでは、各LLMがMCPクライアントを、各ツールがMCPサーバーを実装するだけで、3+4=7の実装で済みます。
MCPは、クライアント（LLMアプリ側）とサーバー（ツール側）の間に標準プロトコルを設けることで、これを解決。必要な実装は M + N 個 で済み、開発効率が劇的に向上します。
ここで､かつての日本のガラケー時代を思い出してみます。(世代がバレそうですが･･･)各キャリアが独自の充電端子を採用していた時代には、友人の家に泊まっても充電器を借りられないという悲劇が日常茶飯事でした。しかし、USB-Cの普及により、現在ではほとんどのデバイスが同じケーブル一本で充電できるようになりました。Model Context Protocol (MCP) は、まさにAI界のUSB-Cなのです。
さらに重要なのが、MCPが促す疎結合 (Loosely Coupled) アーキテクチャです。MCPサーバーはLLMアプリ本体とは独立した外部プロセス/サービスとして実装できます。これは、関数ロジックをアプリ内に実装することが多かった従来手法（例: Function Calling）との大きな違いです。
クライアントとサーバーは標準化されたJSON-RPCで通信するため、互いの内部実装を意識する必要がありません。これにより、以下のメリットが生まれます。

役割分担: LLMアプリ開発者とツール開発者が独立して開発可能に。例えば、データベース専門家がSQLツールを、AIアプリケーション開発者がLLMインターフェースを、それぞれの専門領域に集中して開発できます。

再利用性と交換可能性: 作成したサーバーは様々なクライアントから再利用でき、ツールの差し替え（ホットスワップ）も容易に。例えば、一度作成したWebスクレイピングツールは、Claude DesktopでもCursorでも、自社開発のLLMアプリでも利用可能です。

独立したライフサイクル: 各サーバーを独立して開発・デプロイ・スケーリング可能。あるツールの更新が他のシステムに影響を与えにくくなります。

クラウド・ローカルの橋渡し: 同じMCPサーバーが、クラウドLLMとローカルLLMの両方から利用可能。例えば、Ollama上で動作するLlamaモデルからでも、Claude DesktopからでもMCPサーバーにアクセスでき、一貫した体験が提供できます。
MCPは、AIエージェントシステム全体のモジュール性と柔軟性を高める設計思想を提供します。

 2.3 着想源：LSP (Language Server Protocol) に学ぶ成功体験MCPのアイデアって、実は全く新しいものではないんです。ソフトウェア開発の世界には、Language Server Protocol (LSP) という、コードエディタとプログラミング言語のサポート機能を標準化するプロトコルがあり､まさにMCPの「先輩」のような存在がいます。MCPの設計は、Microsoftが開発したLanguage Server Protocol (LSP)に着想を得ています。これを知ると、MCPが目指すものがグッと分かりやすくなるはずです｡
コードエディタ（VS Codeとか）がプログラミング言語（PythonとかJavaとか）の賢い機能（コード補完とかエラー表示とか）を提供しようとすると、それはもう大変でした。エディタごとに、言語ごとに、それぞれ専用の複雑なプログラムを書かなきゃいけなかったんです。まさにM（エディタ）× N（言語）問題！ 新しいエディタや言語が出るたびに、開発者は頭を抱えていました。
そこで登場したのがLSPです。LSPは賢い方法を考え出しました。
エディタ（LSPクライアント）と、言語機能を提供する別プログラム（Language Server）を分離する。
そのクライアントとサーバー間の「会話ルール」（通信プロトコル、実はこれもJSON-RPCベース！）を以下のように標準化する。
たったこれだけで、状況は一変！ エディタ開発者はLSPクライアントを一度作れば、いろんな言語のLSPサーバーと繋がれる。言語開発者はLSPサーバーを一度作れば、いろんなLSP対応エディタで使ってもらえる。M×N問題がM+N問題に解消され、開発効率は爆上がり、エコシステム全体が活性化したんですね。VS Codeがあんなにたくさんの言語を賢くサポートできるのも、LSPのおかげが大きいんです。
MCPは、まさにこのLSPの成功体験をAIエージェントの世界で再現しようとしているわけです。
AIアプリ/LLM（MCPクライアント/ホスト） ←→ LSPクライアント（エディタ）
外部ツール/データソース（MCPサーバー） ←→ Language Server
この間で使われる「会話ルール」をMCPとして標準化することで、AIエージェント開発におけるM×N問題を解決し、相互運用性を高め、エコシステム全体のイノベーションを加速させよう！というのが、MCPの根底にある大きな狙いであり、多くの開発者がザワザワしている理由だと推測されます。

 2.4 主要概念：ツール・リソース・プロンプトという「共通語彙」MCPを使いこなすための基本的な「概念」は以下の3つです。
ツール (Tool): エージェントの「実行能力」
LLMが実行できる具体的な操作やアクション。

例: filesystem/writeFile, web/search, database/query。

定義: 名前、説明 (LLM理解用)、入力スキーマ (JSON Schema)、出力形式、ツールアノテーション (副作用等を示すメタデータ、2025-03-26仕様) で定義。

使い方: サーバーが tools/list で提供。LLMが選択・引数生成し、クライアントが tools/call で実行要求。
具体例:{
  "name": "weather/get",
  "description": "Get current weather for a location",
  "inputSchema": {
    "type": "object",
    "properties": {
      "location": {
        "type": "string",
        "description": "City name or coordinates"
      }
    },
    "required": ["location"]
  },
  "annotations": ["@mcp.tool.readonly"]
}
リソース (Resource): エージェントが扱う「対象物」
LLMが参照・利用できるデータやコンテンツ。主に情報源。「開いているエディタのコード」「DBスキーマ」「ベクトル検索結果」などもリソースになり得ます。

識別: URI で一意に識別。例えば file://project/readme.md や db://postgres/users/schema。

使い方: サーバーが resources/list で通知、クライアントが resources/provide で提供。LLMはURIをプロンプトで参照。アクセス範囲はクライアント/ホストが制御。
具体例:{
  "uri": "file://project/docs/api.md",
  "name": "API Documentation",
  "description": "REST API reference documentation",
  "mimeType": "text/markdown"
}
プロンプト (Prompt): サーバーからの「対話支援テンプレート」
サーバー側から提供できる指示テンプレート、ガイドライン、応答フォーマット例など。

目的: LLMのパフォーマンス向上とクライアント側のプロンプト設計負担軽減。特定のタスク（レポート生成、メール作成など）の品質を安定させるのに役立ちます。

使い方: サーバーが prompts/list で公開、クライアントが prompts/get で取得・利用。
具体例:{
  "name": "email_composer",
  "description": "Create professional email based on intent",
  "arguments": [
    {
      "name": "recipient",
      "description": "Email recipient",
      "required": true
    },
    {
      "name": "intent",
      "description": "Purpose of the email",
      "required": true
    }
  ]
}
これらの3つの概念は、MCPを通じてLLMがどのように外部世界と相互作用するかを定義する基本的な枠組みを提供しています。ツールは「行動」、リソースは「情報」、プロンプトは「対話の形式」に対応しており、これらを組み合わせることで豊かな相互作用が可能になります。

 2.5 通信の仕組み：JSON-RPCと選べる経路 (Stdio/HTTP)
 基本構成要素MCP通信は、以下の主要なコンポーネントから構成されています：

MCPホスト: ユーザーとAIモデル間のインターフェースを提供するアプリケーション（例：Claude Desktop、Cursor IDE、あるいはサードパーティのAIアプリケーション、Ollama UIなど）

MCPクライアント: ホスト内に組み込まれ、MCPサーバーとの通信を処理する部分。複数のサーバーへの接続を管理する

MCPサーバー: 特定のツール、リソース、プロンプトを提供する独立したサービス（例：ファイルシステム、データベース、APIアクセスなど）

サービス: MCPサーバーが実際に連携する外部サービスやシステム（例：OpenWeatherMap API、PostgreSQLデータベース、Google Driveなど）
この構成は柔軟性が高く、1つのホストが複数のMCPサーバーを同時に利用できます。また、MCPサーバーは複数のクライアントにサービスを提供することも可能です。典型的なデプロイメントでは、ホストとMCPサーバーは同じマシン上で実行されますが、リモートサーバーを利用することも可能です。

 通信プロトコルMCPは軽量なJSON-RPC 2.0を使用します。これは単純なJSONフォーマットでリクエスト (id, method, params) とレスポンス (id, result または error) を交換する方式です。
// リクエスト例 (tools/call)
{ "jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": { "toolName": "weather/get", "inputs": {"city": "Tokyo"} } }
// レスポンス例 (成功)
{ "jsonrpc": "2.0", "id": 1, "result": { "temperature": "15℃", "condition": "晴れ" } }
// レスポンス例 (エラー)
{ "jsonrpc": "2.0", "id": 1, "error": { "code": -32000, "message": "APIキーが無効です" } }
JSON-RPCはHTTPやRESTと比較して冗長性が低く、単一の接続を継続的に使用するため効率的です。また、実装が比較的簡単で、多くのプログラミング言語でサポートされているのも利点です。REST APIとの主な違いは以下の通りです：


特徴
JSON-RPC
REST API


通信形式
メソッド呼び出し中心
リソース操作中心

URLの役割
エンドポイントのみ
リソースの識別子

HTTPメソッド
主にPOST
GET, POST, PUT, DELETE等を使い分け

状態
ステートレス
ステートレス

複数操作
バッチ処理サポート
個別リクエスト

標準化
形式が明確に定義
実装によりばらつきあり

ドキュメント
スキーマで自己説明的
別途API仕様書が必要

MCPがJSON-RPCを選んだ理由は、その軽量性、関数呼び出しのような明確なモデル、バッチ処理のサポート、そして厳格な形式定義によるところが大きいと考えられます。

 初期ハンドシェイク通信を開始する前に、クライアントとサーバーは initialize メソッドを使って互換性を確認します：
// initialize リクエスト例
{ "jsonrpc": "2.0", "id": 0, "method": "initialize", "params": { "clientInfo": {"name": "Claude Desktop", "version": "1.2.3"}, "protocolVersion": "2025-03-26", "capabilities": {} } }
// initialize レスポンス例
{ "jsonrpc": "2.0", "id": 0, "result": { "serverInfo": {"name": "weather-service", "version": "1.0.0"}, "protocolVersion": "2025-03-26", "capabilities": {"tools": true, "resources": false} } }
このハンドシェイクで、両者は対応するプロトコルバージョン（例: "2024-11-26"、最新版は"2025-03-26"）とケイパビリティの情報を交換し、互換性を確認します。この初期ハンドシェイクによって、クライアントはサーバーがサポートする機能を動的に理解し、適応することができます。

 接続方式（トランスポート）MCPは2つの主な接続方式をサポートしています：
Stdio（標準入出力）:
ローカル子プロセスと標準入出力経由で通信
シンプルでローカルツールや開発時に最適
親プロセス(ホスト)から子プロセス(サーバー)へのパイプとして機能
同一マシン上での効率的な通信を実現
セキュリティ面でのメリットあり（ネットワーク経由なし）
コマンドラインツールが直接MCPサーバーとして機能可能
HTTP (Streamable HTTP Transport - 最新推奨):
HTTPエンドポイントに接続して通信
単一接続上で双方向ストリーミングが可能
リモート/クラウド連携、OAuth認証に適している
ローカルだけでなくリモートサーバーとの連携も可能
より柔軟な構成を実現
旧来のHTTP+SSEは置き換えが推奨されている
実際には、Streamable HTTP Transportは次のような仕組みで実装されます：
クライアントはHTTP POSTリクエストをサーバーに送信し、接続を確立
サーバーはこの接続を開いたまま保持（ロングポーリング）
クライアントはこの接続を通じてJSON-RPCメッセージを送信
サーバーは同じ接続を使用してレスポンスをストリーミング
大きなデータはチャンクに分割して送信可能
これにより、データベースクエリの結果や大きなファイルの内容などを効率的に転送できます。

 効率化機能（2025-03-26仕様）最新仕様では、JSON-RPCバッチングという機能が追加され、複数のリクエストを一括送信できるようになりました。これにより通信オーバーヘッドを削減し、特に遅延が懸念されるネットワーク環境でのパフォーマンスが向上します。
例えば複数のファイルを読み取る場合：
[
  { "jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": { "toolName": "filesystem/readFile", "inputs": {"path": "/path/to/file1.txt"} } },
  { "jsonrpc": "2.0", "id": 2, "method": "tools/call", "params": { "toolName": "filesystem/readFile", "inputs": {"path": "/path/to/file2.txt"} } }
]
サーバーは同様にバッチレスポンスを返します：
[
  { "jsonrpc": "2.0", "id": 1, "result": { "content": "file1の内容..." } },
  { "jsonrpc": "2.0", "id": 2, "result": { "content": "file2の内容..." } }
]
このバッチ処理により、複数の往復通信を1回にまとめることができ、特にネットワークレイテンシが高い環境でパフォーマンスが向上します。

 典型的な通信フロー例MCPの通信フローを簡単な具体例で説明すると、例えばClaude Desktopを使ってファイルシステムにアクセスする場合：
ホスト起動とクライアント初期化:
Claude Desktopが起動
設定に基づいてファイルシステムMCPサーバーを子プロセスとして起動
MCPクライアントがサーバーと初期ハンドシェイクを実行
ツール探索:
クライアントが tools/list を呼び出し
サーバーが提供する filesystem/readFile, filesystem/writeFile などのツールリストを取得
ユーザー入力とLLM解析:
ユーザーが「デスクトップ上のtodo.txtファイルを読んで要約して」と入力
Claude（LLM）がこれを解析し、適切なツール（filesystem/readFile）の呼び出しが必要と判断
ツール呼び出し:
クライアントが tools/call メソッドで filesystem/readFile ツールを呼び出し
引数として path: "/Users/username/Desktop/todo.txt" を指定
ツール実行と結果返却:
サーバーがファイルを読み込み、その内容をレスポンスとして返却
LLM応答生成:
Claudeがファイル内容を受け取り、要約して結果をユーザーに表示
この全プロセスはユーザーからは見えず、シームレスに行われます。ユーザーは単にAIアシスタントに自然言語で指示を出すだけで、裏側でMCPプロトコルを通じた複雑な相互作用が行われているのです。

 シーケンス図による通信フロー詳細以下のシーケンス図でMCPの通信フローをより詳細に見てみましょう：

 パフォーマンスとスケーラビリティMCPプロトコルは性能面でも優れており、大規模な展開に適しています。統計によると、適切に実装されたMCPサーバーは以下のようなパフォーマンスを実現できます：

レイテンシ: 典型的なツール呼び出しは10-100ミリ秒程度（ネットワーク遅延を除く）

スループット: 単一サーバーで数百〜数千リクエスト/秒を処理可能

スケーラビリティ: 水平スケーリングに対応（ロードバランサー背後に複数サーバーを配置可能）

効率性: JSON-RPCの軽量さにより、帯域幅使用量を最小限に抑制
複数のベンチマークでは、MCPがRESTful APIと比較して30-60%のパフォーマンス向上を示しています。これは、接続の再利用、プロトコルのオーバーヘッド削減、バッチ処理のサポートなどによるものです。実際のパフォーマンスは、実装、ハードウェア、ネットワーク状況によって変わりますが、MCPは効率性を重視して設計されています。

 2.6 セキュリティの基本：同意、認証、アノテーション (概要)MCPは便利ですが、外部連携にはセキュリティが不可欠です。仕様には基本的な安全策が組み込まれています（詳細は第6章で深く考察）。
ユーザー同意: 重要な操作の前にユーザーに許可を求める仕組みが組み込まれています。例えば、ファイル削除やメール送信などの取り消せない操作を実行する前に、ユーザーに確認ダイアログを表示します。これにより、AIが誤って重要なデータを変更したり削除したりするリスクを軽減します。確認ダイアログは通常以下の情報を含みます：
実行されようとしているツールの名前
ツールに渡される引数（パス、ID、その他のパラメータ）
操作の潜在的な影響（特に破壊的な操作の場合）
同意または拒否のオプション
認証 (OAuth 2.1): HTTP接続で安全な認証・認可を行う標準フレームワークが2025-03-26仕様で導入されました。これは、Webサービスやクラウドリソースへのアクセスを安全に行うための業界標準の認証プロトコルで、以下のメリットがあります：
細かいアクセス権限の制御（スコープ）
トークンベースの認証（APIキーよりも安全）
リフレッシュトークンによるセキュリティ強化
サードパーティアプリケーションとの安全な統合
ユーザーIDの伝達と検証
この認証フローの一般的な実装では、MCPクライアントがユーザーをOAuthプロバイダの認証ページにリダイレクトし、認証が成功すると、クライアントはアクセストークンを受け取ります。このトークンは、MCPサーバーへのリクエストに付加され、サーバーは関連するAPIやデータを安全に呼び出すことができます。
ツールアノテーション: ツールのリスク（破壊的操作など）をLLMに伝え、安全な判断を促す機能が2025-03-26仕様で追加されました。例えば：

@mcp.tool.readonly - 読み取り専用で安全なツール

@mcp.tool.destructive - データ削除や変更など取り消せない操作を行うツール

@mcp.tool.costly - APIコストやリソースを大量消費する可能性があるツール

@mcp.tool.requires_human_approval - 人間の承認が必要なツール

@mcp.tool.network_access - ネットワークアクセスを伴うツール

@mcp.tool.local_execution - ローカルコンピュータでコードを実行するツール
これらのアノテーションはLLMのプロンプト内で解釈され、モデルがツールの使用決定を行う際に考慮されます。例えば、破壊的操作のアノテーションがあるツールを使用する前に、モデルはユーザーに確認を求めるか、代替手段を提案するよう促されます。
ゼロトラストモデル: MCP実装の多くは「ゼロトラスト」セキュリティモデルを採用しています。これは次のような原則に基づいています：
すべてのリクエストは常に検証される（信頼することなく検証）
最小権限の原則を適用（必要な最小限のアクセスのみを許可）
すべてのアクセスは監査・記録される
認証と認可は継続的に行われる（セッション中の任意の時点で）
これらの安全策は基盤となるものですが、実際の実装・運用では多層的なセキュリティ対策が必要です。企業環境でMCPを活用する場合は特に以下のような対策を検討すべきです：

アクセス権限の厳格な制限: ツールごとにRBAC（ロールベースアクセス制御）を実装し、ユーザーやグループに応じて利用可能なツールを制限します。

ネットワークレベルでの保護: ファイアウォール、VPN、IPホワイトリストなどを使用して、MCPサーバーへのアクセスを制限します。

データの暗号化: 保存データ（データベース、ファイル）と転送中のデータ（API通信）の両方を暗号化します。

監査ログの取得と分析: すべてのツール呼び出し、その引数、結果を詳細に記録し、異常を検出します。

定期的なセキュリティレビュー: 脆弱性スキャン、ペネトレーションテスト、コードレビューを定期的に実施します。

インシデント対応計画: セキュリティインシデントが発生した場合の対応手順を事前に策定しておきます。
セキュリティは単一の機能ではなく、設計から運用まで一貫した多層防御アプローチが必要です。MCPサーバーの開発と運用においては、セキュリティを最初から組み込んだ「セキュリティ・バイ・デザイン」の考え方が重要です。この考え方に基づけば、セキュリティはあとから追加する機能ではなく、システム設計の最初から考慮すべき基本的要素となります。

 2.7 ツールアノテーション：LLMの判断力を高める新機能 (2025-03-26仕様)最新仕様で導入されたツールアノテーションは、LLMがツールをより賢く、安全に使うための重要な情報を提供します。

目的: ツールの特性や副作用（読み取り専用か、破壊的か、コストがかかるか、人間の承認が必要か等）をメタデータとして伝える。これによりLLMはツールの性質をより正確に理解し、適切に使用できるようになります。

アノテーション例:

@mcp.tool.readonly - データを変更しない安全なツール

@mcp.tool.destructive - データの削除や修正など、取り消せない変更を行うツール

@mcp.tool.costly - APIコストや計算リソースを大量に消費する可能性があるツール

@mcp.tool.requires_human_approval - 実行前に人間の確認が必要なツール


LLMへの効果: LLMはリスクやコストを考慮し、より安全で最適な行動計画を立てられるようになります。例えば、重要なファイルを削除する必要がある場合、LLMは@mcp.tool.destructiveアノテーションを認識して、まずバックアップを作成するなどの安全策を取るよう促されます。
具体的な実装例を見てみましょう：
from mcp.server.fastmcp import FastMCP

mcp_server = FastMCP(name="file-manager")

# 読み取り専用ツールの例
@mcp_server.tool(annotations=["@mcp.tool.readonly"])
async def read_file(path: str) -> str:
    """
    指定されたファイルの内容を読み取ります。
    
    Args:
        path: 読み取るファイルのパス
        
    Returns:
        ファイルの内容を文字列で返します
    """
    with open(path, "r") as f:
        return f.read()

# 破壊的ツールの例
@mcp_server.tool(annotations=["@mcp.tool.destructive", "@mcp.tool.requires_human_approval"])
async def delete_file(path: str) -> str:
    """
    指定されたファイルを完全に削除します。この操作は取り消せません。
    
    Args:
        path: 削除するファイルのパス
        
    Returns:
        削除結果のステータスメッセージ
    """
    import os
    os.remove(path)
    return f"ファイル '{path}' が削除されました"
この例では、read_fileは読み取り専用で安全なツールとしてマークされていますが、delete_fileは破壊的であり、さらに人間の承認が必要であることが明示されています。LLMはこれらのアノテーションを認識し、特に破壊的なツールを使用する前には、ユーザーに警告を出したり、代替案を提案したりすることができます。
ツールアノテーションの導入は、特に企業環境でのMCP採用を促進する重要な進化です。セキュリティチームや管理者はツールの特性を明示的に定義でき、LLMによる不適切なツール使用のリスクを軽減できます。

 2.8 MCP vs 従来技術：何がユニークで、何を目指すのか？MCPの独自性と価値を明確にするために､独断と偏見にはなりますが､従来のAPI､混同されがちなFunction Calling､RAGと比較した表が以下です。


特徴
個別API連携 (例: REST)
Function Calling (例: OpenAI)
RAG (典型的な実装)
Model Context Protocol (MCP)


標準化
なし (独自仕様)
限定的 (プラットフォーム内)
なし (実装パターン)
あり (オープン標準)

LLM依存性
なし
高い
低い
低い (LLM非依存)

ツール実装場所
外部サービス
通常アプリ内
外部知識ソース
独立した外部サーバー

疎結合性
高い
低い
中程度
非常に高い

M×N問題
発生
発生
-
解決 (M+N)

連携の種類
主にHTTP
主に関数呼び出し
主に知識検索
ツール, リソース, プロンプト, 非同期通知など包括的

エコシステム
巨大だが断片的
ベンダー主導
ツール・ライブラリ多数
急成長中、オープン性と相互運用性による効果を目指す

安全性
APIキー管理等
アプリ実装依存
アクセス制御
OAuth 2.1, 同意, スコープ, アノテーション等で多層的に考慮

主な価値
既存API活用
引数生成容易
回答精度向上
相互運用性, 疎結合性, 再利用性, 標準化, 安全性考慮

主な課題
標準化欠如, M×Nコスト
ロックイン, 低疎結合性
検索精度, 最新性, コスト
標準化途上, 実装複雑度(HTTP), デバッグ, コミュニティ成熟度

つまり､従来のAPIと違い、MCPでは各機能がツールとして自己記述的に公開されます。通常のAPIは固定エンドポイント（例：/products、/orders）を持ち、新機能追加やパラメータ変更にはバージョニングという困難が伴います。
一方MCPでは、各ツールが自らの機能、パラメータの意味、出力フォーマット、制約情報を内包しており、以下のような特徴を持ちます：

柔軟な適応性：パラメータ変更があっても、クライアントは動的に対応可能

モジュール性：新ツール追加時にクライアント側の修正が不要

コンテキスト連動：実行環境や状況に応じ、利用可能なツールを適切に制御できる
このように、MCPは単なるAPI呼び出しに留まらず、LLMと外部システム（サーバーやデータソース）との間で、JSON-RPCをベースとした標準化された双方向通信を実現し、広範な連携と再利用性を確保しています。
また、Function Callingとの違いとして、Function CallingはOpenAIなど特定のLLMベンダーが提供する機能で、ツール選択とパラメータ抽出の最適化を実現するために、関数のロジックがアプリ内で実装されます。一方、MCPはツールロジックを独立したサーバーに実装し、LLMと外部機能の連携をベンダー中立のオープン標準で実現することで、ツール選択後の処理の交通整理を行い、システム全体の柔軟性を高めています。
以下の記事も参考になりますのでぜひお読みください｡

https://qiita.com/7shi/items/e27866ce51c6b9a0f605
MCPとRAGの関係については、これらは競合するものではなく、むしろ相補的な技術です。RAG（Retrieval Augmented Generation）はLLMが外部知識を参照して回答の質を向上させる技術パターンですが、その「検索」部分をMCPを通じて標準化することで、より効率的かつ統一的に実装できます。MCPサーバーがベクトル検索APIやドキュメント取得機能を提供することで、RAGシステムをより柔軟かつ統合的に構築できるようになります。
まとめると､MCPは特に相互運用性と疎結合性で優位性を持ち、RAGを含む様々な連携を標準化・促進する基盤プロトコルとして期待されています。

 RESTではなくMCPを使う理由MCPの価値をより明確にするために、なぜ既存のRESTful APIではなくMCPが必要なのかについても考察しておきましょう。
RESTの限界：

AI非親和的: RESTはリソース指向であり、LLMが必要とする関数的な相互作用パターンとミスマッチがあります。LLMはしばしば「何をしたいか」を表現するために関数呼び出しのような形式を好みます。

複雑なデータ型とスキーマ: 多くのREST APIは複雑なデータ型や入れ子構造を持ち、LLMがこれらを正確に構築するのは困難です。

ドキュメント依存: RESTはしばしば人間向けドキュメントに依存し、API自体が自己記述的ではありません。

コンテキスト管理の欠如: RESTは本質的にステートレスであり、複数ステップにわたる対話の文脈を維持するメカニズムが組み込まれていません。
MCPの強み：

AI中心設計: LLMの思考・行動パターンに合わせた設計。ツールやリソースという概念はLLMの自然な操作モデルと一致します。

自己記述的: ツールやリソースは自分自身の機能、パラメータ、制約を説明する情報を持ち、LLMがそれらを理解して適切に使用できます。

統一されたインターフェース: 異なるサービスやデータソースへの標準化されたアクセス方法を提供し、LLMが一貫した方法でツールを使用できます。

バッチ処理と長時間実行: JSON-RPCバッチングと非同期通知により、複雑なタスクや時間のかかる処理をサポートします。
これらの理由から、MCPはAIエージェントが外部システムと相互作用するための、より適切なプロトコルとして設計されています。RESTは一般的なWeb APIとしては優れていますが、LLMとの連携という特定のコンテキストではMCPがより適しているのです。

 2.9 まとめ:なぜMCPが標準として普及しつつあるのかMCPが急速に業界標準として採用されている理由をまとめると以下の通りです：
AI-Native設計: MCPはLLMの特性を考慮して設計されたプロトコルです。従来のOpenAPIやGraphQLなどの汎用APIプロトコルはデータ交換に優れていましたが、LLMの文脈理解や推論能力を最大限に活かすための設計ではありませんでした。MCPはLLMが「どう考えて行動するか」を中心に設計されています。
大企業のバッキング: AnthropicというClaude開発企業がオープンスタンダードとして公開した点が重要です。独自標準ではロックインの懸念がありますが、Anthropicは他のLLMベンダーも使えるよう設計し、Microsoftなど大手の迅速な採用につながりました。2025年3月現在、Microsoft、Amazon、GitHub、Cursor、Replit、Stackblitzなど多くの企業がMCPを採用または積極的にサポートしています。
優れた開発者エクスペリエンス: MCPはSDKやツール群が充実しており、すぐに使い始められます。豊富なリファレンス実装と詳細なドキュメントが公開され、開発者が素早く採用できる環境が整いました。特にPython向けのFastMCPのようなフレームワークは、数行のコードでMCPサーバー実装が可能です。
LSPベースの設計: Language Server Protocol（LSP）というMicrosoftが開発した既存の成功事例をベースに設計されています。VS CodeなどのIDEとLanguage Serverの通信を標準化したLSPの設計パターンをAI領域に応用したことで、既に検証済みのアーキテクチャを活用しています。
明確な開発アプローチ: MCPは「どのようにAIにツールを使わせるか」という重要な問いに対する共通理解と明確な指針を提供します。開発者は自身の実装をゼロから考える代わりに、確立されたパターンに従うことができます。
AI市場の成熟: 2025年のAI市場はツールの連携と統合に重点を置いており、MCPはまさにその需要に応えています。特にAIエージェントの重要性が高まる中で、MCPのタイミングは市場の流れと合致しています。
コミュニティの活性化: GitHub上のawesome-mcp-serversリポジトリに見られるように、多様なツール連携サーバーの開発が急速に進んでいます。開発者コミュニティが積極的に貢献し、エコシステムの拡大を後押ししています。
クラウドからローカルまでの柔軟性: MCPはクラウドLLMとローカルLLM（Ollama、Qwenなど）の両方で利用できるため、組織の環境やニーズに合わせた柔軟な運用が可能です。これにより、大企業から個人開発者まで、幅広いユーザーにとって魅力的な選択肢となっています。
この章ではMCPの基本概念とアーキテクチャについて詳しく解説しました。次の章では、実際にMCPサーバーとクライアントを実装する方法を、シンプルな例から段階的に見ていきます。

 3. MCPサーバー/クライアントの基本実装前章ではMCPの基本概念と理論的背景を理解しました。この章では、実際にMCPサーバーとクライアントを実装する方法を、シンプルな例から始めて段階的に学びます。

 この章で学べることMCPサーバーの基本的な実装方法（Python FastMCP使用）
様々なトランスポート方式（Stdio/HTTP）の実装方法
ツール、リソース、プロンプトの実装パターン
MCP Inspectorによるデバッグと検証方法
JSON-RPCメッセージの直接操作方法
リモート連携の実装テクニック
ハンズオン形式で進めるため、実際にコードを書いて試していただくことをお勧めします。また、ローカルLLM（Ollama、Qwenなど）とクラウドLLM（Claude、GPT-4など）の両方と連携できるMCPの柔軟性も体験できます。

 3.0 簡易入門：素数チェッカーサーバーの実装まずは最もシンプルなMCPサーバーの例として、与えられた数値が素数かどうかを判定する「素数チェッカー」サーバーを作ります。この例を通じて、MCPサーバーの基本構造を理解しましょう。
# prime_checker_server.py
from mcp.server.fastmcp import FastMCP  # FastMCPフレームワークをインポート
import logging

# ロギング設定（デバッグやトラブルシューティングに重要）
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

# MCPサーバーインスタンスの作成
mcp = FastMCP(
    name="prime-checker",  # サーバーの名前（クライアントに表示される）
    description="数値が素数かどうかを判定するサービス",  # サーバーの説明
)

# 素数判定ツール - @mcp.toolデコレータで登録
@mcp.tool(annotations=["@mcp.tool.readonly"])  # 読み取り専用ツールであることを示す
async def is_prime(number: int) -> bool:
    """
    与えられた数値が素数かどうかを判定します。
    
    Args:
        number (int): 判定する数値
    
    Returns:
        bool: 素数の場合はTrue、そうでない場合はFalse
    """
    logger.info(f"素数判定リクエスト: number={number}")
    
    # 2未満の数は素数ではない
    if number < 2:
        return False
    
    # 2は素数
    if number == 2:
        return True
    
    # 偶数（2以外）は素数ではない
    if number % 2 == 0:
        return False
    
    # 3以上の奇数について素数判定
    i = 3
    while i * i <= number:
        if number % i == 0:
            return False
        i += 2
    
    return True

# リソースとしても実装 - @mcp.resourceデコレータで登録
@mcp.resource("prime://{number}")  # URIパターンを指定
async def check_prime(number: int) -> bool:
    """数値が素数かどうかをリソースとして提供します。"""
    logger.info(f"素数判定リソースリクエスト: prime://{number}")
    return await is_prime(number)  # ツール実装を再利用

if __name__ == "__main__":
    logger.info("素数チェッカーMCPサーバー (Stdio) を起動します...")
    mcp.run()  # サーバー起動（デフォルトでStdioトランスポート使用）

 実行方法このサーバーを実行するには、まずPython環境にmcpパッケージをインストールします：
pip install "mcp[cli]"  # CLI機能も含めてインストール
次に、サーバーを起動します：
python prime_checker_server.py

 実行結果サーバーを起動すると、以下のようなログが表示されます：
2025-03-30 10:15:23,456 - INFO - 素数チェッカーMCPサーバー (Stdio) を起動します...
この例で重要なポイント：

FastMCP クラスを使って簡単にMCPサーバーを実装

@mcp.tool デコレータでツールを定義（アノテーション付き）

@mcp.resource デコレータでリソースを定義（URI形式でアクセス可能）

型アノテーションとdocstringから自動的にスキーマが生成される

mcp.run() でサーバーを標準入出力（Stdio）で起動
このシンプルな例からでも、MCPの基本的な設計思想である「ツール」と「リソース」の概念が実装されていることがわかります。ツールは「関数として呼び出す操作」を、リソースは「URI経由でアクセスする情報」を表現しています。

 3.1 Stdioサーバー実装例 (Python FastMCP)次に、もう少し実用的な電卓サーバーの例を見てみましょう。この例では、基本的な算術演算を行うツールをいくつか定義します。
# simple_calculator_server.py
from mcp.server.fastmcp import FastMCP
import logging

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

# サーバーインスタンスの作成
mcp = FastMCP(
    name="calculator",
    description="簡単な算術計算を実行します。",
)

# 加算ツール
@mcp.tool(annotations=["@mcp.tool.readonly"])
async def add(a: float, b: float) -> float:
    """2つの数値を加算します。"""
    logger.info(f"ツール 'add' 呼び出し: a={a}, b={b}")
    result = a + b
    logger.info(f"計算結果: {result}")
    return result

# 減算ツール
@mcp.tool(annotations=["@mcp.tool.readonly"])
async def subtract(a: float, b: float) -> float:
    """最初の数値から2番目の数値を減算します。"""
    logger.info(f"ツール 'subtract' 呼び出し: a={a}, b={b}")
    result = a - b
    logger.info(f"計算結果: {result}")
    return result

# 乗算ツール
@mcp.tool(annotations=["@mcp.tool.readonly"])
async def multiply(a: float, b: float) -> float:
    """2つの数値を乗算します。"""
    logger.info(f"ツール 'multiply' 呼び出し: a={a}, b={b}")
    result = a * b
    logger.info(f"計算結果: {result}")
    return result

# 除算ツール（エラーハンドリングの例）
@mcp.tool(annotations=["@mcp.tool.readonly"])
async def divide(a: float, b: float) -> float:
    """最初の数値を2番目の数値で除算します。"""
    logger.info(f"ツール 'divide' 呼び出し: a={a}, b={b}")
    
    if b == 0:
        error_msg = "0による除算はできません"
        logger.error(error_msg)
        raise ValueError(error_msg)  # 適切なエラーハンドリング
    
    result = a / b
    logger.info(f"計算結果: {result}")
    return result

if __name__ == "__main__":
    logger.info("シンプルな電卓MCPサーバー (Stdio) を起動します...")
    mcp.run()

 実行と動作検証サーバーを起動し、MCP Inspectorで動作を確認できます：
python -m mcp dev simple_calculator_server.py
MCP Inspectorでは、各ツールを手動で呼び出して結果を確認できます。例えば、addツールに a=5, b=3 を渡すと、結果として 8 が返ってきます。

 エラーハンドリングこの例では、divideツールで0除算のエラーハンドリングも実装しています。MCPサーバーでのエラーは、適切な例外を発生させることで処理します。これらのエラーはJSON-RPCのエラーレスポンスとしてクライアントに返されます：
{
  "jsonrpc": "2.0",
  "id": 1,
  "error": {
    "code": -32000,
    "message": "0による除算はできません",
    "data": {
      "traceback": "..."
    }
  }
}
さらに、外部APIとの連携例として、天気情報サーバーも実装できます：
# weather_mcp_server.py
from mcp.server.fastmcp import FastMCP
import httpx
import os
import json
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

mcp = FastMCP(
    name="weather-service",
    description="現在の天気情報を取得するサービス",
)

# 環境変数からAPIキーを取得（推奨方法）
API_KEY = os.getenv("OPENWEATHERMAP_API_KEY")

@mcp.tool(annotations=["@mcp.tool.readonly", "@mcp.tool.network_access"])
async def get_current_weather(city: str, units: str = "metric") -> str:
    """指定された都市の現在の天気情報を取得します。"""
    logger.info(f"天気情報リクエスト: city={city}, units={units}")
    
    if not API_KEY:
        return json.dumps({"error": "APIキーが設定されていません"})
    
    try:
        url = f"https://api.openweathermap.org/data/2.5/weather"
        params = {
            "q": city,
            "units": units,
            "appid": API_KEY
        }
        
        async with httpx.AsyncClient() as client:
            response = await client.get(url, params=params)
            response.raise_for_status()
            data = response.json()
        
        # 結果を整形
        result = {
            "city": data["name"],
            "temperature": f"{data['main']['temp']}°{'C' if units == 'metric' else 'F'}",
            "condition": data["weather"][0]["main"],
            "description": data["weather"][0]["description"]
        }
        
        return json.dumps(result, ensure_ascii=False)
    
    except Exception as e:
        return json.dumps({"error": f"天気情報の取得中にエラーが発生: {str(e)}"})

if __name__ == "__main__":
    logger.info("天気情報MCPサーバーを起動します...")
    mcp.run()
この例では、外部WebAPIと連携する方法と、環境変数を使って安全にAPIキーを管理する方法を示しています。また、@mcp.tool.network_accessアノテーションを使用して、このツールがネットワークアクセスを行うことをLLMに明示しています。

 3.2 HTTPサーバー実装のポイントStdioだけでなく、HTTPを使ったリモート接続も可能です。最新仕様のStreamable HTTP Transportは、FastAPIなどのWebフレームワークを使って実装できます。
# http_mcp_server.py
from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import StreamingResponse
import asyncio
import json
import logging
from pydantic import BaseModel
from datetime import datetime
from typing import Dict, List, Optional, Union

# FastAPIアプリの設定
app = FastAPI(title="MCP HTTP Server")
app.add_middleware(CORSMiddleware, allow_origins=["*"])

# ロギング設定
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# JSON-RPCリクエストモデル
class JsonRpcRequest(BaseModel):
    jsonrpc: str
    id: Optional[int] = None
    method: str
    params: Dict = {}

# 利用可能なツール定義
available_tools = {
    "calculator/add": {
        "name": "calculator/add",
        "description": "Add two numbers together",
        "inputSchema": {
            "type": "object",
            "properties": {
                "a": {"type": "number", "description": "First number"},
                "b": {"type": "number", "description": "Second number"}
            },
            "required": ["a", "b"]
        },
        "annotations": ["@mcp.tool.readonly"]
    },
    "datetime/now": {
        "name": "datetime/now",
        "description": "Get current date and time",
        "inputSchema": {
            "type": "object",
            "properties": {
                "format": {"type": "string", "description": "Output format (optional)"}
            }
        },
        "annotations": ["@mcp.tool.readonly"]
    }
}

# ツール実装
async def execute_tool(tool_name: str, params: dict) -> dict:
    """ツール名とパラメータに基づいて実行処理を行う関数"""
    logger.info(f"ツール実行: {tool_name}, パラメータ: {params}")
    
    if tool_name == "calculator/add":
        if "a" not in params or "b" not in params:
            raise ValueError("Parameters 'a' and 'b' are required")
        return {"result": params["a"] + params["b"]}
    
    elif tool_name == "datetime/now":
        format_str = params.get("format", "%Y-%m-%d %H:%M:%S")
        try:
            return {"datetime": datetime.now().strftime(format_str)}
        except ValueError as e:
            raise ValueError(f"Invalid datetime format: {str(e)}")
    
    raise ValueError(f"Unknown tool: {tool_name}")

# メッセージハンドラ
async def handle_message(message: JsonRpcRequest) -> dict:
    """JSON-RPCメッセージを処理する関数"""
    try:
        if message.method == "initialize":
            # 初期化リクエスト処理
            return {
                "jsonrpc": "2.0",
                "id": message.id,
                "result": {
                    "serverInfo": {"name": "mcp-http-server", "version": "1.0.0"},
                    "protocolVersion": "2025-03-26",
                    "capabilities": {"tools": True}
                }
            }
        elif message.method == "tools/list":
            # ツール一覧を返す
            return {
                "jsonrpc": "2.0",
                "id": message.id,
                "result": {"tools": list(available_tools.values())}
            }
        elif message.method == "tools/call":
            # ツール呼び出し処理
            tool_name = message.params.get("toolName")
            inputs = message.params.get("inputs", {})
            
            if not tool_name:
                return {
                    "jsonrpc": "2.0",
                    "id": message.id,
                    "error": {"code": -32602, "message": "Missing required parameter: toolName"}
                }
            
            try:
                result = await execute_tool(tool_name, inputs)
                return {
                    "jsonrpc": "2.0",
                    "id": message.id,
                    "result": result
                }
            except Exception as e:
                return {
                    "jsonrpc": "2.0",
                    "id": message.id,
                    "error": {"code": -32000, "message": str(e)}
                }
        else:
            # 未知のメソッド
            return {
                "jsonrpc": "2.0",
                "id": message.id,
                "error": {"code": -32601, "message": f"Method not found: {message.method}"}
            }
    except Exception as e:
        # 予期しないエラー
        logger.exception("メッセージ処理中にエラーが発生")
        return {
            "jsonrpc": "2.0",
            "id": message.id,
            "error": {"code": -32603, "message": f"Internal error: {str(e)}"}
        }

# MCP HTTPエンドポイント
@app.post("/mcp")
async def mcp_endpoint(request: Request):
    """MCPリクエストを処理するエンドポイント"""
    client_id = id(request)
    logger.info(f"新しい接続: client_id={client_id}")
    
    async def process_messages():
        """受信メッセージを処理してレスポンスを生成するジェネレーター"""
        try:
            async for chunk in request.stream():
                if not chunk:
                    continue
                
                try:
                    data = json.loads(chunk.decode("utf-8"))
                    logger.debug(f"受信データ: {data}")
                    
                    # バッチリクエスト処理
                    if isinstance(data, list):
                        logger.info(f"バッチリクエスト処理: {len(data)}件")
                        responses = []
                        for item in data:
                            req = JsonRpcRequest(**item)
                            resp = await handle_message(req)
                            responses.append(resp)
                        yield json.dumps(responses).encode("utf-8") + b"\n"
                    else:
                        req = JsonRpcRequest(**data)
                        resp = await handle_message(req)
                        yield json.dumps(resp).encode("utf-8") + b"\n"
                
                except json.JSONDecodeError:
                    logger.error("JSONパースエラー")
                    error_resp = {
                        "jsonrpc": "2.0",
                        "id": None,
                        "error": {"code": -32700, "message": "Parse error"}
                    }
                    yield json.dumps(error_resp).encode("utf-8") + b"\n"
        except Exception as e:
            logger.exception("ストリーム処理中にエラーが発生")
            error_resp = {
                "jsonrpc": "2.0",
                "id": None,
                "error": {"code": -32603, "message": f"Stream processing error: {str(e)}"}
            }
            yield json.dumps(error_resp).encode("utf-8") + b"\n"
        finally:
            logger.info(f"接続終了: client_id={client_id}")
    
    return StreamingResponse(
        process_messages(),
        media_type="application/x-ndjson"
    )

if __name__ == "__main__":
    import uvicorn
    logger.info("MCP HTTPサーバーを起動します...")
    uvicorn.run(app, host="0.0.0.0", port=8080)

 実行方法HTTPサーバーを起動するには、以下のようにします：
python http_mcp_server.py
サーバーが起動すると、以下のようなログが表示されます：
INFO:root:MCP HTTPサーバーを起動します...
INFO:     Started server process [12345]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://0.0.0.0:8080 (Press CTRL+C to quit)

 HTTPベースのMCPサーバーの主な特徴双方向ストリーミング通信：
単一HTTP接続上で複数のリクエスト/レスポンスをやり取り
ロングポーリングによる効率的な通信
大きなデータも効率的に転送可能
リモートアクセスと分散配置：
異なるマシンやクラウド環境からのアクセスが可能
マイクロサービスアーキテクチャに適合
複数のLLMアプリケーションから同時利用可能
セキュリティ強化：
OAuth 2.1認証の統合が容易
TLS/SSLによる通信暗号化
APIキー、JWT、OAuthなどの様々な認証方式をサポート
スケーラビリティ：
水平スケーリングが容易（ロードバランサー背後に複数サーバー配置可能）
クラウドネイティブな展開に最適
コンテナ化が容易
バッチ処理：
複数リクエストの一括処理による効率化
複雑なワークフローの最適化
レイテンシを削減
HTTPサーバー実装は特に企業環境やクラウドサービスでの利用に適しています。一方、Stdioトランスポートはローカルアプリケーションや開発/テスト環境でより簡単に使用できます。目的や環境に応じて適切なトランスポートを選択することが重要です。

 3.3 リソースとプロンプトの実装例ツールだけでなく、リソースとプロンプトを組み合わせた例です。この例では、開発支援のためのツール、リソース、プロンプトを提供するMCPサーバーを実装します。
# full_mcp_server.py
from mcp.server.fastmcp import FastMCP
import logging
from typing import Dict, List, Any
import uuid
import datetime

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

mcp = FastMCP(
    name="development-helper",
    description="開発支援ツールとリソースを提供するMCPサーバー",
)

# ======== ツール実装 ========
@mcp.tool(annotations=["@mcp.tool.readonly"])
async def generate_uuid() -> str:
    """ランダムなUUIDを生成します。"""
    result = str(uuid.uuid4())
    logger.info(f"UUIDを生成しました: {result}")
    return result

@mcp.tool(annotations=["@mcp.tool.readonly"])
async def convert_timestamp(timestamp: int, format: str = "%Y-%m-%d %H:%M:%S") -> str:
    """Unixタイムスタンプを読みやすい日時形式に変換します。"""
    try:
        dt = datetime.datetime.fromtimestamp(timestamp)
        formatted = dt.strftime(format)
        logger.info(f"タイムスタンプ変換: {timestamp} -> {formatted}")
        return formatted
    except ValueError as e:
        logger.error(f"タイムスタンプ変換エラー: {e}")
        raise ValueError(f"無効なタイムスタンプまたはフォーマット: {e}")

# ======== リソース実装 ========
@mcp.resource("file://project/README.md")
async def get_readme() -> str:
    """プロジェクトのREADME.mdファイルを返します。"""
    content = """# サンプルプロジェクト

このプロジェクトはMCPの機能を示すためのサンプルです。

## 機能
- ツール： UUID生成、タイムスタンプ変換
- リソース: README、APIドキュメント
- プロンプト: コード生成テンプレート
"""
    logger.info("README.mdリソースが要求されました")
    return content

@mcp.resource("file://project/docs/api.md")
async def get_api_docs() -> str:
    """API仕様ドキュメントを返します。"""
    content = """# API仕様

## エンドポイント

### GET /api/v1/users
ユーザー一覧を取得します。

#### パラメータ
- `limit` (オプション): 取得する最大件数
- `offset` (オプション): 取得開始位置

#### レスポンス例

{
  "users": [
    {
      "id": 1,
      "name": "山田太郎",
      "email": "yamada@example.com"
    }
  ],
  "total": 100
}
"""
    logger.info("API仕様リソースが要求されました")
    return content

# ======== プロンプト実装 ========
@mcp.prompt()
def generate_api_endpoint(
    method: str, 
    endpoint: str, 
    description: str
) -> str:
    """新しいAPIエンドポイントの仕様を生成するためのプロンプト。"""
    logger.info(f"APIエンドポイント生成プロンプトが要求されました: {method} {endpoint}")
    
    return f"""プロジェクトのAPI仕様に新しいエンドポイントを追加します。
以下の情報に基づいて、Markdown形式でAPIエンドポイントの仕様を作成してください。

- HTTPメソッド: {method}
- エンドポイント: {endpoint}
- 説明: {description}

仕様には以下の項目を含めてください：
1. リクエストパラメータ
2. レスポンス形式（JSONスキーマ）
3. 考えられるエラーコードと説明
4. サンプルリクエスト
5. サンプルレスポンス
"""

@mcp.prompt()
def create_model_class(
    model_name: str, 
    attributes: List[Dict[str, str]]
) -> str:
    """データモデルクラスを生成するためのプロンプト。"""
    logger.info(f"モデルクラス生成プロンプトが要求されました: {model_name}")
    
    attributes_text = "\n".join([
        f"- {attr['name']}: {attr['type']} - {attr['description']}"
        for attr in attributes
    ])
    
    return f"""以下の情報に基づいて、Pythonのデータモデルクラスを作成してください。
SQLAlchemyとPydanticの両方のバージョンを提供してください。

モデル名: {model_name}

属性:
{attributes_text}

両方のモデルで適切なインポート文、型ヒント、ドキュメント文字列を含めてください。
また、必要に応じてバリデーションロジックも追加してください。
"""

if __name__ == "__main__":
    logger.info("開発支援MCPサーバーを起動します...")
    mcp.run()

 実行結果例MCP Inspectorでサーバーに接続すると、ツール、リソース、プロンプトが一覧表示されます。以下はcreate_model_classプロンプトを実行した例です：
プロンプトパラメータ: 
{
  "model_name": "User",
  "attributes": [
    {"name": "id", "type": "int", "description": "ユーザーID"},
    {"name": "name", "type": "str", "description": "ユーザー名"},
    {"name": "email", "type": "str", "description": "メールアドレス"}
  ]
}

プロンプト出力:
以下の情報に基づいて、Pythonのデータモデルクラスを作成してください。
SQLAlchemyとPydanticの両方のバージョンを提供してください。

モデル名: User

属性:
- id: int - ユーザーID
- name: str - ユーザー名
- email: str - メールアドレス

両方のモデルで適切なインポート文、型ヒント、ドキュメント文字列を含めてください。
また、必要に応じてバリデーションロジックも追加してください。
この例では：

ツール - 具体的な操作を実行する機能（UUID生成、タイムスタンプ変換）

リソース - URIでアクセスできる情報（README、API仕様書）

プロンプト - LLMの回答生成を支援するテンプレート（API仕様生成、モデルクラス生成）
をそれぞれ実装しています。これらの組み合わせにより、開発者支援のための豊富な機能を提供するMCPサーバーが構築できます。

 3.4 デバッグツール: MCP Inspectormcp SDK付属のMCP Inspectorは、サーバーをテストするための強力なWebインターフェースを提供します。これにより、MCPサーバーの機能を視覚的に検証できます。

 基本的な使い方# 基本的な使用方法
python -m mcp dev simple_calculator_server.py

# HTTPサーバーの場合
python -m mcp dev --uri http://localhost:8080/mcp

# 特定ポートを指定する場合
python -m mcp dev simple_calculator_server.py --port 8888

 MCP Inspectorの主な機能サーバー情報の表示:
サーバー名、バージョン、対応プロトコルバージョン
サポート機能（ケイパビリティ）一覧
ツール操作:
利用可能なツール一覧の表示
ツールの詳細情報（説明、入力スキーマ、アノテーション）の確認
ツールの対話的な呼び出しと結果確認
入力パラメータのバリデーション
リソース操作:
利用可能なリソース一覧の表示
URIパターンの確認
リソース内容の取得と表示
プロンプト操作:
利用可能なプロンプト一覧の表示
プロンプトパラメータの設定
プロンプトテンプレートの取得と表示
通信ログ:
JSON-RPCメッセージの完全なログ
リクエスト/レスポンスのペアリング表示
タイムスタンプと応答時間の表示
エラー分析:
エラーレスポンスの詳細表示
トレースバックの確認
エラーコードの説明

 デバッグのヒントMCP Inspectorを使ったデバッグでは、以下のポイントが役立ちます：

エラーメッセージの確認: エラー発生時はInspectorの「Error」タブで詳細を確認できます。

コンソールログの併用: サーバー側のログ出力も同時に確認すると問題発見が容易になります。

通信シーケンスの確認: 「Log」タブでは通信の順序や内容を確認できます。

スキーマの検証: ツールの入力スキーマが正しく定義されているか確認できます。

パフォーマンス測定: レスポンス時間を確認して、最適化が必要な箇所を特定できます。
MCP Inspectorは開発時の強力な味方であり、サーバーの機能確認だけでなく、クライアント側の接続テストにも活用できます。

 3.5 JSON-RPCメッセージの直接操作MCPプロトコルを深く理解するには、JSON-RPCメッセージを直接操作する方法も知っておくと便利です。以下は主要なメッセージ例です：

 1. 初期化メッセージ// クライアントからの初期化リクエスト
{
    "jsonrpc": "2.0",
    "id": 1,
    "method": "initialize",
    "params": {
        "protocolVersion": "2025-03-26",
        "capabilities": {},
        "clientInfo": {
            "name": "debug-client",
            "version": "1.0"
        }
    }
}

// サーバーからの初期化レスポンス
{
    "jsonrpc": "2.0",
    "id": 1,
    "result": {
        "serverInfo": {
            "name": "weather-service",
            "version": "1.0.0"
        },
        "protocolVersion": "2025-03-26",
        "capabilities": {
            "tools": true,
            "resources": true,
            "prompts": false
        }
    }
}

 2. ツール一覧取得// クライアントからのツール一覧リクエスト
{
    "jsonrpc": "2.0",
    "id": 2,
    "method": "tools/list",
    "params": {}
}

// サーバーからのツール一覧レスポンス
{
    "jsonrpc": "2.0",
    "id": 2,
    "result": {
        "tools": [
            {
                "name": "add",
                "description": "2つの数値を加算します。",
                "inputSchema": {
                    "type": "object",
                    "properties": {
                        "a": {"type": "number"},
                        "b": {"type": "number"}
                    },
                    "required": ["a", "b"]
                },
                "annotations": ["@mcp.tool.readonly"]
            },
            // 他のツール定義...
        ]
    }
}

 3. ツール呼び出し// クライアントからのツール呼び出しリクエスト
{
    "jsonrpc": "2.0",
    "id": 3,
    "method": "tools/call",
    "params": {
        "toolName": "add",
        "inputs": {"a": 5, "b": 3}
    }
}

// サーバーからのツール呼び出しレスポンス
{
    "jsonrpc": "2.0",
    "id": 3,
    "result": {
        "result": 8
    }
}

 4. リソース読み取り// クライアントからのリソース読み取りリクエスト
{
    "jsonrpc": "2.0",
    "id": 4,
    "method": "resources/read",
    "params": {
        "uri": "file://project/README.md"
    }
}

// サーバーからのリソース読み取りレスポンス
{
    "jsonrpc": "2.0",
    "id": 4,
    "result": {
        "content": "# サンプルプロジェクト\n\nこのプロジェクトはMCPの機能を示すためのサンプルです。..."
    }
}

 5. プロンプト取得// クライアントからのプロンプト取得リクエスト
{
    "jsonrpc": "2.0",
    "id": 5,
    "method": "prompts/get",
    "params": {
        "name": "create_model_class",
        "args": {
            "model_name": "User",
            "attributes": [
                {"name": "id", "type": "int", "description": "ユーザーID"},
                {"name": "name", "type": "str", "description": "ユーザー名"}
            ]
        }
    }
}

// サーバーからのプロンプト取得レスポンス
{
    "jsonrpc": "2.0",
    "id": 5,
    "result": {
        "content": "以下の情報に基づいて、Pythonのデータモデルクラスを作成してください。..."
    }
}

 Pythonでの直接操作例以下は、Pythonスクリプトを使ってMCPサーバーと直接通信する例です：
# mcp_direct_client.py
import asyncio
import json
import httpx
import sys
import subprocess
import os

async def debug_mcp_server_http(server_url, method, params=None):
    """HTTPトランスポートでMCPサーバーと通信"""
    if params is None:
        params = {}
    
    request = {
        "jsonrpc": "2.0",
        "id": 1,
        "method": method,
        "params": params
    }
    
    print(f"\n送信リクエスト:\n{json.dumps(request, indent=2, ensure_ascii=False)}")
    
    async with httpx.AsyncClient() as client:
        try:
            response = await client.post(server_url, json=request)
            response_data = response.json()
            print(f"\n受信レスポンス:\n{json.dumps(response_data, indent=2, ensure_ascii=False)}")
            return response_data
        except Exception as e:
            print(f"エラー: {e}")
            return None

async def debug_mcp_server_stdio(command, args, method, params=None):
    """Stdioトランスポートでサーバーと通信"""
    if params is None:
        params = {}
    
    request = {
        "jsonrpc": "2.0",
        "id": 1,
        "method": method,
        "params": params
    }
    
    print(f"\n送信リクエスト:\n{json.dumps(request, indent=2, ensure_ascii=False)}")
    
    # サーバープロセスを起動
    cmd = [command] + args
    proc = subprocess.Popen(
        cmd,
        stdin=subprocess.PIPE,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=True,
        encoding='utf-8'
    )
    
    # リクエスト送信
    request_str = json.dumps(request) + "\n"
    proc.stdin.write(request_str)
    proc.stdin.flush()
    
    # レスポンス読み取り
    response_str = proc.stdout.readline()
    try:
        response_data = json.loads(response_str)
        print(f"\n受信レスポンス:\n{json.dumps(response_data, indent=2, ensure_ascii=False)}")
        return response_data
    except json.JSONDecodeError:
        print(f"JSONデコードエラー: {response_str}")
        return None
    finally:
        proc.terminate()

async def main():
    # コマンドライン引数から実行モードを決定
    if len(sys.argv) < 2:
        print("使用方法:")
        print("  HTTP: python mcp_direct_client.py http http://localhost:8080/mcp")
        print("  Stdio: python mcp_direct_client.py stdio python calculator_server.py")
        return
    
    mode = sys.argv[1]
    
    # 初期化リクエストのパラメータ
    init_params = {
        "protocolVersion": "2025-03-26",
        "capabilities": {},
        "clientInfo": {"name": "debug-client", "version": "1.0"}
    }
    
    if mode == "http":
        # HTTPモード
        if len(sys.argv) < 3:
            print("HTTPモードではサーバーURLを指定してください")
            return
        
        server_url = sys.argv[2]
        print(f"HTTPモードでMCPサーバーに接続: {server_url}")
        
        # 初期化
        await debug_mcp_server_http(server_url, "initialize", init_params)
        
        # ツール一覧取得
        await debug_mcp_server_http(server_url, "tools/list")
        
        # ツール呼び出し例
        await debug_mcp_server_http(server_url, "tools/call", {
            "toolName": "calculator/add",
            "inputs": {"a": 10, "b": 20}
        })
        
    elif mode == "stdio":
        # Stdioモード
        if len(sys.argv) < 3:
            print("Stdioモードではコマンドを指定してください")
            return
        
        command = sys.argv[2]
        args = sys.argv[3:] if len(sys.argv) > 3 else []
        print(f"Stdioモードでサーバーに接続: {command} {' '.join(args)}")
        
        # 初期化
        await debug_mcp_server_stdio(command, args, "initialize", init_params)
        
    else:
        print(f"不明なモード: {mode}")

if __name__ == "__main__":
    asyncio.run(main())

 実行例# HTTPサーバーとの通信
python mcp_direct_client.py http http://localhost:8080/mcp

# Stdioサーバーとの通信
python mcp_direct_client.py stdio python calculator_server.py
このスクリプトはMCPプロトコルの仕組みを理解するのに役立ちます。また、自動テストやCI/CDパイプラインでのMCPサーバー検証にも活用できます。

 3.6 リモートMCPサーバーとクライアント連携リモートMCPサーバーは、ネットワーク越しにMCPサービスを提供するために重要です。HTTPトランスポートを使用したサーバー/クライアント連携の例を見てみましょう。

 サーバー側実装例（FastAPI + MCP）# remote_weather_server.py
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
from mcp.server import Server
from mcp.server.fastmcp import FastMCP
import uvicorn
import asyncio
import logging
import httpx
import os
import json

# ロギング設定
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# FastAPIアプリ
app = FastAPI(title="Weather MCP Server")

# MCPサーバー
mcp_server = FastMCP(
    name="weather-service",
    description="天気情報を提供するMCPサーバー",
)

# 環境変数からAPIキーを取得
API_KEY = os.getenv("OPENWEATHERMAP_API_KEY", "")

@mcp_server.tool(annotations=["@mcp.tool.readonly", "@mcp.tool.network_access"])
async def get_weather(city: str, units: str = "metric") -> dict:
    """指定された都市の現在の天気情報を取得します。"""
    logger.info(f"天気情報リクエスト: city={city}, units={units}")
    
    if not API_KEY:
        return {"error": "APIキーが設定されていません"}
    
    try:
        url = "https://api.openweathermap.org/data/2.5/weather"
        params = {
            "q": city,
            "units": units,
            "appid": API_KEY
        }
        
        async with httpx.AsyncClient() as client:
            response = await client.get(url, params=params)
            response.raise_for_status()
            data = response.json()
        
        # 結果を整形
        result = {
            "city": data["name"],
            "country": data["sys"]["country"],
            "temperature": data["main"]["temp"],
            "feels_like": data["main"]["feels_like"],
            "humidity": data["main"]["humidity"],
            "pressure": data["main"]["pressure"],
            "condition": data["weather"][0]["description"],
            "icon": data["weather"][0]["icon"]
        }
        
        return result
    
    except Exception as e:
        logger.error(f"天気情報取得エラー: {str(e)}")
        return {"error": f"天気情報の取得中にエラーが発生: {str(e)}"}

@mcp_server.prompt()
def weather_forecast_prompt(city: str) -> str:
    """天気予報レポートのプロンプトを生成します。"""
    return f"""以下の都市の現在の天気情報に基づいて、天気予報レポートを作成してください。
都市: {city}

get_weather ツールを使用して、現在の天気情報を取得できます。

レポートには以下の情報を含めてください:
1. 現在の気温と体感温度
2. 天気の状態（晴れ、曇り、雨など）
3. 湿度と気圧
4. その日の適切な服装や活動についてのアドバイス

レポートは一般向けで、わかりやすく、友好的な文体で書いてください。
"""

# MCP JSON-RPC エンドポイント
@app.post("/mcp")
async def mcp_endpoint(request: Request):
    """MCPサーバーのHTTPエンドポイント"""
    client_id = id(request)
    logger.info(f"新しい接続: client_id={client_id}")
    
    # リクエストをストリーミング
    async def request_generator():
        async for chunk in request.stream():
            if chunk:
                yield chunk

    # レスポンスをストリーミング
    async def response_streaming():
        transport = StreamableHttpTransport(request_generator())
        server_stream = ManagedTransportStream(transport)
        
        await mcp_server.run_on_stream(server_stream)
        
        async for response in server_stream:
            yield response

    return StreamingResponse(
        response_streaming(),
        media_type="application/x-ndjson"
    )

# ストリーム管理クラス
class ManagedTransportStream:
    """MCPのStreamableHttpTransportを管理するクラス"""
    def __init__(self, transport):
        self.transport = transport
        self.queue = asyncio.Queue()
        self._task = asyncio.create_task(self._process_transport())
    
    async def _process_transport(self):
        try:
            async for msg in self.transport:
                await self.queue.put(msg)
        except Exception as e:
            logger.error(f"トランスポート処理エラー: {e}")
        finally:
            await self.queue.put(None)  # 終了シグナル
    
    async def read(self):
        msg = await self.queue.get()
        if msg is None:
            raise EOFError("ストリームが終了しました")
        return msg
    
    async def write(self, data):
        await self.transport.write(data)
    
    def __aiter__(self):
        return self
    
    async def __anext__(self):
        try:
            msg = await self.read()
            return msg
        except EOFError:
            raise StopAsyncIteration

# StreamableHttpTransport実装
class StreamableHttpTransport:
    """MCPのStreamable HTTP Transportの実装"""
    def __init__(self, request_stream):
        self.request_stream = request_stream
        self.response_queue = asyncio.Queue()
    
    async def read(self):
        async for chunk in self.request_stream:
            return chunk.decode('utf-8')
        raise EOFError("リクエストストリームが終了しました")
    
    async def write(self, data):
        if isinstance(data, dict):
            data = json.dumps(data)
        if not isinstance(data, bytes):
            data = data.encode('utf-8')
        await self.response_queue.put(data + b'\n')
    
    def __aiter__(self):
        return self
    
    async def __anext__(self):
        try:
            return await self.read()
        except EOFError:
            raise StopAsyncIteration

if __name__ == "__main__":
    logger.info("リモート天気MCPサーバーを起動します...")
    uvicorn.run(app, host="0.0.0.0", port=8000)

 クライアント側実装例# weather_client.py
import asyncio
import httpx
import json
import sys
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class MCPClient:
    """MCPクライアント実装"""
    def __init__(self, server_url):
        self.server_url = server_url
        self.request_id = 0
        self.client = httpx.AsyncClient()
    
    async def initialize(self):
        """サーバーとの初期化ハンドシェイク"""
        self.request_id += 1
        request = {
            "jsonrpc": "2.0",
            "id": self.request_id,
            "method": "initialize",
            "params": {
                "protocolVersion": "2025-03-26",
                "capabilities": {},
                "clientInfo": {"name": "MCPClient", "version": "1.0"}
            }
        }
        
        logger.info("サーバーに接続しています...")
        response = await self._send_request(request)
        
        if response and "result" in response:
            server_info = response["result"].get("serverInfo", {})
            logger.info(f"接続成功: {server_info.get('name', 'Unknown')} {server_info.get('version', '')}")
            return True
        else:
            logger.error("接続に失敗しました")
            return False
    
    async def list_tools(self):
        """利用可能なツール一覧を取得"""
        self.request_id += 1
        request = {
            "jsonrpc": "2.0",
            "id": self.request_id,
            "method": "tools/list",
            "params": {}
        }
        
        response = await self._send_request(request)
        if response and "result" in response and "tools" in response["result"]:
            tools = response["result"]["tools"]
            logger.info(f"{len(tools)}個のツールが利用可能です:")
            for i, tool in enumerate(tools):
                logger.info(f"  {i+1}. {tool['name']} - {tool.get('description', 'No description')}")
            return tools
        else:
            logger.error("ツール一覧の取得に失敗しました")
            return []
    
    async def list_prompts(self):
        """利用可能なプロンプト一覧を取得"""
        self.request_id += 1
        request = {
            "jsonrpc": "2.0",
            "id": self.request_id,
            "method": "prompts/list",
            "params": {}
        }
        
        response = await self._send_request(request)
        if response and "result" in response and "prompts" in response["result"]:
            prompts = response["result"]["prompts"]
            logger.info(f"{len(prompts)}個のプロンプトが利用可能です:")
            for i, prompt in enumerate(prompts):
                logger.info(f"  {i+1}. {prompt['name']} - {prompt.get('description', 'No description')}")
            return prompts
        else:
            logger.info("プロンプトはありません")
            return []
    
    async def call_tool(self, tool_name, inputs):
        """ツールを呼び出す"""
        self.request_id += 1
        request = {
            "jsonrpc": "2.0",
            "id": self.request_id,
            "method": "tools/call",
            "params": {
                "toolName": tool_name,
                "inputs": inputs
            }
        }
        
        logger.info(f"ツール '{tool_name}' を呼び出し中...")
        response = await self._send_request(request)
        
        if response and "result" in response:
            logger.info("ツール呼び出し成功")
            return response["result"]
        elif response and "error" in response:
            logger.error(f"ツール呼び出しエラー: {response['error'].get('message', 'Unknown error')}")
            return response["error"]
        else:
            logger.error("ツール呼び出しに失敗しました")
            return None
    
    async def get_prompt(self, prompt_name, args):
        """プロンプトを取得"""
        self.request_id += 1
        request = {
            "jsonrpc": "2.0",
            "id": self.request_id,
            "method": "prompts/get",
            "params": {
                "name": prompt_name,
                "args": args
            }
        }
        
        logger.info(f"プロンプト '{prompt_name}' を取得中...")
        response = await self._send_request(request)
        
        if response and "result" in response and "content" in response["result"]:
            logger.info("プロンプト取得成功")
            return response["result"]["content"]
        elif response and "error" in response:
            logger.error(f"プロンプト取得エラー: {response['error'].get('message', 'Unknown error')}")
            return None
        else:
            logger.error("プロンプト取得に失敗しました")
            return None
    
    async def _send_request(self, request):
        """リクエストを送信して応答を待つ"""
        try:
            response = await self.client.post(
                self.server_url,
                json=request,
                timeout=30.0
            )
            
            if response.status_code == 200:
                try:
                    return response.json()
                except json.JSONDecodeError:
                    logger.error(f"JSONデコードエラー: {response.text}")
                    return None
            else:
                logger.error(f"HTTPエラー: {response.status_code} - {response.text}")
                return None
        
        except Exception as e:
            logger.error(f"リクエスト送信エラー: {e}")
            return None
    
    async def close(self):
        """クライアントをクローズ"""
        await self.client.aclose()

async def main():
    """メイン関数"""
    if len(sys.argv) < 2:
        print("使用法: python weather_client.py <server_url> [city]")
        print("例: python weather_client.py http://localhost:8000/mcp Tokyo")
        return
    
    server_url = sys.argv[1]
    city = sys.argv[2] if len(sys.argv) > 2 else "Tokyo"
    
    client = MCPClient(server_url)
    
    try:
        # サーバーに接続
        if not await client.initialize():
            return
        
        # ツール一覧を取得
        tools = await client.list_tools()
        if not tools:
            return
        
        # プロンプト一覧を取得
        prompts = await client.list_prompts()
        
        # 天気情報を取得
        weather_result = await client.call_tool("get_weather", {"city": city})
        if weather_result:
            print("\n🌤️ 天気情報:")
            print(json.dumps(weather_result, indent=2, ensure_ascii=False))
        
        # 天気予報プロンプトを取得
        if any(p["name"] == "weather_forecast_prompt" for p in prompts):
            prompt = await client.get_prompt("weather_forecast_prompt", {"city": city})
            if prompt:
                print("\n📝 天気予報プロンプト:")
                print(prompt)
    
    finally:
        # クライアントをクローズ
        await client.close()

if __name__ == "__main__":
    asyncio.run(main())

 実行方法# サーバー側（APIキーを設定）
export OPENWEATHERMAP_API_KEY="あなたのAPIキー"
python remote_weather_server.py

# クライアント側
python weather_client.py http://localhost:8000/mcp Tokyo

 実行結果例INFO:__main__:サーバーに接続しています...
INFO:__main__:接続成功: weather-service 1.0.0
INFO:__main__:1個のツールが利用可能です:
INFO:__main__:  1. get_weather - 指定された都市の現在の天気情報を取得します。
INFO:__main__:1個のプロンプトが利用可能です:
INFO:__main__:  1. weather_forecast_prompt - 天気予報レポートのプロンプトを生成します。
INFO:__main__:ツール 'get_weather' を呼び出し中...
INFO:__main__:ツール呼び出し成功

🌤️ 天気情報:
{
  "city": "Tokyo",
  "country": "JP",
  "temperature": 18.5,
  "feels_like": 18.1,
  "humidity": 76,
  "pressure": 1012,
  "condition": "scattered clouds",
  "icon": "03d"
}

INFO:__main__:プロンプト 'weather_forecast_prompt' を取得中...
INFO:__main__:プロンプト取得成功

📝 天気予報プロンプト:
以下の都市の現在の天気情報に基づいて、天気予報レポートを作成してください。
都市: Tokyo

get_weather ツールを使用して、現在の天気情報を取得できます。

レポートには以下の情報を含めてください:
1. 現在の気温と体感温度
2. 天気の状態（晴れ、曇り、雨など）
3. 湿度と気圧
4. その日の適切な服装や活動についてのアドバイス

レポートは一般向けで、わかりやすく、友好的な文体で書いてください。
このように、HTTPトランスポートを使用することで、リモートのMCPサーバーとクライアントが効率的に通信できます。これはクラウド環境やマイクロサービスアーキテクチャでのMCP活用に適した方法です。

 3.7 MCPとローカルLLMの連携MCPはクラウドLLMだけでなく、ローカルで動作するLLM（Ollama、Qwenなど）とも連携できます。以下では、Ollamaで動作するLlama 3またはQwenなどのローカルLLMとMCPを連携させる方法を紹介します。

 前提条件
Ollamaのインストール
Function Callingをサポートするモデル（Llama 3、Qwen、Mistralなど）

mcp-cliのインストール
# mcp-cli のインストール
pip install mcp-cli

 MCPサーバーの設定server_config.jsonファイルを作成し、使用したいMCPサーバーを定義します：
{
  "mcpServers": {
    "filesystem": {
      "command": "npx",
      "args": [
        "-y",
        "@modelcontextprotocol/server-filesystem",
        "/path/to/your/directory"
      ]
    },
    "weather": {
      "command": "python",
      "args": [
        "weather_mcp_server.py"
      ],
      "env": {
        "OPENWEATHERMAP_API_KEY": "your_api_key_here"
      }
    }
  }
}

 ローカルLLMとMCPサーバーの接続# Ollamaを使ったLlama 3モデルとMCPサーバーの接続（チャットモード）
mcp-cli chat --server filesystem --provider ollama --model llama3

# または、Qwenモデルを使用
mcp-cli chat --server filesystem --provider ollama --model qwen2.5:7b

 インタラクティブチャットの例> List files in the current directory

I'll help you list files in the current directory. Let me use the filesystem tools to get that information.

I need to use the tools/call method to execute the listDirectory tool.

The result shows the following files and directories:
- .git/ (directory)
- README.md
- server_config.json
- weather_mcp_server.py
- mcp_client.py

> What's inside README.md?

Let me read the contents of README.md for you.

I'll use the readFile tool to get the contents of this file.

The content of README.md is:

# MCP Demo Project

This project demonstrates the integration of Model Context Protocol (MCP) with local LLMs through Ollama.

## Features
- File system access through MCP
- Weather information retrieval
- Works with Llama 3, Qwen, and other function-calling capable models

## Setup
...

 ローカルLLMを使用する利点
プライバシーとセキュリティ: センシティブデータが組織外に流出しない

コスト効率: 利用量に応じた課金がなく、固定コストで運用可能

レイテンシ: インターネット接続に依存せず、低レイテンシでの応答が可能

カスタマイズ性: 特定ドメイン向けにファインチューニングしたモデルを使用可能

オフライン動作: インターネット接続なしでも機能
ローカルLLMのパフォーマンスは一般的にクラウドLLMに比べて制限がありますが、特定のユースケースでは十分に実用的です。MCPの標準化により、同じツールセットをクラウドとローカル両方のLLMで活用できる柔軟性は大きな利点です。

 3.8 まとめ：MCPサーバー実装の要点MCPサーバー実装の主なポイントをまとめると：
基本実装:

FastMCPを使って簡潔に実装
デコレータ、型ヒント、docstringを活用
適切なトランスポート（StdioかHTTP）を選択
十分なログ出力で動作状況を把握
高度な実装:
HTTPトランスポートでリモート接続
Streamable HTTP Transportで双方向通信を実現
リソースとプロンプトの活用
セキュリティとエラーハンドリングの考慮
クラウドとローカルLLMの両方をサポート
開発ツール:
MCP Inspectorで視覚的にテスト
JSON-RPCメッセージを直接操作してデバッグ
効果的なロギングを実装
自動テストでの検証
ベストプラクティス:
ツールとリソースの適切な使い分け
明確なドキュメントとスキーマ定義
多層防御によるセキュリティ対策
適切なエラーハンドリング
パフォーマンスを考慮した実装
APIキーなどの機密情報は環境変数で管理
MCPサーバーの実装は単なるAPIの提供ではなく、AIエージェントとのインタラクションを設計する作業です。これらの知識を基に、独自のMCPサーバーを開発し、組織のAIエコシステムに新たな価値を加えることができます。
次章では、Azure AI FoundryとMCPの連携に焦点を当て、エンタープライズレベルのAIエージェントをどのように構築できるかを学びます。
次章以降は以下のPart2となります↓

https://zenn.dev/chips0711/articles/ba72417a1f6c34
免責事項本記事は情報提供を目的としており、2025年3月30日時点の情報に基づいています。本記事について､内容の正確性・完全性は保証されず、誤りを含む可能性があります｡公式ドキュメントで最新情報をご確認ください。記事内のコードサンプルは自己責任でご利用ください。APIキー等の機密情報は適切に管理し、公開環境での使用時はセキュリティに十分ご注意ください。本記事内容の利用によって生じたいかなる損害（サービスの中断、データ損失、営業損失等を含む）についても、著者は一切の責任を負いません。OpenAI社およびMicrosoft社､Anthropic社などの各社製品・サービスは各社の利用規約に従ってご利用ください。
特徴	JSON-RPC	REST API
通信形式	メソッド呼び出し中心	リソース操作中心
URLの役割	エンドポイントのみ	リソースの識別子
HTTPメソッド	主にPOST	GET, POST, PUT, DELETE等を使い分け
状態	ステートレス	ステートレス
複数操作	バッチ処理サポート	個別リクエスト
標準化	形式が明確に定義	実装によりばらつきあり
ドキュメント	スキーマで自己説明的	別途API仕様書が必要
特徴	個別API連携 (例: REST)	Function Calling (例: OpenAI)	RAG (典型的な実装)	Model Context Protocol (MCP)
標準化	なし (独自仕様)	限定的 (プラットフォーム内)	なし (実装パターン)	あり (オープン標準)
LLM依存性	なし	高い	低い	低い (LLM非依存)
ツール実装場所	外部サービス	通常アプリ内	外部知識ソース	独立した外部サーバー
疎結合性	高い	低い	中程度	非常に高い
M×N問題	発生	発生	-	解決 (M+N)
連携の種類	主にHTTP	主に関数呼び出し	主に知識検索	ツール, リソース, プロンプト, 非同期通知など包括的
エコシステム	巨大だが断片的	ベンダー主導	ツール・ライブラリ多数	急成長中、オープン性と相互運用性による効果を目指す
安全性	APIキー管理等	アプリ実装依存	アクセス制御	OAuth 2.1, 同意, スコープ, アノテーション等で多層的に考慮
主な価値	既存API活用	引数生成容易	回答精度向上	相互運用性, 疎結合性, 再利用性, 標準化, 安全性考慮
主な課題	標準化欠如, M×Nコスト	ロックイン, 低疎結合性	検索精度, 最新性, コスト	標準化途上, 実装複雑度(HTTP), デバッグ, コミュニティ成熟度
Discussion

ログインするとコメントできます