はじめに
Google Next 2025 でも AI 関連の発表が数多くありました。
その中でも AI Agent 開発周りでも多くの発表があり、A2A プロトコルが発表されました。
A2A プロトコルは MCP プロトコルを補完する機能という説明があったため、改めて MCP を利用するだけでなく、開発保守などを実施していく観点から詳細を調べてみることにしました。
準備
MCP サーバーとそれを利用するシステムを開発する準備を行います。
# 開発環境の準備
uv init mcp-test
cd mcp-test
uv venv
source .venv/bin/activate
# 利用するライブラリのインストール
uv add "mcp[cli]" httpx google-genai
uv は、Rust 製の超高速 Python パッケージインストーラー兼仮想環境マネージャーです。
従来のツールより圧倒的に速く、パッケージ管理と仮想環境作成をこれ一つでこなします。
Python 開発の効率を上げる新定番ツールとして、今大きな注目を集めています。
検証
mcp サーバーは Anthropic の例 をそのまま利用することにしました。
mcp サーバーのコード
weather.py
from typing import Any
import httpx
from mcp.server.fastmcp import FastMCP
# Initialize FastMCP server
mcp = FastMCP("weather")
# Constants
NWS_API_BASE = "https://api.weather.gov"
USER_AGENT = "weather-app/1.0"
async def make_nws_request(url: str) -> dict[str, Any] | None:
"""Make a request to the NWS API with proper error handling."""
headers = {
"User-Agent": USER_AGENT,
"Accept": "application/geo+json"
}
async with httpx.AsyncClient() as client:
try:
response = await client.get(url, headers=headers, timeout=30.0)
response.raise_for_status()
return response.json()
except Exception:
return None
def format_alert(feature: dict) -> str:
"""Format an alert feature into a readable string."""
props = feature["properties"]
return f"""
Event: {props.get('event', 'Unknown')}
Area: {props.get('areaDesc', 'Unknown')}
Severity: {props.get('severity', 'Unknown')}
Description: {props.get('description', 'No description available')}
Instructions: {props.get('instruction', 'No specific instructions provided')}
"""
@mcp.tool()
async def get_alerts(state: str) -> str:
"""Get weather alerts for a US state.
Args:
state: Two-letter US state code (e.g. CA, NY)
"""
url = f"{NWS_API_BASE}/alerts/active/area/{state}"
data = await make_nws_request(url)
if not data or "features" not in data:
return "Unable to fetch alerts or no alerts found."
if not data["features"]:
return "No active alerts for this state."
alerts = [format_alert(feature) for feature in data["features"]]
return "\n---\n".join(alerts)
@mcp.tool()
async def get_forecast(latitude: float, longitude: float) -> str:
"""Get weather forecast for a location.
Args:
latitude: Latitude of the location
longitude: Longitude of the location
"""
# First get the forecast grid endpoint
points_url = f"{NWS_API_BASE}/points/{latitude},{longitude}"
points_data = await make_nws_request(points_url)
if not points_data:
return "Unable to fetch forecast data for this location."
# Get the forecast URL from the points response
forecast_url = points_data["properties"]["forecast"]
forecast_data = await make_nws_request(forecast_url)
if not forecast_data:
return "Unable to fetch detailed forecast."
# Format the periods into a readable forecast
periods = forecast_data["properties"]["periods"]
forecasts = []
for period in periods[:5]: # Only show next 5 periods
forecast = f"""
{period['name']}:
Temperature: {period['temperature']}°{period['temperatureUnit']}
Wind: {period['windSpeed']} {period['windDirection']}
Forecast: {period['detailedForecast']}
"""
forecasts.append(forecast)
return "\n---\n".join(forecasts)
if __name__ == "__main__":
# Initialize and run the server
mcp.run(transport='stdio')
次に、mcp クライアントのコードは折角なので Gemini を利用できるように Google の公式ドキュメントの例 を流用しました。
Google Next 2025 に参加した記念に、ラスベガスの天気を調べてもらうことにしました。
mcp クライアントのコード
main.py
import asyncio
import os
from datetime import datetime
from zoneinfo import ZoneInfo
from google import genai
from google.genai import types
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
client = genai.Client(api_key=os.getenv("GEMINI_API_KEY"))
# Create server parameters for stdio connection
server_params = StdioServerParameters(
command="uv", # Executable
args=["run", "logger.py"], # Weather MCP Server
env=None, # Optional environment variables
)
async def run():
async with stdio_client(server_params) as (read, write):
async with ClientSession(read, write) as session:
# Prompt to get the weather for the current day in Las Vegas.
prompt = f"What is the weather in Las Vegas(latitude: 36.1699, longitude: -115.1398) in today?"
# Initialize the connection between client and server
await session.initialize()
# Get tools from MCP session and convert to Gemini Tool objects
mcp_tools = await session.list_tools()
tools = [
types.Tool(
function_declarations=[
{
"name": tool.name,
"description": tool.description,
"parameters": {
k: v
for k, v in tool.inputSchema.items()
if k not in ["additionalProperties", "$schema"]
},
}
]
)
for tool in mcp_tools.tools
]
# Send request to the model with MCP function declarations
response = client.models.generate_content(
model="gemini-2.0-flash",
contents=prompt,
config=types.GenerateContentConfig(
temperature=0,
tools=tools,
),
)
# Check for a function call
if response.candidates[0].content.parts[0].function_call:
function_call = response.candidates[0].content.parts[0].function_call
print(function_call)
# Call the MCP server with the predicted tool
result = await session.call_tool(
function_call.name, arguments=function_call.args
)
print(result.content[0].text)
# Continue as shown in step 4 of "How Function Calling Works"
# and create a user friendly response
else:
print("No function call found in the response.")
print(response.text)
# Start the asyncio event loop and run the main function
asyncio.run(run())
実際にコードを実行すると、以下の通りラスベガスの天気を取得することができました。
$ uv run main.py
[05/07/25 15:10:34] INFO Processing request of type ListToolsRequest server.py:534
id=None args={'longitude': -115.1398, 'latitude': 36.1699} name='get_forecast'
[05/07/25 15:10:35] INFO Processing request of type CallToolRequest server.py:534
INFO HTTP Request: GET https://api.weather.gov/points/36.1699,-115.1398 "HTTP/1.1 200 OK" _client.py:1740
[05/07/25 15:10:36] INFO HTTP Request: GET https://api.weather.gov/gridpoints/VEF/123,98/forecast "HTTP/1.1 200 OK" _client.py:1740
Tonight:
Temperature: 59°F
Wind: 5 mph SW
Forecast: A slight chance of rain showers before 11pm. Mostly cloudy, with a low around 59. Southwest wind around 5 mph. Chance of precipitation is 20%.
---
Wednesday:
Temperature: 80°F
Wind: 5 mph ESE
Forecast: Sunny, with a high near 80. East southeast wind around 5 mph.
---
Wednesday Night:
Temperature: 64°F
Wind: 3 mph W
Forecast: Mostly clear, with a low around 64. West wind around 3 mph.
---
Thursday:
Temperature: 88°F
Wind: 5 mph ENE
Forecast: Sunny, with a high near 88. East northeast wind around 5 mph.
---
Thursday Night:
Temperature: 69°F
Wind: 5 mph WSW
Forecast: Mostly clear, with a low around 69. West southwest wind around 5 mph.
ちなみに、以下のようにファイルを変更し、東京の天気を取得するように指示をすると、以下の通り天気を取得できないと応答が行われます。
weather.py
58c58
< """Get weather forecast for a location.
---
> """Get weather forecast for a location In US.
main.py
23c23
< prompt = f"What is the weather in Tokyo(latitude: 35.709, longitude: 139.7319) in today?"
---
> prompt = f"What is the weather in Las Vegas(latitude: 36.1699, longitude: -115.1398) in today?"
$ uv run main2.py
[05/07/25 15:24:20] INFO Processing request of type ListToolsRequest server.py:534
No function call found in the response.
I can only provide weather forecasts for locations in the US.
mcp クライアントと mcp サーバーとの通信は stdio もしくは SSE 経由で JSON-RPC でやりとりを実施していることは ドキュメント に記載があったのですが、通信の詳細がどうなっているかが分からなかったため詳細を確認してみることにしました。
そのため、以下のように stdin をロギングするスクリプトを作成して確認しました。
(Python では stdout の読み込みはブロッキングされてしまうため、応答が帰らないときに処理がそこで止まってしまうため stdin だけをまずは確認してみることにしました。)
クライアント / サーバー間の通信を確認するコード
logger.py
import sys
import subprocess
# --- 実行したい外部コマンドを指定 ---
# 例1: tee コマンドで標準入力の内容を標準出力と test.txt に追記する
external_command = ['uv', 'run', 'weather.py', '>' '/Users/nobuhiro.ohashi/Workspace/99.work/blog/20250502/mcp-client/test.txt']
# ---------------------------------
process = None # subprocess.Popen オブジェクトを保持する変数
try:
process = subprocess.Popen(
external_command,
stdin=subprocess.PIPE,
stdout=sys.stdout,
stderr=sys.stderr,
text=True,
encoding='utf-8'
)
# 標準入力の内容を保存するファイルを開く
fstdin = open("input.txt", 'w', encoding='UTF-8')
# Pythonスクリプト自身の標準入力からデータを読み込み、それを外部コマンドの標準入力に書き込みます。
# sys.stdin はデフォルトでシステムのエンコーディングでテキストモードで開かれています。
for line in sys.stdin:
try:
# ファイルへ書き込み
fstdin.write(line)
fstdin.flush()
# サブプロセスの標準入力パイプに読み込んだ行を書き込む
# write() は文字列またはバイト列を受け取ります。text=True なので文字列を渡します。
process.stdin.write(line)
# すぐにサブプロセスにデータが渡されるように、サブプロセスのstdinバッファをフラッシュする
process.stdin.flush()
except BrokenPipeError:
# サブプロセスが予期せず終了した場合などに BrokenPipeError が発生します
sys.stderr.write("Error: Subprocess pipe broken. Subprocess might have terminated early.\n")
break # エラーが発生したら入力の転送を中断
if process.stdin:
process.stdin.close()
# サブプロセスが終了するのを待ち、終了コードを取得します。
returncode = process.wait()
# サブプロセスの終了コードを確認します。0以外は通常エラーを示します。
if returncode != 0:
sys.stderr.write(f"Error: Subprocess '{external_command[0]}' exited with code {returncode}\n")
sys.exit(returncode) # サブプロセスと同じ終了コードでスクリプトも終了
except FileNotFoundError:
sys.stderr.write(f"Error: Command not found: '{external_command[0]}'\n")
sys.exit(127)
except Exception as e:
sys.stderr.write(f"An unexpected error occurred: {e}\n")
sys.exit(1)
finally:
if process and process.poll() is None: # poll() が None ならプロセスはまだ実行中
sys.stderr.write("Warning: Subprocess is still running. Attempting to terminate it.\n")
try:
process.terminate()
process.wait(timeout=5)
if process.poll() is None:
sys.stderr.write("Warning: Subprocess did not terminate. Killing it.\n")
process.kill()
process.wait()
except Exception as term_e:
sys.stderr.write(f"Error during subprocess termination: {term_e}\n")
ログを確認すると mcp クライアントから mcp サーバーに送られたリクエストは以下のようになっていることがわかりました。
{"method":"initialize","params":{"protocolVersion":"2024-11-05","capabilities":{"sampling":{},"roots":{"listChanged":true}},"clientInfo":{"name":"mcp","version":"0.1.0"}},"jsonrpc":"2.0","id":0}
{"method":"notifications/initialized","jsonrpc":"2.0"}
{"method":"tools/list","jsonrpc":"2.0","id":1}
{"method":"tools/call","params":{"name":"get_forecast","arguments":{"longitude":-115.1398,"latitude":36.1699}},"jsonrpc":"2.0","id":2}
次にそれぞれのリクエストがどのような応答を返すかを確認します。 uv run weather.py を実行すると、コンソールで入力待ちになるので、上で確認した入力を順番に実行していきます。
$ uv run weather.py
# input
{"method":"initialize","params":{"protocolVersion":"2024-11-05","capabilities":{"sampling":{},"roots":{"listChanged":true}},"clientInfo":{"name":"mcp","version":"0.1.0"}},"jsonrpc":"2.0","id":0}
# output
{"jsonrpc":"2.0","id":0,"result":{"protocolVersion":"2024-11-05","capabilities":{"experimental":{},"prompts":{"listChanged":false},"resources":{"subscribe":false,"listChanged":false},"tools":{"listChanged":false}},"serverInfo":{"name":"weather","version":"1.7.0"}}}
# input
{"method":"notifications/initialized","jsonrpc":"2.0"}
# ouput
なし
# input
{"method":"tools/list","jsonrpc":"2.0","id":1}
# output
{"jsonrpc":"2.0","id":1,"result":{"tools":[{"name":"get_alerts","description":"Get weather alerts for a US state.\n\nArgs:\n state: Two-letter US state code (e.g. CA, NY)\n","inputSchema":{"properties":{"state":{"title":"State","type":"string"}},"required":["state"],"title":"get_alertsArguments","type":"object"}},{"name":"get_forecast","description":"Get weather forecast for a location In US.\n\nArgs:\n latitude: Latitude of the location\n longitude: Longitude of the location\n","inputSchema":{"properties":{"latitude":{"title":"Latitude","type":"number"},"longitude":{"title":"Longitude","type":"number"}},"required":["latitude","longitude"],"title":"get_forecastArguments","type":"object"}}]}}
# input
{"method":"tools/call","params":{"name":"get_forecast","arguments":{"longitude":-115.1398,"latitude":36.1699}},"jsonrpc":"2.0","id":2}
# output
{"jsonrpc":"2.0","id":2,"result":{"content":[{"type":"text","text":"\nOvernight:\nTemperature: 59°F\nWind: 3 mph SW\nForecast: Mostly cloudy, with a low around 59. Southwest wind around 3 mph.\n\n---\n\nWednesday:\nTemperature: 80°F\nWind: 5 mph ESE\nForecast: Sunny, with a high near 80. East southeast wind around 5 mph.\n\n---\n\nWednesday Night:\nTemperature: 64°F\nWind: 3 mph W\nForecast: Mostly clear, with a low around 64. West wind around 3 mph.\n\n---\n\nThursday:\nTemperature: 88°F\nWind: 5 mph ENE\nForecast: Sunny, with a high near 88. East northeast wind around 5 mph.\n\n---\n\nThursday Night:\nTemperature: 69°F\nWind: 5 mph WSW\nForecast: Mostly clear, with a low around 69. West southwest wind around 5 mph.\n"}],"isError":false}}
これで mcp クライアントがサーバーから受け取る応答がわかりました。
特に以下の 3 番目の tools/list の応答で mcp サーバーがどのようなことができるのかをクライアントに提示していることがわかります。
{
"jsonrpc": "2.0",
"id": 1,
"result": {
"tools": [
{
"name": "get_alerts",
"description": "Get weather alerts for a US state.\n\nArgs:\n state: Two-letter US state code (e.g. CA, NY)\n",
"inputSchema": {
"properties": {
"state": {
"title": "State",
"type": "string"
}
},
"required": [
"state"
],
"title": "get_alertsArguments",
"type": "object"
}
},
{
"name": "get_forecast",
"description": "Get weather forecast for a location In US.\n\nArgs:\n latitude: Latitude of the location\n longitude: Longitude of the location\n",
"inputSchema": {
"properties": {
"latitude": {
"title": "Latitude",
"type": "number"
},
"longitude": {
"title": "Longitude",
"type": "number"
}
},
"required": [
"latitude",
"longitude"
],
"title": "get_forecastArguments",
"type": "object"
}
}
]
}
}
また、初期化を実施する前に tools/list や実際の tools/call を実施すると、 RuntimeError: Received request before initialization was complete が発生します。
これを見ると全体で一度だけ実行しておきたい初期化の処理を実装しておくのが良いことがわかりました。
結論
mcp クライアントとサーバー間の通信を実際に見てみて、mcp クライアントがサーバーの機能を利用するまでの流れを把握することができました。
これまで、設定して利用するだけだったため、内部ではどのようなことが行われているのだろうかと興味本位で調べてみたのですが、
今回の調査のおかげで mcp サーバーを利用するクライアントを自分で作成する際や、mcp サーバーを利用した際に想定した通り動かない際などの調査としてできることが増えたように思われます。
また、mcp サーバーを Agent に組み込むことで高度な Agent を作成し、それらの Agent を組み合わせて利用するための A2A プロトコルと考えると、Google が A2A プロトコルが mcp を補完するということの意味なのかなと理解しました。
今後、さらに A2A プロトコルについても詳細を調査していきたいと思います。
※ データ分析、データ基盤構築、及び AI 活用に関するご相談は、以下よりお気軽にお問い合わせください。
お問い合わせフォーム
Discussion