掘金 人工智能 04月02日 09:59
从零搭建 MCP 服务的体验之旅
index_new5.html
../../../zaker_core/zaker_tpl_static/wap/tpl_guoji1.html

 

本文分享了开发者从零开始搭建基于MCP协议的天气查询服务的经验。文章详细介绍了环境配置、MCP服务代码编写、客户端调用过程,以及在不同AI工具上的尝试。通过实践,作者深入理解了MCP协议的工作原理,并展望了其在AI智能体协作中的应用前景。

🛠️ 作者首先搭建了开发环境,使用`uv`进行依赖安装,为后续MCP服务的构建奠定了基础。

⚙️ 核心在于编写MCP服务代码,包括基础方法和mcp接口。通过定义`get_alerts`和`get_forecast`两个工具,分别用于获取天气警报和预报,并使用`@mcp.tool()`装饰器进行标注。

💻 客户端调用环节,作者尝试了多种AI工具,最终在Cursor代码编辑器中成功配置并调用了MCP服务,实现了天气信息的获取。

💡 总结部分,作者认为MCP协议易于理解和上手,并预见到未来会有更多接口以MCP方式开放,从而促进AI智能体之间的协作。

一、引言

在 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智能体来调用,这将成为产品的一种新形态。

Fish AI Reader

Fish AI Reader

AI辅助创作,多种专业模板,深度分析,高质量内容生成。从观点提取到深度思考,FishAI为您提供全方位的创作支持。新版本引入自定义参数,让您的创作更加个性化和精准。

FishAI

FishAI

鱼阅,AI 时代的下一个智能信息助手,助你摆脱信息焦虑

联系邮箱 441953276@qq.com

相关标签

MCP协议 AI Agent 天气查询 FastMCP AI工具
相关文章