🌸
MCP の Quickstart を HTTP+SSE で実装してみる
はじめに
すごい勢いで MCP(Model Context Protocol) が広がってきています。Anthropic が主導する規格でありながら、OpenAI の Agents SDK でもサポートすることを発表しました。
また Github Star 数で比較しても、他の AI エージェント系のフレームワークと比較しても MCP の注目度は群を抜いていることがわかるかと思います。
MCP クライアントとサーバー間の通信として stdio（標準入力および標準出力を介した通信）と SSE(Server-Sent Events) が定義されています。本記事では MCP の通信規格のうち HTTP + SSE を使ってサーバー、クライアントを python-sdk を使って実装してみました。
MCP HTTP + SSE 通信規格
以下は、公式ドキュメント内の HTTP + SSE の通信規格の解説です。
HTTP + SSE
HTTP + SSE トランスポートでは、サーバは独立したプロセスとして動作し、複数のクライアント接続を同時に処理できる設計になっています。以下の 2 つのエンドポイントが必要です：
- SSE エンドポイント：クライアントが接続し、サーバからのメッセージを受信するためのストリーム。
- HTTP POST エンドポイント：クライアントがサーバへメッセージを送信するための通常の HTTP エンドポイント。
MCP 通信の流れ
1. 接続の確立
- まず、クライアントは SSE エンドポイントに接続。
- 接続後、サーバは必ず
endpointイベントを送信し、クライアントが今後メッセージ送信に使用する HTTP POST 用の URI を通知。
2. メッセージのやり取り
- クライアント → サーバ：
endpointイベントで通知された URI に対して、HTTP POST で JSON 形式のメッセージを送信。 - サーバ → クライアント：SSE の
messageイベントとして、JSON エンコードされたメッセージを送信。
3. 切断
- 通信終了時は、クライアントが SSE 接続を明示的にクローズすることで通信を終了
通信シーケンス図
なお最新仕様（2025 年 3 月 26 日版）では、HTTP+SSE に変わる新たな通信手段として Streamable HTTP に置き換わるようです。ただし python-sdk の最新バージョン v.1.6.0 では未実装のようでしたので、本記事では HTTP+SSE で実装しています
HTTP + SSE をサポートする MCP サーバーの実装
公式ドキュメントの Quickstart では stdio による MCP サーバーの実装例が消化されていましたので、それをそのまま HTTP + SSE に対応させてみます。
python 3.12
mcp[cli] v1.6.0
MCP Server
Quickstart の Server の実装はそのままで、Running the server の箇所を HTTP + SSE 向けに実装し直しました。
from typing import Any
import click
import httpx
import uvicorn
from mcp.server.fastmcp import FastMCP
from mcp.server.sse import SseServerTransport
from starlette.applications import Starlette
from starlette.requests import Request
from starlette.routing import Mount, Route
# (天気取得ToolsはQuickstartのまま)
@click.command()
@click.option("--port", default=8000, help="Port to listen on for SSE")
def main(port: int) -> int:
"""Main function"""
sse = SseServerTransport("/messages/")
async def handle_sse(request: Request) -> None:
_server = mcp._mcp_server
async with sse.connect_sse(
request.scope,
request.receive,
request._send,
) as (reader, writer):
await _server.run(
reader,
writer,
_server.create_initialization_options(),
)
starlette_app = Starlette(
debug=True,
routes=[
Route("/sse", endpoint=handle_sse),
Mount("/messages/", app=sse.handle_post_message),
],
)
uvicorn.run(starlette_app, host="localhost", port=port)
return 0
if __name__ == "__main__":
main()
サーバーを実行すると、Starlette で定義した MCP サーバーが起動します。
❯ uv run server_sse.py
INFO: Started server process [46746]
INFO: Waiting for application startup.
INFO: Application startup complete.
INFO: Uvicorn running on http://localhost:8000 (Press CTRL+C to quit)
MCP Client
Client 側も基本的な実装は Quickstart のままです。connect_to_server の箇所を HTTP + SSE 向けに実装し直しました。
"""client for MCP"""
import asyncio
import json
import os
from contextlib import AsyncExitStack
import openai
from dotenv import load_dotenv
from mcp import ClientSession
from mcp.client.sse import sse_client
# load environment variables from .env
load_dotenv()
class MCPClient:
"""Client for MCP"""
def __init__(self) -> None:
"""Initialize session and client objects"""
self.session: ClientSession | None = None
self.exit_stack = AsyncExitStack()
self.azure_openai_client = openai.AzureOpenAI(
api_version="2024-10-01-preview",
azure_endpoint=os.environ["AZURE_OPENAI_ENDPOINT"],
azure_deployment=os.environ["AZURE_OPENAI_DEPLOYMENT"],
api_key=os.environ["AZURE_OPENAI_API_KEY"],
)
async def connect_to_server(self, url: str) -> None:
"""Connect to an MCP server
Args:
url: MCP server URL
"""
stdio_transport = await self.exit_stack.enter_async_context(
sse_client(url),
)
self.stdio, self.write = stdio_transport
self.session = await self.exit_stack.enter_async_context(
ClientSession(self.stdio, self.write),
)
await self.session.initialize()
# List available tools
response = await self.session.list_tools()
tools = response.tools
print(
"\nConnected to server with tools:",
)
for tool in tools:
print(f"Tool: {tool.name}")
print(f"Description: {tool.description}")
print(f"Input Schema: {tool.inputSchema}")
print("-" * 80)
async def process_query(self, query: str) -> str:
"""Process a query using Claude and available tools"""
(省略)
async def chat_loop(self) -> None:
"""Run an interactive chat loop"""
(省略)
async def cleanup(self) -> None:
"""Clean up resources"""
(省略)
async def main() -> None:
"""Main function"""
if len(sys.argv) < 2:
print("Usage: python client_sse.py <url>")
sys.exit(1)
client = MCPClient()
try:
await client.connect_to_server(sys.argv[1])
await client.chat_loop()
finally:
await client.cleanup()
if __name__ == "__main__":
import sys
asyncio.run(main())
MCP サーバーを起動した状態で、サーバーの URL を指定してクライアントを実行します。
❯ uv run client_sse.py http://localhost:8000/sse
Connected to server with tools:
Tool: get_alerts
Description: Get weather alerts for a US state.
Args:
state: Two-letter US state code (e.g. CA, NY)
Input Schema: {'properties': {'state': {'title': 'State', 'type': 'string'}}, 'required': ['state'], 'title': 'get_alertsArguments', 'type': 'object'}
--------------------------------------------------------------------------------
Tool: get_forecast
Description: Get weather forecast for a location.
Args:
latitude: Latitude of the location
longitude: Longitude of the location
Input Schema: {'properties': {'latitude': {'title': 'Latitude', 'type': 'number'}, 'longitude': {'title': 'Longitude', 'type': 'number'}}, 'required': ['latitude', 'longitude'], 'title': 'get_forecastArguments', 'type': 'object'}
--------------------------------------------------------------------------------
MCP Client Started!
Type your queries or 'quit' to exit.
Query: 明日からサンフランシスコに旅行に行くのですが、 服装はどんな感じがよいですかね？
=================
Use Tool: get_forecast
- Tool Arguments: {'latitude': 37.7749, 'longitude': -122.4194}
- Tool Result: [{'type': 'text', 'text': '\nToday:\nTemperature: 61°F\nWind: 7 to 16 mph NW\nForecast: Sunny. High near 61, with temperatures falling to around 59 in the afternoon. Northwest wind 7 to 16 mph, with gusts as high as 24 mph.\n\n---\n\nTonight:\nTemperature: 46°F\nWind: 5 to 15 mph WNW\nForecast: Mostly clear, with a low around 46. West northwest wind 5 to 15 mph, with gusts as high as 23 mph.\n\n---\n\nThursday:\nTemperature: 62°F\nWind: 1 to 12 mph NW\nForecast: Sunny, with a high near 62. Northwest wind 1 to 12 mph.\n\n---\n\nThursday Night:\nTemperature: 47°F\nWind: 2 to 12 mph WNW\nForecast: Clear, with a low around 47. West northwest wind 2 to 12 mph.\n\n---\n\nFriday:\nTemperature: 67°F\nWind: 2 to 10 mph NW\nForecast: Sunny, with a high near 67. Northwest wind 2 to 10 mph.\n', 'annotations': None}]
=================
サンフランシスコの天気予報によると、明日の最高気温は約62°F（約17°C）、最低気温は47°F（約8°C）で、主に晴れの予報です。風は比較的穏やかですが、少し肌寒く感じることもあるでしょう。
**おすすめの服装：**
- **レイヤリング**: 朝晩は少し冷えるので、軽いジャケットやカーディガンを羽織ると良いです。
- **Tシャツや長袖シャツ**: 日中は暖かくなるので、Tシャツや薄手の長袖シャツが快適です。
- **パンツ**: ジーンズや軽めのパンツが適しています。
- **靴**: 歩きやすいスニーカーやカジュアルな靴を選ぶと良いでしょう。
- **帽子やサングラス**: 晴れた日は日差しが強いので、帽子やサングラスもお勧めです。
特にサンフランシスコは風が強くなることがあるので、風を防げる服装があると良いでしょう。楽しい旅行を！
Function Calling にて MCP サーバーから天気予報情報をコンテキストとして取得できていることが確認できました。
またサーバー側のログを確認すると、セッションごとに session_id が生成され、それぞれのセッションでメッセージのやり取りが行われていることがわかります。
INFO: ::1:53162 - "GET /sse HTTP/1.1" 200 OK
INFO: ::1:53163 - "POST /messages/?session_id=f3fb9710e372468ebbac4044b8093a0c HTTP/1.1" 202 Accepted
INFO: ::1:53163 - "POST /messages/?session_id=f3fb9710e372468ebbac4044b8093a0c HTTP/1.1" 202 Accepted
INFO: ::1:53163 - "POST /messages/?session_id=f3fb9710e372468ebbac4044b8093a0c HTTP/1.1" 202 Accepted
[04/02/25 22:41:29] INFO Processing request of type ListToolsRequest server.py:534
INFO: ::1:53172 - "POST /messages/?session_id=f3fb9710e372468ebbac4044b8093a0c HTTP/1.1" 202 Accepted
[04/02/25 22:42:03] INFO Processing request of type ListToolsRequest server.py:534
INFO: ::1:53172 - "POST /messages/?session_id=f3fb9710e372468ebbac4044b8093a0c HTTP/1.1" 202 Accepted
[04/02/25 22:42:05] INFO Processing request of type CallToolRequest server.py:534
[04/02/25 22:42:06] INFO HTTP Request: GET https://api.weather.gov/points/37.7749,-122.4194 "HTTP/1.1 200 OK" _client.py:1740
INFO HTTP Request: GET https://api.weather.gov/gridpoints/MTR/85,105/forecast "HTTP/1.1 200 OK" _client.py:1740
でもセキュリティには注意が必要
HTTP 通信が可能ということで、MCP サーバーをリモートサーバーとして構築し、複数のユーザーで共有してサーバーを利用することも可能になるわけですが、そこは注意が必要です。例えば MCP が サーバー として公式提供している Githubや Google Drive ではアクセストークンなどの認証情報をサーバーが保持しているため、複数ユーザーが同じ認証情報を使用することになっていまいます。
理想的には、各ユーザーや AI エージェントに対して適切な権限を持つ認証情報を割り当てるプロセスを MCP サーバーに実装することが求められます。この点については、MCP を利用しない場合でも考慮すべき重要な課題ですので、AI エージェントのセキュリティや認証・認可の仕組みについても引き続き検討していきたいと考えています。
Discussion