前几篇我们给铁矿石智能体加了多轮记忆（Checkpoint）、可视化回滚（Studio）、人工审批（Interrupts）。这一篇换个方向——给它接一个外部能力：通过 MCP（Model Context Protocol） 让大模型学会「查天气」。我们会从零创建一个 MCP 服务，把它接进智能体，并用「上海的天气怎么样」这个测试用例，画出从用户提问 → 大模型决策 → 调用 MCP → 再次调用大模型 → 返回答案的完整时序图。

1. MCP 是什么，为什么要用它

大模型本身不能联网、不能查数据库、不能调用任何外部系统——它只会「生成文本」。要让它「会查天气」，必须给它一个工具（tool）。

那为什么不直接在项目里写个 Python 函数当工具，而要用 MCP？

MCP（Model Context Protocol） 是一套标准协议，用来把「工具 / 数据 / 能力」以统一的方式暴露给任意 AI 应用。它的价值在于解耦：

对比 本地函数工具 MCP 工具
部署 和主程序绑死在一个进程 独立服务，可单独部署、单独重启、单独扩容
复用 只能本项目用 任何支持 MCP 的客户端（Claude、Cursor、你的 agent）都能用
语言 必须同语言 服务端可用任意语言实现
远程 不支持 原生支持远程调用（HTTP）

一句话：MCP 把「能力」做成了可插拔的标准化服务。我们这次创建的天气服务，既能被自己的智能体调用，也能被 Claude、Cursor 等任何 MCP 客户端直接使用。

关于传输方式：远程 MCP 现在的标准是 Streamable HTTP（单一 /mcp 端点，旧的纯 SSE 已废弃）。本文的天气服务就用它——所以即使现在跑在本机，调用方式和「真·远程」完全一致，将来换成云端只需改一个 URL。

2. 整体架构

公网

天气 MCP 服务 进程B

主应用 进程A

持有

Streamable HTTP

地理编码+天气

用户/前端

FastAPI /chat

Deep Agent

DeepSeek 大模型

SQLite 记忆

MCP 客户端
MultiServerMCPClient

FastMCP get_weather

Open-Meteo API

注意这里是两个进程：

  • 进程 A：主应用（智能体 + Web）。
  • 进程 B：天气 MCP 服务（独立运行）。

两者通过 HTTP 通信——这正是 MCP「能力即服务」的体现。

3. 第一步：创建 MCP 服务

我们用官方的 FastMCP（来自 mcp 包）创建服务，只需要 @mcp.tool() 装饰一个函数，它就变成了一个标准 MCP 工具。天气数据用 Open-Meteo（免费、免注册、免 Key）：先把城市名地理编码成经纬度，再查该坐标的实时天气。

3.1 定义工具

@mcp.tool()
async def get_weather(city: str) -> str:
    """根据城市名称查询当前实时天气。

    Args:
        city: 城市名称，支持中文或英文，例如「北京」「上海」「Tokyo」。

    Returns:
        一段中文天气描述（含天气状况、气温、湿度、风速）。
    """
    async with httpx.AsyncClient(timeout=10.0) as client:
        # 1) 地理编码：城市名 → 经纬度
        geo = await client.get(
            "https://geocoding-api.open-meteo.com/v1/search",
            params={"name": city, "count": 1, "language": "zh", "format": "json"},
        )
        geo.raise_for_status()
        results = geo.json().get("results")
        if not results:
            return f"未找到城市「{city}」，请确认名称是否正确。"
        loc = results[0]
        lat, lon = loc["latitude"], loc["longitude"]
        name = loc.get("name", city)
        country = loc.get("country", "")
        admin = loc.get("admin1", "")

        # 2) 查询该坐标的当前天气
        wx = await client.get(
            "https://api.open-meteo.com/v1/forecast",
            params={
                "latitude": lat,
                "longitude": lon,
                "current": "temperature_2m,relative_humidity_2m,weather_code,wind_speed_10m",
                "timezone": "auto",
            },
        )
        wx.raise_for_status()
        current = wx.json().get("current", {})

    code = current.get("weather_code")
    desc = _WEATHER_CODE.get(code, f"未知天气({code})")
    place = "".join(p for p in [country, admin, name] if p)
    return (
        f"{place} 当前天气：{desc}，"
        f"气温 {current.get('temperature_2m')}°C，"
        f"相对湿度 {current.get('relative_humidity_2m')}%，"
        f"风速 {current.get('wind_speed_10m')} km/h。"
        f"（数据来源：Open-Meteo）"
    )

