一、引言
在 AI Agent 生态蓬勃发展的 2025 年,MCP协议成为连接不同智能体的技术桥梁。作为开发者,我一直对这个能够实现 AI 模型间无缝通信的协议充满好奇。通过官方文档的学习,我了解到 MCP 服务端不仅是协议的核心载体,更是构建智能体协作网络的基石。为了深入理解其工作原理,我决定按照官方文档从零开始搭建一个基于天气查询的 MCP 服务,并在客户端调用过程中经历了一段充满挑战的技术探索之旅。
二、环境搭建
工欲善其事,必先利其器。根据官方指南,我首先完成了以下环境配置:
首先是安装uv,如果有Python环境的话很简单,一句话就搞定了。
pip install uv
三、编写MCP服务代码
1、基础方法
先导入依赖,编写发送请求的基础方法以及格式化天气信息的方法
from typing import Anyimport httpxfrom mcp.server.fastmcp import FastMCPmcp = FastMCP("weather")NWS_API_BASE = "https://api.weather.gov"USER_AGENT = "weather-app/1.0"async def make_nws_request(url: str) -> dict[str, Any] | None: """ 发送一个请求到NWS API,并进行适当的错误处理。 """ headers = { "User-Agent": USER_AGENT, "Accept": "application/geo+json", } async with httpx.AsyncClient() as client: try: response = await client.get(url, headers=headers, timeout=60.0) response.raise_for_status() return response.json() except httpx.RequestError as e: print(f"Request error: {e}") except httpx.HTTPStatusError as e: print(f"HTTP error: {e}") except Exception as e: print(f"Unexpected error: {e}") return Nonedef format_alert(feature: dict) -> str: """ 将天气警报信息格式化为一个可读的字符串。 """ props = feature["properties"] event = props.get("event", "未知事件") area = props.get("areaDesc", "未知区域") severity = props.get("severity", "未知严重性") description = props.get("description", "无描述") instructions = props.get("instruction", "无说明") return f""" 事件: {event} 区域: {area} 严重性: {severity} 描述: {description} 说明: {instructions} """
2、mcp接口
然后编写两个mcp接口方法,get_alerts
获取天气警报,get_forecast
获取天气预报,注意要用mcp.tool
装饰。
@mcp.tool()async def get_alerts(state: str) -> str: """ 获取指定州的天气警报。 """ url = f"{NWS_API_BASE}/alerts/active/area={state}" data = await make_nws_request(url) if data and "features" in data: alerts = [format_alert(feature) for feature in data["features"]] return "\n---\n".join(alerts) if alerts else "没有天气警报。" return "无法获取天气警报。"@mcp.tool()async def get_forecast(latitude: float, longitude: float) -> str: """ 获取指定经纬度的天气预报。 """ url = f"{NWS_API_BASE}/points/{latitude},{longitude}" data = await make_nws_request(url) if data and "properties" in data: forecast_url = data["properties"]["forecast"] forecast_data = await make_nws_request(forecast_url) if forecast_data and "properties" in forecast_data: periods = forecast_data["properties"]["periods"] forecasts = [] for period in periods[:5]: forecast = f""" 时间: {period['name']} 温度: {period['temperature']}°{period['temperatureUnit']} 风向: {period['windDirection']} 风速: {period['windSpeed']} 详细: {period['detailedForecast']} """ forecasts.append(forecast) return "\n---\n".join(forecasts) if forecasts else "没有天气预报。" return "无法获取天气预报。"
3、完整代码
from typing import Anyimport httpxfrom mcp.server.fastmcp import FastMCPmcp = FastMCP("weather")NWS_API_BASE = "https://api.weather.gov"USER_AGENT = "weather-app/1.0"async def make_nws_request(url: str) -> dict[str, Any] | None: """ 发送一个请求到NWS API,并进行适当的错误处理。 """ headers = { "User-Agent": USER_AGENT, "Accept": "application/geo+json", } async with httpx.AsyncClient() as client: try: response = await client.get(url, headers=headers, timeout=60.0) response.raise_for_status() return response.json() except httpx.RequestError as e: print(f"Request error: {e}") except httpx.HTTPStatusError as e: print(f"HTTP error: {e}") except Exception as e: print(f"Unexpected error: {e}") return Nonedef format_alert(feature: dict) -> str: """ 将天气警报信息格式化为一个可读的字符串。 """ props = feature["properties"] event = props.get("event", "未知事件") area = props.get("areaDesc", "未知区域") severity = props.get("severity", "未知严重性") description = props.get("description", "无描述") instructions = props.get("instruction", "无说明") return f""" 事件: {event} 区域: {area} 严重性: {severity} 描述: {description} 说明: {instructions} """@mcp.tool()async def get_alerts(state: str) -> str: """ 获取指定州的天气警报。 """ url = f"{NWS_API_BASE}/alerts/active/area={state}" data = await make_nws_request(url) if data and "features" in data: alerts = [format_alert(feature) for feature in data["features"]] return "\n---\n".join(alerts) if alerts else "没有天气警报。" return "无法获取天气警报。"@mcp.tool()async def get_forecast(latitude: float, longitude: float) -> str: """ 获取指定经纬度的天气预报。 """ url = f"{NWS_API_BASE}/points/{latitude},{longitude}" data = await make_nws_request(url) if data and "properties" in data: forecast_url = data["properties"]["forecast"] forecast_data = await make_nws_request(forecast_url) if forecast_data and "properties" in forecast_data: periods = forecast_data["properties"]["periods"] forecasts = [] for period in periods[:5]: forecast = f""" 时间: {period['name']} 温度: {period['temperature']}°{period['temperatureUnit']} 风向: {period['windDirection']} 风速: {period['windSpeed']} 详细: {period['detailedForecast']} """ forecasts.append(forecast) return "\n---\n".join(forecasts) if forecasts else "没有天气预报。" return "无法获取天气预报。"if __name__ == "__main__": mcp.run(transport="stdio")
四、客户端调用
Claude桌面应用当然是我们的首选,但在安装登录后,很不幸,由于地区限制,我的账号无法使用。
接下来我尝试了最近一直在用的AI工具Cherry Studio,按照文档我完成了mcp配置,但却一直提示启动失败,我在命令行中执行了启动mcp服务的命令,发现uvx不支持直接运行py文件,而且Cherry Studio不支持uv命令。
最后我只能尝试一下一直在用的AI代码编辑器Cursor。
Cursor添加MCP服务非常简单,就是编写一个mcp的配置文件。
{ "mcpServers": { "bfgDcOhO_98WWPcPd3HXu": { "name": "MCP 服务器 - demo", "description": "mcp demo ", "isActive": false, "command": "uv", "args": [ "run", "/Users/rainstop_3/mcp-demo/weather/weather.py" ] } }}
这次终于成功了!赶快试一下,可以清楚地看到回复中调用了MCP tool,第一次调用的是get_alerts接口,没有返回结果,第二次调用了get_forecast接口,返回了天气信息。
五、总结
总体上来说MCP并不难理解,也比较容易上手,可以想象到未来很多接口都会以MCP的方式开放共各种AI智能体来调用,这将成为产品的一种新形态。