環境配置煩請移步上一篇博客
這里直接步入主題,天氣查詢的api用的是openweather,免費注冊就可以使用了 每天1000次內使用時免費的,大概的api 如下
https://api.openweathermap.org/data/2.5/weather?q=Beijing,cn&APPID=注冊后可以拿到一個appid
這里如果切換別的查詢工具都是可以的
返回結果如下
{"coord": {"lon": 116.3972,"lat": 39.9075},"weather": [{"id": 800,"main": "Clear","description": "clear sky","icon": "01n"}],"base": "stations","main": {"temp": 295.09,"feels_like": 294.36,"temp_min": 295.09,"temp_max": 295.09,"pressure": 1012,"humidity": 39,"sea_level": 1012,"grnd_level": 1007},"visibility": 10000,"wind": {"speed": 3.9,"deg": 198,"gust": 10.01},"clouds": {"all": 1},"dt": 1748445454,"sys": {"type": 1,"id": 9609,"country": "CN","sunrise": 1748379012,"sunset": 1748432008},"timezone": 28800,"id": 1816670,"name": "Beijing","cod": 200
}
那我們對應的mcp server代碼如下
import json
import httpx
from typing import Any
from mcp.server.fastmcp import FastMCP# 初始化 MCP 服務器
mcp = FastMCP("WeatherServer")# OpenWeather API 配置
OPENWEATHER_API_BASE = "https://api.openweathermap.org/data/2.5/weather"
API_KEY = "xxxxx" # 請替換為你自己的 OpenWeather API Key
USER_AGENT = "weather-app/1.0"async def fetch_weather(city: str) -> dict[str, Any] | None:"""從 OpenWeather API 獲取天氣信息。:param city: 城市名稱(需使用英文,如 Beijing):return: 天氣數據字典;若出錯返回包含 error 信息的字典"""params = {"q": city,"appid": API_KEY}headers = {"User-Agent": USER_AGENT}async with httpx.AsyncClient() as client:try:response = await client.get(OPENWEATHER_API_BASE, params=params, headers=headers, timeout=30.0)response.raise_for_status()return response.json() # 返回字典類型except httpx.HTTPStatusError as e:return {"error": f"HTTP 錯誤: {e.response.status_code}"}except Exception as e:return {"error": f"請求失敗: {str(e)}"}def format_weather(data: dict[str, Any] | str) -> str:"""將天氣數據格式化為易讀文本。:param data: 天氣數據(可以是字典或 JSON 字符串):return: 格式化后的天氣信息字符串"""# 如果傳入的是字符串,則先轉換為字典if isinstance(data, str):try:data = json.loads(data)except Exception as e:return f"無法解析天氣數據: {e}"# 如果數據中包含錯誤信息,直接返回錯誤提示if "error" in data:return f"?? {data['error']}"# 提取數據時做容錯處理city = data.get("name", "未知")country = data.get("sys", {}).get("country", "未知")temp = data.get("main", {}).get("temp", "N/A")humidity = data.get("main", {}).get("humidity", "N/A")wind_speed = data.get("wind", {}).get("speed", "N/A")# weather 可能為空列表,因此用 [0] 前先提供默認字典weather_list = data.get("weather", [{}])description = weather_list[0].get("description", "未知")return (f"🌍 {city}, {country}\n"f"🌡 溫度: {temp}°C\n"f"💧 濕度: {humidity}%\n"f"🌬 風速: {wind_speed} m/s\n"f"🌤 天氣: {description}\n")@mcp.tool()
async def query_weather(city: str) -> str:"""輸入指定城市的英文名稱,返回今日天氣查詢結果。:param city: 城市名稱(需使用英文):return: 格式化后的天氣信息"""data = await fetch_weather(city)return format_weather(data)if __name__ == "__main__":# 以標準 I/O 方式運行 MCP 服務器mcp.run()
結果如下