はじめに
前回は、OpenAIが公開したAgents SDKを用いてお天気エージェントを作成した。今回は、MCPを実現するためのPython SDKを用いてお天気サーバを作成し、Claude Desktopからそのサーバに問い合わせる仕組みを作る（Claude DesktopとはClaudeのデスクトップアプリである）。
MCPとは
MCP（Model Context Protocol）は、大規模言語モデル（LLM）から外部ツールを呼び出すためのインタフェース仕様である。このMCPは、2024年にAnthropic（Claudeを開発している企業）が最初に提唱した。その後、多くのLLMベンダーが賛同し、2025年にはOpenAIとGoogleもMCPへの参加を表明し、標準化の流れが加速することになった。つまり、MCPに従うインタフェースを持つサーバ（MCPサーバ）を一度作れば、さまざまなLLMと接続できるようになるということである。今回は、簡単なお天気サーバをMCPサーバとして作成し、実際にClaude Desktopから呼び出してみる。
準備
pyenvとpoetryを用いてPythonの実行環境を作成した。使用したPythonのバージョンは3.12.0、その他に必要なライブラリのインストール手順は以下のとおりである。
|
1 2 3 |
poetry add mcp[cli] poetry add httpx poetry add python-dotenv |
最初の命令で、MCPとMCP CLIをインストールしている。MCP CLIはMCPサーバの開発を効率化するツールであるが、今回のお天気サーバの開発には使用しなかった。
アーキテクチャ
今回作成するお天気サービスのアーキテクチャを以下に示す。
天気予報の情報取得には、OpenWeather APIを使う。OpenWeatherは、世界中の都市や地域の天気情報と気象データを提供するAPIサービスである（前回のブログでも使用した）。上図のようにローカル上にMCPサーバを作り、これをClaude Desktopと接続する。また、実行環境はWindows11である。
お天気サーバの実装
Pythonコードは以下の通り。
|
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 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 |
import logging import os from datetime import datetime from typing import Any, Dict import httpx from dotenv import load_dotenv from mcp.server.fastmcp import FastMCP # 環境変数の読み込み load_dotenv() # ログファイルの絶対パス設定 log_directory = os.path.dirname(os.path.abspath(__file__)) log_filepath = os.path.join(log_directory, "weather_service.log") # ログの準備 logging.basicConfig( level=logging.INFO, filename=log_filepath, filemode="a", format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", ) # logging.basicConfig(level=logging.INFO) logger = logging.getLogger("weather-server") # APIキーの準備 API_KEY = os.getenv("OPENWEATHER_API_KEY") if not API_KEY: raise ValueError("OPENWEATHER_API_KEY environment variable required") # 定数の準備 API_BASE_URL = "http://api.openweathermap.org/data/2.5" DEFAULT_CITY = "Tokyo" CURRENT_WEATHER_ENDPOINT = "weather" FORECAST_ENDPOINT = "forecast" # MCPサーバーの作成 mcp = FastMCP("weather_service") # OpenWeatherで天気情報を取得 async def fetch_weather(city: str) -> Dict[str, Any]: """OpenWeather APIから現在の天気情報を取得する""" async with httpx.AsyncClient() as client: response = await client.get( f"{API_BASE_URL}/{CURRENT_WEATHER_ENDPOINT}", params={"q": city, "appid": API_KEY, "units": "metric"}, # 摂氏温度を使用 ) response.raise_for_status() return response.json() # OpenWeatherで天気予報を取得 async def fetch_forecast(city: str, days: int = 3) -> Dict[str, Any]: """OpenWeather APIから天気予報を取得する""" async with httpx.AsyncClient() as client: response = await client.get( f"{API_BASE_URL}/{FORECAST_ENDPOINT}", params={ "q": city, "appid": API_KEY, "units": "metric", # 摂氏温度を使用 "cnt": min(days * 8, 40), # 3時間ごとのデータ、1日8回 }, ) response.raise_for_status() return response.json() # 天気データの整形 def format_weather_data(data: Dict[str, Any]) -> str: """天気データを読みやすいテキスト形式にフォーマットする""" city_name = data.get("name", "Unknown") temp = data["main"]["temp"] conditions = data["weather"][0]["description"] humidity = data["main"]["humidity"] wind_speed = data["wind"]["speed"] return ( f"Current weather in {city_name}:\n" f"Temperature: {temp}°C\n" f"Conditions: {conditions}\n" f"Humidity: {humidity}%\n" f"Wind Speed: {wind_speed} m/s" ) # 天気予報データの整形 def format_forecast_data(data: Dict[str, Any]) -> str: """天気予報データを読みやすいテキスト形式にフォーマットする""" city_name = data.get("city", {}).get("name", "Unknown") # 日付ごとにデータをグループ化 daily_forecasts = {} for item in data["list"]: date = datetime.fromtimestamp(item["dt"]).date().isoformat() if date not in daily_forecasts: daily_forecasts[date] = {"temp_sum": 0, "temp_count": 0, "conditions": []} daily_forecasts[date]["temp_sum"] += item["main"]["temp"] daily_forecasts[date]["temp_count"] += 1 daily_forecasts[date]["conditions"].append(item["weather"][0]["description"]) # 結果のフォーマット result = f"Weather forecast for {city_name}:\n\n" for date, info in daily_forecasts.items(): avg_temp = info["temp_sum"] / info["temp_count"] # 最も頻度の高い天気状態を選択 from collections import Counter most_common_condition = Counter(info["conditions"]).most_common(1)[0][0] result += f"Date: {date}\n" result += f"Temperature: {avg_temp:.1f}°C\n" result += f"Conditions: {most_common_condition}\n\n" return result.strip() @mcp.tool() async def get_weather(city: str = DEFAULT_CITY) -> str: """Get current weather conditions for a specified city. Args: city: The name of the city to get weather for. Defaults to Tokyo. """ try: data = await fetch_weather(city) return format_weather_data(data) except Exception as e: logger.error(f"Error fetching weather for {city}: {e}") return f"Sorry, I couldn't get weather data for {city}. Error: {str(e)}" @mcp.tool() async def get_forecast(city: str = DEFAULT_CITY, days: int = 3) -> str: """Get weather forecast for the next few days for a specified city. Args: city: The name of the city to get forecast for. Defaults to Tokyo. days: Number of days to forecast. Defaults to 3, max is 5. """ try: # 日数を制限 days = min(max(days, 1), 5) data = await fetch_forecast(city, days) return format_forecast_data(data) except Exception as e: logger.error(f"Error fetching forecast for {city}: {e}") return f"Sorry, I couldn't get forecast data for {city}. Error: {str(e)}" if __name__ == "__main__": logger.info("Starting MCP weather server...") mcp.run() |
プログラムの構成は以下の通りである。
＊APIアクセス関数
fetch_weather(): 現在の天気情報を取得する非同期関数
fetch_forecast(): 3日間（デフォルト）の天気予報を取得する非同期関数
＊データ整形関数
format_weather_data(): 現在の天気データを読みやすいテキスト形式に変換
format_forecast_data(): 予報データを日付ごとに集計し、平均温度や代表的な天気状態を表示
＊MCPツール化（外部に公開されるAPI）
get_weather(): 指定した都市の現在の天気を取得して返す関数
get_forecast(): 指定した都市の天気予報を取得して返す関数
Claude Desktopとの接続
Claude Desktopを起動し、左上3本線のマークをクリック、ファイル > 設定を選択する（下図参照）。
開いた窓の「開発者」を選び、「構成を編集」をクリックする（下図参照）。
すると、Windowsのフォルダが開く。このフォルダ内に「claude_desktop_config.json」ファイルを作成し、以下を記載する。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 |
{ "mcpServers": { "weather_service": { "command": "C:\\projects\\mcp_trial\\.venv\\Scripts\\python.exe", "args": [ "C:\\projects\\mcp_trial\\src\\weather_server.py" ], "env": { "OPENWEATHER_API_KEY": "xxx" } } } } |
4行目はpythonコマンドまでの絶対パス、6行目はpythonで実行するファイルまでの絶対パスである。9行目はOpenWeather APIを使うためのアクセスキーである。このあと、Claude Desktopを終了し再起動する。アプリ右上の×印をクリックしてもバックグラウンドで動き続けているので、タスクバーからたどって終了させる（下図参照）。
再起動後、先と同じ設定を開くと以下のようにJSONファイルの内容が反映されていることが分かる（下図参照）。
これは、Claude Desktopが起動するたびに、記載されたPythonスクリプトが実行されMCPサーバが起動状態になることを意味する。Claude DesktopのUIの方に戻ると以下のようにトンカチマークが追加されているはずである（下図参照）。
このトンカチマークをクリックすると2つのツールが提供されていることが表示される（下図参照）。
これらはいずれもソースコードの中に書いた関数名である。1つは天気予報を得る関数、もう一つは現在の天気を得る関数である。以上でClaude Desktopとの接続は完了である。
お天気サーバの呼び出し
Claude Deskto内で「明日の大阪の天気を教えてください」と打つと、お天気サーバを使ってよいか許可を求める窓が開く。
許可をクリックすると以下のように、今回作成したお天気サーバと遣り取りした結果が表示される（下図参照）。
まとめ
前回のブログでは、OpenWeatherという外部サービスをAIが直接利用するエージェントの作り方を紹介した。今回は、MCPサーバを介して外部サービスを利用するAIの作り方を見た。この2つは競合する手法ではなく、それぞれを補完するものである。例えば、「Excelを開いて合計値を出す」のような単発の作業ならMCPで十分であるが、「CVSを読み込んで分析、グラフ化、PowerPointに自動貼り付け」のような複雑な作業にはエージェントが良い。もっと言えば、エージェントが各作業を実現するために、MCPサーバを利用するというパターンも考えられるだろう。これからのAIエンジニアは、AIの中身でなく、それをツールとして使いこなす技量が問われそうである。
