从零开始构建MCP服务器：让Claude获得精准的地理位置天气查询能力
引言：精确到坐标的天气查询
传统的天气查询通常依赖城市名称，但这种方式存在不少问题：同名城市容易混淆、城区与郊区气象差异明显，而且具体位置往往无法精确匹配。
相比之下，直接通过经纬度坐标获取天气信息，能大幅提升查询的精度，尤其适用于需要定位到街道、园区，甚至具体建筑物的场景。
今天我将带你构建一个基于地理坐标的MCP天气服务器，让Claude能够根据精确的经纬度位置提供准确的天气预报和实时天气信息。
要接入国内的天气预报api，经过调研后我选择了和风天气，有免费额度还不错。
效果如下：
为什么使用经纬度查询？
精确定位：经纬度坐标可以精确定位到具体位置，避免同名地区的混淆。
灵活性强：支持任意地理位置查询，不局限于预定义的城市列表。
数据准确性：基于精确坐标的天气数据更能反映查询位置的实际情况。
国际化支持：经纬度是通用的地理坐标系统，便于扩展到全球范围。
核心架构设计
项目初始化
uv init weather-coordinate-mcp cd weather-coordinate-mcp uv venv source .venv/bin/activate # Windows: .venv\Scripts\activate uv add "mcp[cli]" httpx python-dotenv 服务基础架构
import os import re from typing import Any, Optional, Tuple import httpx from mcp.server.fastmcp import FastMCP from dotenv import load_dotenv # 加载环境变量 load_dotenv() # 初始化FastMCP服务器 mcp = FastMCP("qweather-coordinate") # 配置常量 QWEATHER_API_KEY = os.getenv("QWEATHER_API_KEY") GEO_API_BASE = "https://geoapi.qweather.com/v2" WEATHER_API_BASE = "https://devapi.qweather.com/v7" USER_AGENT = "qweather-coordinate-mcp/1.0" if not QWEATHER_API_KEY: raise ValueError("请在.env文件中设置QWEATHER_API_KEY") 地理坐标处理核心功能
坐标验证与格式化
def validate_and_format_coordinates(lat: float, lon: float) -> Optional[str]: """验证并格式化经纬度坐标 Args: lat: 纬度 (-90 到 90) lon: 经度 (-180 到 180) Returns: 格式化的坐标字符串或None（如果坐标无效） """ # 验证纬度范围 if not -90 <= lat <= 90: return None # 验证经度范围 if not -180 <= lon <= 180: return None # 格式化为两位小数 return f"{lon:.2f},{lat:.2f}" def parse_location_input(location: str) -> Optional[str]: """解析位置输入，支持多种格式 支持的格式： - "116.41,39.92" - "116.41, 39.92" - "经度116.41纬度39.92" - "北京天安门" (通过地名转换为坐标) """ # 尝试匹配数字坐标格式 coord_pattern = r'(-?\d+.?\d*),\s*(-?\d+.?\d*)' match = re.search(coord_pattern, location.strip()) if match: try: lon, lat = float(match.group(1)), float(match.group(2)) return validate_and_format_coordinates(lat, lon) except ValueError: return None # 如果不是坐标格式，则通过地名查询转换 return None async def location_name_to_coordinates(location_name: str) -> Optional[str]: """通过地名获取坐标""" url = f"{GEO_API_BASE}/city/lookup" params = {"location": location_name, "key": QWEATHER_API_KEY} async with httpx.AsyncClient() as client: try: response = await client.get(url, params=params, timeout=30.0) response.raise_for_status() data = response.json() if data.get("code") == "200" and data.get("location"): first_location = data["location"][0] lat, lon = float(first_location["lat"]), float(first_location["lon"]) return validate_and_format_coordinates(lat, lon) except Exception: pass return None HTTP请求封装
async def make_qweather_request(url: str, params: dict) -> Optional[dict[str, Any]]: """向和风天气API发送请求""" headers = {"User-Agent": USER_AGENT} params["key"] = QWEATHER_API_KEY async with httpx.AsyncClient() as client: try: response = await client.get(url, params=params, headers=headers, timeout=30.0) response.raise_for_status() data = response.json() if data.get("code") != "200": print(f"和风天气API错误: {data.get('code')}") return None return data except Exception as e: print(f"请求失败: {e}") return None 核心天气查询工具实现
天气预报查询
@mcp.tool() async def get_weather_forecast(location: str, days: int = 3) -> str: """根据地理位置获取天气预报 Args: location: 地理位置，支持以下格式： - 经纬度坐标："116.41,39.92" - 地名："北京"、"上海浦东" days: 预报天数，支持3、7、10、15、30天 """ # 解析位置信息 coordinates = parse_location_input(location) if not coordinates: # 尝试通过地名转换为坐标 coordinates = await location_name_to_coordinates(location) if not coordinates: return f"无法解析位置信息：{location}\n请使用经纬度格式（如：116.41,39.92）或有效的地名" # 根据天数选择API端点 if days <= 3: endpoint = "3d" elif days <= 7: endpoint = "7d" elif days <= 10: endpoint = "10d" elif days <= 15: endpoint = "15d" else: endpoint = "30d" days = min(days, 30) # 最多支持30天 url = f"{WEATHER_API_BASE}/weather/{endpoint}" params = { "location": coordinates, "lang": "zh" # 使用中文 } data = await make_qweather_request(url, params) if not data or not data.get("daily"): return f"无法获取坐标 {coordinates} 的天气预报信息" return format_detailed_forecast(data, coordinates, days) def format_detailed_forecast(data: dict, coordinates: str, days: int) -> str: """格式化详细天气预报""" daily_forecasts = data["daily"][:days] update_time = data.get("updateTime", "未知") result = f"🌍 坐标位置: {coordinates}\n" result += f"📅 {days}天天气预报\n" result += f"🕒 更新时间: {update_time}\n" result += f"🔗 详细信息: {data.get('fxLink', '')}\n\n" for i, day in enumerate(daily_forecasts): day_label = get_day_label(i) result += f"📅 {day_label} ({day['fxDate']})\n" result += f"🌅 日出: {day.get('sunrise', '无')} | 🌇 日落: {day.get('sunset', '无')}\n" result += f"🌙 月升: {day.get('moonrise', '无')} | 🌑 月落: {day.get('moonset', '无')}\n" result += f"🌕 月相: {day.get('moonPhase', '无')}\n" result += f"🌡️ 温度: {day['tempMin']}°C ~ {day['tempMax']}°C\n" result += f"☀️ 白天: {day['textDay']} | 🌙 夜间: {day['textNight']}\n" # 风力信息 result += f"💨 白天风况: {day['windDirDay']} {day['windScaleDay']}级 ({day['windSpeedDay']}km/h)\n" result += f"💨 夜间风况: {day['windDirNight']} {day['windScaleNight']}级 ({day['windSpeedNight']}km/h)\n" # 详细气象数据 result += f"💧 湿度: {day['humidity']}%\n" result += f"🌧️ 降水量: {day['precip']}mm\n" result += f"🔽 气压: {day['pressure']}hPa\n" result += f"👁️ 能见度: {day['vis']}km\n" result += f"☁️ 云量: {day['cloud']}%\n" result += f"☀️ 紫外线指数: {day['uvIndex']}\n\n" # 添加数据来源信息 if data.get("refer"): sources = data["refer"].get("sources", []) result += f"📊 数据来源: {', '.join(sources)}\n" return result def get_day_label(day_index: int) -> str: """获取日期标签""" if day_index == 0: return "今天" elif day_index == 1: return "明天" elif day_index == 2: return "后天" else: return f"{day_index + 1}天后" 实时天气查询
@mcp.tool() async def get_current_weather(location: str) -> str: """获取实时天气信息 Args: location: 地理位置，支持经纬度坐标或地名 """ # 解析位置信息 coordinates = parse_location_input(location) if not coordinates: coordinates = await location_name_to_coordinates(location) if not coordinates: return f"无法解析位置信息：{location}" url = f"{WEATHER_API_BASE}/weather/now" params = { "location": coordinates, "lang": "zh" } data = await make_qweather_request(url, params) if not data or not data.get("now"): return f"无法获取坐标 {coordinates} 的实时天气信息" return format_current_weather(data, coordinates) def format_current_weather(data: dict, coordinates: str) -> str: """格式化实时天气数据""" now = data["now"] update_time = data.get("updateTime", "未知") result = f"🌍 坐标位置: {coordinates}\n" result += f"🕒 观测时间: {now.get('obsTime', update_time)}\n\n" result += f"🌡️ 当前温度: {now['temp']}°C\n" result += f"🌡️ 体感温度: {now['feelsLike']}°C\n" result += f"🌤️ 天气状况: {now['text']}\n" result += f"💨 风向风力: {now['windDir']} {now['windScale']}级\n" result += f"🌪️ 风速: {now['windSpeed']}km/h\n" result += f"💧 相对湿度: {now['humidity']}%\n" result += f"🌧️ 小时降水: {now['precip']}mm\n" result += f"🔽 大气压强: {now['pressure']}hPa\n" result += f"👁️ 能见度: {now['vis']}km\n" result += f"☁️ 云量: {now['cloud']}%\n" # 添加数据来源 if data.get("refer"): sources = data["refer"].get("sources", []) result += f"\n📊 数据来源: {', '.join(sources)}" return result 坐标转换与地理信息查询
@mcp.tool() async def search_coordinates(location_name: str) -> str: """根据地名搜索地理坐标 Args: location_name: 地名（支持中英文） """ url = f"{GEO_API_BASE}/city/lookup" params = { "location": location_name, "key": QWEATHER_API_KEY, "range": "world" # 全球范围搜索 } data = await make_qweather_request(url, params) if not data or not data.get("location"): return f"未找到 '{location_name}' 的地理位置信息" locations = data["location"] result = f"🔍 搜索 '{location_name}' 找到 {len(locations)} 个结果：\n\n" for i, loc in enumerate(locations[:8]): # 最多显示8个结果 result += f"{i+1}. 📍 {loc['name']}\n" result += f" 🌏 完整路径: {loc['country']} > {loc['adm1']} > {loc['adm2']}\n" result += f" 📍 经纬度: {loc['lon']},{loc['lat']}\n" result += f" 🆔 位置ID: {loc['id']}\n" result += f" 🕒 时区: {loc['tz']}\n\n" return result @mcp.tool() async def get_weather_by_coordinates(latitude: float, longitude: float, days: int = 3) -> str: """直接通过经纬度获取天气预报 Args: latitude: 纬度 (-90 到 90) longitude: 经度 (-180 到 180) days: 预报天数 (1-30) """ coordinates = validate_and_format_coordinates(latitude, longitude) if not coordinates: return f"无效的坐标：纬度 {latitude}, 经度 {longitude}\n纬度范围：-90到90，经度范围：-180到180" return await get_weather_forecast(coordinates, days) 服务器配置与使用
启动服务器
if __name__ == "__main__": print(f"启动和风天气坐标查询MCP服务器...") print(f"支持的功能：") print(f" - 基于坐标的天气预报查询") print(f" - 实时天气信息获取") print(f" - 地名到坐标转换") print(f" - 直接坐标天气查询") mcp.run(transport='stdio') Claude桌面版配置
{ "mcpServers": { "qweather-coordinate": { "command": "uv", "args": [ "--directory", "/your/absolute/path/to/weather-coordinate-mcp", "run", "weather.py" ], "env": { "QWEATHER_API_KEY": "your_qweather_api_key" } } } } 实际应用场景
精确位置天气查询
"请查询经纬度116.41,39.92（天安门广场）的天气情况"
地名转坐标查询
"西安今天天气如何？"
直接坐标调用
"请查询北纬31.23、东经121.47位置未来7天的天气预报"
旅行规划支持
"我要去几个城市旅行，请分别查询以下城市的天气：（北京）、（上海）、（深圳）"
优化建议与扩展思路
坐标处理优化
坐标精度控制：根据查询需求自动调整坐标精度，城市级查询使用较低精度，具体地点使用高精度。
坐标缓存策略：对地名到坐标的转换结果进行缓存，减少重复查询。
功能扩展
批量查询：支持一次查询多个坐标点的天气信息，便于路线规划。
天气对比：提供多个位置的天气对比功能，帮助用户做出决策。
历史天气：结合历史天气API，提供某个坐标位置的历史天气趋势分析。
总结
基于经纬度坐标的天气查询系统为用户提供了更加精确和灵活的天气信息获取方式。通过直接使用地理坐标，我们避免了地名歧义问题，同时支持任意位置的天气查询，这对于户外活动、物流运输、农业生产等需要精确位置天气信息的场景特别有价值。
这个项目展示了MCP协议的一大优势：它不仅能连接外部专业API，还能整合出一个高可用、易扩展的服务。我们通过清晰的架构和合理的错误处理，搭建了一个稳定好用的天气查询系统。更重要的是，这只是开始。未来我们还可以接入更多功能，比如查询空气质量、获取气象预警，甚至结合地图、路径规划系统，一步步打造成一个真正的“地理智能中枢”。让AI不只“看天气”，还能“看懂环境”。
