快速将聚合数据API转为MCP Server 教程
本教程将带你一步步将一个聚合数据API(或其他任意API)快速转为MCP Server。我们以聚合数据的天气预报查询API为例,详细讲解API to MCP的完整流程。
方法一:使用 Python FastMCP SDK
GitHub 地址:https://github.com/modelcontextprotocol/python-sdk
1. 安装项目环境
本项目基于 Python,推荐使用官方推荐的 Python 环境管理工具 uv。
uv 是 MCP 官方推荐的 Python 项目管理工具,安装方法请参考其官方文档。
2. 创建项目
使用 uv 初始化项目,例如项目名为 juhe-weather-mcp
,在终端执行:
uv init juhe-weather-mcp
执行后会生成如下目录结构:
- .python-version
- main.py
- pyproject.toml
- README.md
3. 安装 MCP 依赖
进入项目目录,安装 MCP SDK 依赖:
cd juhe-weather-mcp
uv add "mcp[cli]"
安装完成后,pyproject.toml
文件的依赖配置中会自动添加:
dependencies = [
"mcp[cli]>=1.9.4",
]
如果依赖安装缓慢或超时,可切换为国内镜像源。在
pyproject.toml
文件末尾添加:[tool.uv] index-url = "https://mirrors.aliyun.com/pypi/simple/"
4. 编写 MCP Server 代码
环境准备就绪后,开始编写第一个 MCP Server。打开 main.py
文件,按如下步骤操作:
a. 引入 FastMCP 包
from mcp.server.fastmcp import FastMCP
b. 创建 MCP Server 实例
mcp = FastMCP("Juhe Weather MCP Server")
c. 定义天气查询工具
@mcp.tool()
def query_weather(city: str) -> str:
"""
查询指定城市的天气信息
"""
return f"天气信息:{city}"
d. 添加服务启动代码
使用 argparse 解析命令行参数,支持多种启动方式(sse、stdio、streamable-http):
if __name__ == "__main__":
import argparse
parser = argparse.ArgumentParser(description="Juhe Weather MCP Server")
parser.add_argument("--transport", type=str, default="streamable-http", choices=["sse", "stdio", "streamable-http"], help="Transport type")
args = parser.parse_args()
mcp.run(transport=args.transport)
e. 完整 main.py 示例
from mcp.server.fastmcp import FastMCP
import os
import argparse
mcp = FastMCP("Juhe Weather MCP Server")
@mcp.tool()
def query_weather(city: str) -> str:
"""
查询指定城市的天气信息
"""
return f"天气信息:{city}"
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="Juhe Weather MCP Server")
parser.add_argument("--transport", type=str, default="streamable-http", choices=["sse", "stdio", "streamable-http"], help="Transport type")
args = parser.parse_args()
mcp.run(transport=args.transport)
5. 启动 MCP 服务
以 streamable-http 方式启动 MCP 服务:
uv run main.py --transport streamable-http
启动成功后,命令行会输出如下内容:
INFO: Started server process [78231]
INFO: Waiting for application startup.
[06/13/25 14:20:25] INFO StreamableHTTP session manager started
INFO: Application startup complete.
INFO: Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
此时,MCP 服务已成功启动。
6. 测试 MCP 服务
推荐使用 CherryStudio 工具进行测试,也可用其他支持 MCP 协议的大模型工具。
在 CherryStudio 设置中,添加 MCP 服务器:
- 名称:聚合天气MCP
- 类型:streamableHTTP
- URL: http://127.0.0.1:8000/mcp
保存后,在工具 Tab 可见 query_weather
工具。
回到对话页面,选择刚创建的 MCP 服务器,输入:
苏州天气
即可自动调用 MCP 工具。
7. 对接聚合天气API
目前 MCP Server 返回的是模拟数据,接下来对接真实的聚合天气API。
聚合天气API文档:https://www.juhe.cn/docs/api/id/73
只需在工具方法中发起网络请求,获取真实天气数据。示例代码如下:
from mcp.server.fastmcp import FastMCP
import os
import argparse
from pydantic import Field
import httpx
mcp = FastMCP("Juhe Weather MCP Server")
@mcp.tool(description="查询指定城市的天气信息")
async def query_weather(city: str = Field(description="城市名称,如:苏州")) -> str:
"""
查询指定城市的天气信息
参数:
- city: 城市名称,如:苏州
"""
url = f"http://apis.juhe.cn/simpleWeather/query?city={city}&key=**********" # 请替换为你的聚合API Key
try:
async with httpx.AsyncClient() as client:
response = await client.get(url)
response.raise_for_status()
data = response.json()
return data
except Exception as e:
return f"查询天气信息失败: {str(e)}"
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="Juhe Weather MCP Server")
parser.add_argument("--transport", type=str, default="streamable-http", choices=["sse", "stdio", "streamable-http"], help="Transport type")
args = parser.parse_args()
mcp.run(transport=args.transport)
重启 MCP 服务后,在 CherryStudio 再次发起对话:
苏州的天气如何啊?
即可获取苏州的实时天气数据。
其他启动方式:
# sse uv run main.py --transport sse # stdio uv run main.py --transport stdio
方法二:一键开通聚合MCP
如果你希望更省事,可以直接前往聚合MCP中心,点击开通即可获得专属的 MCP Server。
访问地址:https://juhe.cn/mcp