关键点：函数的 docstring 不是注释，是给大模型看的「说明书」。 大模型靠工具名 get_weather、参数 city 的描述、以及这段 docstring 来判断「什么时候该调用它、该传什么参数」。写得越清楚，模型用得越准。

3.2 启动为 Streamable HTTP 服务

# 监听 127.0.0.1:8001，工具端点为 /mcp
mcp = FastMCP("weather", host="127.0.0.1", port=8001)

if __name__ == "__main__":
    # 以 Streamable HTTP 方式运行（远程 MCP 的标准传输）
    mcp.run(transport="streamable-http")

启动命令：

python -m mystu.mcptools.weather_server
# 服务起在 http://127.0.0.1:8001/mcp

至此，一个标准 MCP 服务就建好了，对外暴露一个 get_weather 工具。

4. 第二步：调用 MCP（把远程工具拉到本地）

主应用作为 MCP 客户端，用 langchain-mcp-adapters 的 MultiServerMCPClient 连接 MCP 服务，get_tools() 会自动发现服务端的工具，并包装成 LangChain 工具对象。

async def load_mcp_tools() -> list[Any]:
    """连接远程天气 MCP 服务并加载其工具。

    使用 Streamable HTTP（远程 MCP 标准传输）。失败时返回空列表，
    不阻断应用启动——这样即使天气 MCP 没起，智能体的其他能力仍可用。
    """
    url = get_weather_mcp_url()
    if not url:
        return []
    try:
        from langchain_mcp_adapters.client import MultiServerMCPClient

        client = MultiServerMCPClient(
            {"weather": {"transport": "streamable_http", "url": url}}
        )
        tools = await client.get_tools()
        logger.info("已从远程 MCP 加载工具：%s", [t.name for t in tools])
        return tools
    except Exception as exc:  # noqa: BLE001 - 远程不可用时降级而非崩溃
        logger.warning("加载远程 MCP 工具失败（%s）：%s", url, exc)
        return []

两个设计要点：

  • transport: "streamable_http"：和服务端 mcp.run(transport="streamable-http") 对应。
  • 容错降级：MCP 服务没起时只是返回空列表 + 打日志，不让主应用崩溃——智能体的其他能力（下单、预测）照常可用。

MCP 服务地址由配置提供，默认本地，可用环境变量覆盖（将来换成真远程只改这里）：

# mystu/buildagent/agent/modelConfig.py
def get_weather_mcp_url() -> str:
    """默认本地 8001；可用环境变量 WEATHER_MCP_URL 覆盖。"""
    return _get_str("WEATHER_MCP_URL", "http://127.0.0.1:8001/mcp") or ""

5. 第三步：让大模型「会用」这个工具

光把工具加载进来还不够，要让大模型用上它，需要两件事配合：

5.1 把工具注入智能体

build_agent 把内置工具和 MCP 工具合并后，传给 create_deep_agent：

def build_agent(
    checkpointer: BaseCheckpointSaver | None = None,
    extra_tools: Sequence[Any] | None = None,
):
    """构建使用 DeepSeek 的铁矿石价格预测 deep agent（带多轮对话记忆）。

    checkpointer：对话状态存储器。
      - 不传：使用 InMemorySaver（进程内记忆，重启即丢，适合 CLI / 测试）。
      - 传入 AsyncSqliteSaver 等：把各 thread_id 的对话持久化到磁盘，重启不丢。
    extra_tools：额外注入的工具（如从远程 MCP 加载到的工具），与内置工具合并。
    """
    system_prompt = load_iron_ore_forecast_prompt()
    tools = [*_TOOLS, *(extra_tools or [])]
    return create_deep_agent(
        model=build_model(),
        tools=tools,
        system_prompt=system_prompt,
        interrupt_on=_INTERRUPT_ON,
        checkpointer=checkpointer or InMemorySaver(),
    )

工具加载是异步的（get_tools() 要走网络），而且要和 SQLite 记忆一起在应用启动时准备好，所以放进 FastAPI 的 lifespan：

