当前位置：首页 > news >正文

如何手搓一个查询天气的mcp server

news 2025/5/30 13:47:49

环境配置烦请移步上一篇博客
这里直接步入主题，天气查询的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()