前回MCPのことを調べてみましたが、MCPを試しに作ってみました記事になります。
MCPサーバーを作ってみる
参考にしたのは、こちらの記事のPython版で、入力された米国の地名の天気を米国の天気APIから取得して返すものです。
この記事の下の方にCluadeにコードを解説してもらったものを置いておきます。
正常に動作すると、このような形でMCPへアクセスし、天気を教えてれます。通常は学習した日付までの情報を持っていませんが、MCPを通してAPIアクセスすることによって、今時点での天気を取得できます。
東京の天気はダメ。
日本語でホノルルの天気を聞いてみるのはOK。
聞き方が悪かったのか、LLMの性質なのか、実はテストしてた時は日本語はNGでした。
回答もまちまち(これはLLMの動作として正しい)なので、この辺りの使い勝手は要検証ですね。
エラーが出た時には
MCPの起動に失敗するなど、エラーが発生した場合、Claudeの起動時にエラーが表示されます。
この時には「Open MCP Settings」から、
「Open Log Folder」を開き、
作成したMCPのログを開きます。
ログを開くとこんな感じ。
何が問題かわからない場合も、便利な世の中になったもので、ログ自体をそのままClaudeにコピー & ペーストすると、原因と解決策が出てきます。
解決策そのものではないかもしれませんが、とっかかりとしてはわかりやすいと思います。
一旦、まとめ
MCPによってローカルファイルや外部API連携が可能になるのでClaudeの幅がかなり広がり、MCPで生成AIアプリがシステム連携することがエコシステムの一つになると感じています。
色々作ってみると面白いので、ぜひトライしてみてください。
以下、Cluadeにコードを解説してもらったもの
Claude DesktopのMCPサーバーの解説
Claude Desktopは、Anthropicが開発したAIアシスタントで、MCP（Modular Command Processor）サーバーを使用してさまざまなタスクを処理します。MCPサーバーは、特定のコマンドを処理するためのモジュール化されたアプローチを提供し、各コマンドは独立して実行されます。
コードの解説
このコードは、天気情報を取得するためのMCPサーバーを実装しています。以下に、各部分の詳細を説明します。
from typing import Any
import httpx
from mcp.server.fastmcp import FastMCP
# Initialize FastMCP server
mcp = FastMCP("weather")
-
typingモジュールは型ヒントを提供します。 -
httpxは非同期HTTPクライアントです。 -
FastMCPはMCPサーバーを初期化するためのクラスです。 -
mcp = FastMCP("weather")で、天気情報を処理するMCPサーバーを初期化します。
NWS_API_BASE = "https://api.weather.gov"
USER_AGENT = "weather-app/1.0"
-
NWS_API_BASEは、米国国立気象局（NWS）のAPIのベースURLです。 -
USER_AGENTは、HTTPリクエストのユーザーエージェントヘッダーに使用されます。
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
-
make_nws_request関数は、NWS APIにリクエストを送信し、エラーハンドリングを行います。 -
httpx.AsyncClientを使用して非同期リクエストを送信します。 - エラーが発生した場合、
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')}
"""
-
format_alert関数は、アラート情報を読みやすい文字列にフォーマットします。
@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)
-
get_alerts関数は、指定された州の天気アラートを取得します。 -
@mcp.tool()デコレーターは、この関数をMCPツールとして登録します。
@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)
-
get_forecast関数は、指定された緯度と経度の天気予報を取得します。 - 予報データを読みやすい形式にフォーマットします。
if __name__ == "__main__":
# Initialize and run the server
mcp.run(transport='stdio')
-
mcp.run(transport='stdio')で、MCPサーバーを標準入出力を使用して実行します。
このコードは、天気情報を取得するための非同期MCPサーバーの基本的な実装を示しています。質問があれば、どうぞお知らせください！
Discussion