@asynccontextmanager
async def lifespan(app: FastAPI):
    """应用生命周期：启动时打开 SQLite 持久化记忆，关闭时释放。

    AsyncSqliteSaver 自带异步连接（aiosqlite），必须在事件循环内创建并保持存活，
    因此放在 lifespan 里：进入时建表 + 构建带持久化的 agent，退出时回退默认 agent。
    """
    db_path = get_checkpoint_db_path()
    mcp_tools = await load_mcp_tools()
    async with AsyncSqliteSaver.from_conn_string(db_path) as saver:
        await saver.setup()  # 首次运行自动建表（checkpoints / writes）
        set_agent(build_agent(checkpointer=saver, extra_tools=mcp_tools))
        try:
            yield
        finally:
            set_agent(default_agent)

5.2 在提示词里告诉模型「有这个能力」

工具注入后，模型「能」调用它；但要让它「主动想到」调用，最好在系统提示词里点明：

# 工具能力（节选自 prompt/md/iron_ore_forecast.md）
- 当用户询问某城市的天气时，调用 `get_weather` 工具（参数：城市名 city）查询实时天气，
  并用简洁中文转述结果。该工具来自远程天气 MCP 服务。

大模型决定「调不调工具、调哪个、传什么参数」靠两个信息源：① 工具自身的名字 + 参数 + docstring；② 系统提示词的引导。两者都写清楚，准确率最高。

6. 测试用例：「上海的天气怎么样」全流程时序图

这是本文的核心。当用户问「上海的天气怎么样」时，从提问到回答，大模型被调用了 2 次，中间夹着一次 MCP 调用。完整时序如下：

Open-Meteo 天气 MCP 服务 MCP 客户端 DeepSeek 大模型 SQLite 记忆 Deep Agent 图 FastAPI /chat Open-Meteo 天气 MCP 服务 MCP 客户端 DeepSeek 大模型 SQLite 记忆 Deep Agent 图 FastAPI /chat 状态恢复（读档） 第 1 次调用大模型 —— 决策 执行 MCP 工具（不调模型） 第 2 次调用大模型 —— 总结 状态保存（存档） 用户/前端 "上海的天气怎么样" + thread_id 1 ainvoke(消息, config{thread_id}) 2 aget_tuple(thread_id) 3 历史消息（多轮记忆） 4 系统提示 + 历史 + 用户问题 + 可用工具清单 5 决定调用 get_weather(city="上海") 6 调用工具 get_weather(city=上海) 7 Streamable HTTP 请求 8 ① 地理编码 上海→经纬度 9 lat/lon 10 ② 查当前天气 11 天气数据 12 "上海 当前天气：阴，23.9°C..." 13 工具结果（ToolMessage） 14 历史 + 工具返回结果 15 自然语言答复 16 aput(新 checkpoint) 17 最终回答 18 {status: ok, data: "上海当前阴天，23.9°C..."} 19 用户/前端

这张图里发生了什么（分阶段）

阶段 动作 是否调用大模型
读档 从 SQLite 取该 thread_id 的历史消息 否
决策（第 1 次 LLM） 模型看到「问天气」+ 有 get_weather 工具，决定调用它、并从「上海的天气怎么样」中抽取出参数 city="上海" ✅ 是
执行工具 Agent 通过 MCP 客户端 → 天气服务 → Open-Meteo，拿到真实天气字符串 否
总结（第 2 次 LLM） 模型拿到工具返回的原始天气数据，组织成通顺的中文回答 ✅ 是
存档 把这一轮（用户问 + 工具调用 + 最终回答）写回 SQLite 否

为什么要调两次大模型？

  • 第 1 次：模型自己不知道天气，但它知道该用哪个工具、该传什么参数——这是「决策」。
  • 工具执行只是拿到一段原始数据（“上海 阴 23.9°C…”），不经过模型。
  • 第 2 次：把原始数据交回模型，让它润色成自然语言回答用户——这是「总结」。
  • MCP 调用本身不消耗大模型 token，它只是一次普通 HTTP 请求。

实测结果

用户：上海现在天气怎么样？
回答：上海当前天气：阴天，气温 23.9°C，相对湿度 71%，风速 11.8 km/h。（数据来源：Open-Meteo）
     总体来看体感比较舒适，但湿度偏高，略显闷湿。如果出门，建议带把伞以防转雨。

前半段是 MCP 工具返回的客观数据，后半段是大模型基于数据做的「润色与建议」——正好印证了「两次模型调用」的分工。

7. 本次新增的组件列表及其作用

7.1 新增 / 修改的文件

文件 类型 作用
mystu/mcptools/weather_server.py 新增 天气 MCP 服务：用 FastMCP 暴露 get_weather，Streamable HTTP 运行
mystu/mcptools/__init__.py 新增 让 mcptools 成为可导入的包
mystu/buildagent/agent/deepagent.py 修改 新增 load_mcp_tools()；build_agent 支持 extra_tools 注入
mystu/buildagent/agent/modelConfig.py 修改 新增 get_weather_mcp_url() 读取 MCP 地址
mystu/controller/__init__.py 修改 lifespan 中加载 MCP 工具并注入 agent
mystu/buildagent/prompt/md/iron_ore_forecast.md 修改 增加「查天气调用 get_weather」的能力引导
pyproject.toml 修改 新增依赖 mcp、langchain-mcp-adapters

7.2 引入的核心组件 / 库

组件 来源 角色 作用
FastMCP mcp（官方 SDK） 服务端 快速创建 MCP 服务，@mcp.tool() 把函数变成标准工具，run(transport=...) 选择传输方式
@mcp.tool() mcp 服务端 工具声明装饰器；函数签名 + docstring 自动转成工具的「调用规范」给模型识别
Streamable HTTP MCP 协议 传输层 远程 MCP 的标准传输（单一 /mcp 端点）；本地与远程调用方式一致
MultiServerMCPClient langchain-mcp-adapters 客户端 连接一个或多个 MCP 服务，get_tools() 发现并加载工具
get_tools() langchain-mcp-adapters 客户端 把 MCP 工具包装成 LangChain 工具对象，可直接交给 agent
load_mcp_tools() 项目代码 集成 封装连接 + 加载 + 容错降级
get_weather_mcp_url() 项目代码 配置 提供 MCP 服务地址（默认本地，可用 WEATHER_MCP_URL 覆盖）
build_agent(extra_tools=) 项目代码 集成 把 MCP 工具与内置工具合并注入智能体
lifespan FastAPI 启动 应用启动时异步加载 MCP 工具 + 构建 agent
httpx 第三方 服务端 MCP 服务内部访问 Open-Meteo 的异步 HTTP 客户端
Open-Meteo 外部 API 数据源 免费天气与地理编码数据

7.3 依赖（pyproject.toml 节选）

"mcp>=1.27.2",                    # FastMCP 服务端 + 协议实现
"langchain-mcp-adapters>=0.2.2",  # MultiServerMCPClient 客户端

8. 怎么运行（两个进程）

# 终端 1：先起天气 MCP 服务
uv run python -m mystu.mcptools.weather_server

# 终端 2：再起主应用
uv run python main.py

然后打开网页，问「上海现在天气怎么样？」，即可看到智能体调用远程 MCP 返回的实时天气。

想换成真正的远程 MCP？只需设环境变量 WEATHER_MCP_URL=https://你的-mcp/mcp，业务代码一行都不用改。若服务需要鉴权，在 load_mcp_tools() 的连接配置里加 "headers": {"Authorization": "Bearer xxx"} 即可。

9. 小结

  • MCP = 可插拔的标准化能力服务，把工具/数据从主程序里解耦出来，可独立部署、跨客户端复用。
  • 创建：FastMCP + @mcp.tool() + run(transport="streamable-http") 三步搞定一个远程工具。
  • 调用：客户端用 MultiServerMCPClient + get_tools() 发现并加载工具，包装成 LangChain 工具。
  • 让模型使用：把工具注入 create_deep_agent，并在系统提示词里点明能力；模型靠 docstring + 提示词决定何时调用、传什么参数。
  • 全流程：一次「问天气」= 读档 → 第1次LLM决策 → 调MCP工具 → 第2次LLM总结 → 存档，共调用大模型 2 次，MCP 调用本身不耗模型。

至此，这个智能体已经具备：多轮记忆、可视化回滚、人工审批、以及通过 MCP 接入的外部能力——一套相对完整、可扩展的智能体框架就成型了。下一步想加任何新能力（查数据库、发邮件、查股价），只要再写一个 MCP 服务、配一个 URL 即可，主程序几乎零改动。

# python
Logo
AtomGit开源社区

AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念，把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起，为开发者提供从开发、训练到部署的一站式体验。

更多推荐

cover

WalkingLab 组织入驻 AtomGit，释放高校智能体开源科研成果

avatar AtomGit开源社区
cover

海外动态代理行业彻底换代，AI 数据时代新标准已到来

avatar AtomGit开源社区
cover

全球首个！京东全栈开源JoyAI-VL-Interaction，让大模型从“一问一答”走向“边看边说”

avatar AtomGit开源社区