大模型要调用外部工具，常被三个问题卡住：协议不统一、上下文格式混乱、各框架适配成本高。MCP（Model Context Protocol，模型上下文协议）正是为此而生的一套开放标准，它定义了"应用向模型交付工具能力、传递上下文信息"的统一交互格式。本文用轻量框架 FastMCP 落地这套协议，并通过 langchain-mcp-adapters 把服务能力接入 LangChain 生态。
以"天气查询 + 系统时间"两个场景为代表，我们会依次完成自定义工具、静态资源、提示词模板三项核心能力的开发，再逐步演示 LangChain 客户端如何加载工具、读取资源、渲染提示词、跨服务协同调用，并附上完整依赖清单与可运行代码，最终给出一套轻量且标准化的工具调用方案。
MCP 协议与 FastMCP 概览
MCP 是一套开源、通用的模型交互标准，核心价值在于统一规范"应用向大模型交付工具能力、传递上下文信息"的交互格式，让不同工具、不同框架不再各自为政。FastMCP 则用 Python 封装了协议底层，让定义工具、资源、提示词变得轻松，并内置多种传输模式，显著降低了上手门槛。
从架构上看，MCP 由三部分协同：MCP Host、MCP Client、MCP Server。通信层面，客户端与服务端遵循标准 MCP 报文交互，底层可承载于三种传输载体：Stdio 适合本地调试；SSE 适合简易流式交互；Streamable HTTP 则面向生产环境远程部署。本文统一采用 Streamable HTTP 模式——它也是官方推荐的生产首选，兼顾稳定与通用。
环境依赖安装
动手前先把核心依赖装好：FastMCP 与 langchain-mcp-adapters。建议挂清华镜像加速，完整版本清单见下方。
CMD> pip install -i langchain-mcp-adapters fastmcp
CMD>
CMD> pip list
Package Version
------------------------- -----------
langchain 1.3.15
langchain-core 1.5.5
langchain-mcp-adapters 0.3.2
langchain-openai 1.5.1
langchain-protocol 0.0.18
langgraph 1.2.11
langgraph-checkpoint 4.2.0
langgraph-prebuilt 1.1.0
langgraph-sdk 0.4.2
langsmith 0.11.0
mcp 1.29.0
fastmcp 3.4.7
fastmcp-slim 3.4.7
服务端代码实现
我们将搭建两个相互独立的 MCP 服务，便于演示能力拆分与多服务协同：
- WeatherMCP 专注模拟天气数据查询，提供温度、湿度、天气状况、风力计算等工具，并配套资源与提示词；
- SystemDateTimeMCP 专注系统时间处理，提供时间获取、时间戳转换、日期计算等核心能力。
两者分别占用本地 8001、8002 端口，以 Streamable HTTP 模式对外服务。把下面的代码复制到本地保存即可。
import json
from fastmcp import FastMCP
from typing import Union, Optional
mcp = FastMCP("WeatherMCP")
@mcp.tool()
def get_temperature(city: str) -> str:
"""
获取指定城市的模拟实时温度
Args:
city: 城市名称，中文城市名，必填参数，不允许为空字符串
Returns:
str: 返回格式化温度字符串，格式示例：北京 当前温度 26℃
"""
return f"{city} 当前温度 26℃"
@mcp.tool()
def get_humidity(city: str) -> str:
"""
获取指定城市模拟湿度
Args:
city: 中文城市名称
Returns:
str: 湿度信息
"""
return f"{city} 当前相对湿度 62%"
@mcp.tool()
def get_weather_condition(city: str) -> str:
"""
获取城市天气状况：晴/多云/小雨/大雨
Args:
city: 中文城市名称
Returns:
str: 天气状况文字
"""
return f"{city} 天气：多云转晴"
@mcp.tool()
def calculate_wind(speed: float, direction: Optional[str] = "东") -> str:
"""
模拟风力计算工具
Args:
speed: 风速 m/s
direction: 风向，默认东，可选：东、南、西、北
Returns:
str: 风力描述
"""
level = int(speed // 2)
return f"{direction}风，风速{speed}m/s，风力{level}级"
@mcp.resource("config://weather/meta")
def weather_meta() -> str:
"""模拟天气服务元信息资源"""
meta = {
"service_name": "WeatherDemo",
"version": "1.0.0",
"note": "全部为模拟测试数据，非真实气象数据"
}
return json.dumps(meta, ensure_ascii=False, indent=2)
@mcp.resource("config://weather/suggestion")
def weather_suggestion() -> str:
"""天气生活建议资源，穿衣、出行提示"""
data = {
"temp_high": 28,
"temp_low": 18,
"clothes": "薄长袖、短袖，早晚备薄外套",
"uv_tip": "紫外线中等，建议涂防晒",
"rain_tip": "无雨，适合外出游玩"
}
return json.dumps(data, ensure_ascii=False, indent=2)
@mcp.resource("config://weather/wind_level_desc")
def wind_level_desc() -> str:
"""风力等级对照表资源"""
wind_data = {
"0级": "无风",
"1级": "软风",
"2级": "轻风",
"3级": "微风",
"4级": "和风",
"5级": "清风",
"6级": "强风",
"7级": "疾风",
"8级": "大风"
}
return json.dumps(wind_data, ensure_ascii=False, indent=2)
@mcp.resource("file://weather/disclaimer")
def weather_disclaimer() -> str:
"""免责声明文本资源"""
disclaimer_text = """
[免责声明]
本MCP服务所有天气数据均为模拟演示数据，不具备真实气象参考价值。
不可用于生产、出行决策、防灾减灾等实际场景。
仅供MCP协议开发调试学习使用。
""".strip()
return disclaimer_text
@mcp.prompt()
def weather_ask(city: str) -> str:
"""
生成天气查询提示词模板
Args:
city: 城市名
"""
return f"""请帮我查询 {city} 的完整天气，包含温度、湿度、风向，整理成一段通顺自然的中文报告。"""
@mcp.prompt()
def weather_travel_prompt(city: str, travel_day: str) -> str:
"""
出行旅游天气提示模板
Args:
city: 目标城市
travel_day: 出行日期，例如：周六
"""
return f"""
我计划{travel_day}去{city}旅行。
请调用天气工具获取该城市温度、湿度、天气状况、风力信息。
结合天气给出出行建议：穿什么衣服、是否带伞、紫外线提醒。
输出简洁易懂，适合普通游客阅读。
""".strip()
@mcp.prompt()
def weather_json_report(city: str) -> str:
"""
获取结构化JSON格式天气报告提示词
Args:
city: 查询城市
"""
return f"""
请查询{city}的全部天气信息：温度、湿度、天气状况、风力。
将结果整理成标准JSON格式返回，不要多余解释文本。
字段包含：city,temperature,humidity,weather,wind_info。
""".strip()
@mcp.prompt()
def weather_alert_prompt(city: str) -> str:
"""
天气风险提醒提示模板
Args:
city: 查询城市
"""
return f"""
帮我分析{city}的天气情况。
判断是否存在高温、大风、降雨等需要注意的风险。
输出风险提示，同时给出对应的生活注意事项。
语言简短，适合作为预警提示。
""".strip()
if __name__ == "__main__":
mcp.run(transport="streamable-http", host="0.0.0.0", port=8001, path="/mcp")
import json
import time
from datetime import datetime
from fastmcp import FastMCP
from typing import Optional
mcp = FastMCP("SystemDateTimeMCP")
@mcp.tool()
def get_current_datetime(timezone_offset: Optional[int] = None) -> str:
"""
获取当前系统日期时间，默认本地系统时间，可指定时区偏移小时数
Args:
timezone_offset: 时区偏移小时，例如8代表东八区，None使用本机系统时区
Returns:
str: 格式化的完整日期时间字符串，示例：2026‑08‑19 14:30:22 星期三
"""
if timezone_offset is not None:
now = datetime.utcnow()
now = now.replace(hour=now.hour + timezone_offset)
else:
now = datetime.now()
weekday_map = {0: "星期一", 1: "星期二", 2: "星期三", 3: "星期四", 4: "星期五", 5: "星期六", 6: "星期日"}
wd = weekday_map[now.weekday()]
return f"{now.strftime('%Y‑%m‑%d %H:%M:%S')} {wd}"
@mcp.tool()
def get_timestamp(ms: bool = False) -> int:
"""
获取Unix时间戳
Args:
ms: 是否返回毫秒时间戳，True返回毫秒，False返回秒
Returns:
int: unix时间戳
"""
if ms:
return int(time.time() * 1000)
return int(time.time())
@mcp.tool()
def format_timestamp(timestamp: int, ms: bool = False) -> str:
"""
将时间戳转换为可读日期时间字符串
Args:
timestamp: unix时间戳
ms: 输入时间戳是否为毫秒，True则视为毫秒时间戳
Returns:
str: 格式化时间字符串
"""
if ms:
ts = timestamp / 1000
else:
ts = timestamp
dt = datetime.fromtimestamp(ts)
return dt.strftime("%Y‑%m‑%d %H:%M:%S")
@mcp.tool()
def get_date_parts() -> str:
"""
获取当前时间拆解后的年、月、日、时、分、秒、星期数字，返回JSON字符串
"""
now = datetime.now()
data = {
"year": now.year,
"month": now.month,
"day": now.day,
"hour": now.hour,
"minute": now.minute,
"second": now.second,
"weekday_num": now.weekday(),
"isoweekday": now.isoweekday()
}
return json.dumps(data, ensure_ascii=False, indent=2)
@mcp.tool()
def date_calc(base_date_str: str, days: int) -> str:
"""
日期加减计算，输入 yyyy‑MM‑dd 格式日期，增减N天，返回新日期
Args:
base_date_str: 基准日期，格式 yyyy‑MM‑dd
days: 增加/减少天数，正数向后，负数向前
Returns:
str: 计算后的日期字符串 yyyy‑MM‑dd
"""
from datetime import timedelta
base = datetime.strptime(base_date_str, "%Y‑%m‑%d")
res = base + timedelta(days=days)
return res.strftime("%Y‑%m‑%d")
@mcp.resource("config://datetime/service_meta")
def datetime_service_meta() -> str:
"""日期时间MCP服务元信息资源"""
meta = {
"service_name": "SystemDateTimeMCP",
"version": "1.0.0",
"description": "获取系统时间、时间戳、日期计算演示MCP服务",
"note": "读取运行服务机器的本地系统时间"
}
return json.dumps(meta, ensure_ascii=False, indent=2)
@mcp.resource("config://datetime/format_examples")
def datetime_format_examples() -> str:
"""时间格式化示例参考资源"""
examples = {
"%Y‑%m‑%d": "年‑月‑日",
"%H:%M:%S": "时:分:秒(24小时)",
"%Y‑%m‑%d %H:%M:%S": "完整日期时间",
"%Y%m%d": "紧凑日期",
"%H%M%S": "紧凑时间"
}
return json.dumps(examples, ensure_ascii=False, indent=2)
@mcp.resource("file://datetime/disclaimer")
def datetime_disclaimer() -> str:
"""时间服务免责声明"""
text = """
[DateTimeMCP 说明]
时间取自运行MCP服务主机的本地系统时钟。
如果主机系统时间不准，则返回时间会存在偏差。
仅供开发调试学习使用。
""".strip()
return text
@mcp.prompt()
def prompt_now_info() -> str:
"""获取当前完整时间信息提示词"""
return """
请调用工具获取当前系统完整时间，包含可读日期时间、时间戳（秒与毫秒），拆解年月日时分秒，整理成清晰中文报告。
"""
@mcp.prompt()
def prompt_date_calc_report(base_date: str, offset_days: int) -> str:
"""日期计算报告提示词"""
return f"""
基准日期 {base_date}，计算偏移 {offset_days} 天之后的日期。
调用date_calc工具完成计算，同时输出原始基准日期与结果日期，简单说明。
"""
@mcp.prompt()
def prompt_datetime_json() -> str:
"""输出JSON格式时间信息"""
return """
调用相关时间工具获取全部当前时间信息，输出纯JSON，不要多余文字。
字段：datetime_str, timestamp_sec, timestamp_ms, year, month, day, hour, minute, second
"""
@mcp.prompt()
def prompt_timestamp_convert(input_ts: int, is_ms: bool) -> str:
"""时间戳转可读时间提示词"""
return f"""
输入时间戳 {input_ts}，{'是毫秒时间戳' if is_ms else '是秒时间戳'}。
调用format_timestamp工具将时间戳转为人类可读时间，展示转换前后对比。
"""
if __name__ == "__main__":
mcp.run(transport="streamable-http", host="0.0.0.0", port=8002, path="/mcp")
分别运行两个脚本即可在本地开启对应端口的服务，看到如下日志说明监听成功、可以接收请求：
CMD> python WeatherMCP.py
Started server process
Waiting for application startup.
Application startup complete.
Uvicorn running on []
CMD> python SystemDateTimeMCP.py
Started server process
Waiting for application startup.
Application startup complete.
Uvicorn running on []
LangChain 客户端的三种能力
借助 langchain-mcp-adapters，LangChain 客户端能解锁三大能力：工具集加载调用、静态资源读取、提示词模板渲染。客户端通过 Streamable HTTP 对接本地两个服务端点，使工具逻辑、配置参数、提示词内容全部与服务端解耦，业务侧无需硬编码任何工具实现，复用性与标准化程度都大幅提升。
MCP 把服务端能力抽象为三类实体：Tools（可执行函数）、Resources（静态只读资源）、Prompts（标准化提示模板）。客户端可自动发现、加载并调用全部实体。
加载可执行工具函数
Tool 是服务端封装、可远程调用的函数。服务端会标准化声明它的名称、功能描述与 JSON Schema 参数约束。客户端拿到元数据后，可直接交给 LangChain Agent，由模型自行决定何时调用、如何填参，再通过 MCP 协议把请求转发到服务端执行，实现智能化调用。
下面基于 Streamable HTTP，用 client.get_tools() 读取全部工具，并打印名称、参数约束与必填规则——这也是模型智能调度的依据。
import asyncio
from langchain_mcp_adapters.client import MultiServerMCPClient
async def main():
client = MultiServerMCPClient(
{
"weather": {
"transport": "http",
"url": "
}
}
)
tools = await client.get_tools()
for idx, tool in enumerate(tools, 1):
print(f"[{idx}] 工具名称: {tool.name}")
print(f" 功能描述: {tool.description}")
schema = tool.args_schema
required_fields = set(schema.get("required", []))
props = schema.get("properties", {})
print(" 参数列表:")
if not props:
print(" (无参数)")
else:
for param_name, param_info in props.items():
is_required = "必填" if param_name in required_fields else "可选"
typ = param_info.get("type", "unknown")
desc = param_info.get("description", "")
default_val = param_info.get("default", "无")
anyof = param_info.get("anyOf")
if anyof:
typ = " | ".join([t.get("type") for t in anyof if "type" in t])
print(f" · {param_name} [{typ}] {is_required}：{desc}，默认值={default_val}")
print("-" * 80)
print(f"\n>>> 总计工具数量：{len(tools)}")
if __name__ == "__main__":
asyncio.run(main())
运行后完整输出每个工具的说明、参数类型与必填规则，示例如下：
CMD> python main.py
[1] 工具名称: get_temperature
功能描述: 获取指定城市的模拟实时温度
参数列表:
· city [string] 必填：城市名称，中文城市名，例如：北京、上海、广州；必填参数，不允许为空字符串，默认值=无
--------------------------------------------------------------------------------
[2] 工具名称: get_humidity
功能描述: 获取指定城市模拟湿度
参数列表:
· city [string] 必填：中文城市名称，默认值=无
--------------------------------------------------------------------------------
[3] 工具名称: get_weather_condition
功能描述: 获取城市天气状况：晴/多云/小雨/大雨
参数列表:
· city [string] 必填：中文城市名称，默认值=无
--------------------------------------------------------------------------------
[4] 工具名称: calculate_wind
功能描述: 模拟风力计算工具
参数列表:
· speed [number] 必填：风速 m/s，默认值=无
· direction [string | null] 可选：风向，默认东，可选：东、南、西、北，默认值=东
--------------------------------------------------------------------------------
>>> 总计工具数量：4
加载静态只读资源
Resource 是服务端托管的静态只读数据，不带计算逻辑，通常放配置、常量、参考文档、声明文本等。它通过 URI 前缀区分类型：config:// 放元信息、业务配置、常量；file:// 放免责声明、说明文档等纯文本。
资源加载提供两种方式，可任选其一：其一用 client.get_resources("weather")，参数对应初始化 MultiServerMCPClient 时的服务名；其二是直接用 load_mcp_resources 一步到位。
import asyncio
import json
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_mcp_adapters.resources import load_mcp_resources
async def main():
client = MultiServerMCPClient(
{
"weather": {
"transport": "http",
"url": "
}
}
)
# 第一种加载方式
blobs = await client.get_resources("weather")
resource_list = []
for blob in blobs:
uri = str(blob.metadata["uri"])
raw_data = blob.data
parsed = None
try:
parsed = json.loads(raw_data)
except json.JSONDecodeError:
parsed = raw_data
item = {
"uri": uri,
"raw": raw_data,
"content": parsed,
"mimetype": blob.mimetype
}
resource_list.append(item)
for item in resource_list:
print(f"\nURI: {item['uri']}")
print(item["content"])
# 第二种加载方式
async with client.session("weather") as session:
blobs = await load_mcp_resources(session)
for blob in blobs:
print(blob)
# 第三种加载方式
async with client.session("weather") as session:
blobs = await load_mcp_resources(session,uris=["file://weather/disclaimer"])
for blob in blobs:
print(blob)
if __name__ == "__main__":
asyncio.run(main())
运行后，客户端成功读回天气服务的所有静态资源，含元信息、出行建议、风力对照表与免责声明：
CMD> python main.py
URI: config://weather/meta
{
'service_name': 'WeatherDemo',
'version': '1.0.0',
'note': '全部为模拟测试数据，非真实气象数据'
}
URI: config://weather/suggestion
{
'temp_high': 28,
'temp_low': 18,
'clothes': '薄长袖、短袖，早晚备薄外套',
'uv_tip': '紫外线中等，建议涂防晒',
'rain_tip': '无雨，适合外出游玩'
}
URI: config://weather/wind_level_desc
{
'0级': '无风',
'1级': '软风',
'2级': '轻风',
'3级': '微风',
'4级': '和风',
'5级': '清风',
'6级': '强风',
'7级': '疾风',
'8级': '大风'
}
URI: file://weather/disclaimer
[免责声明]
本MCP服务所有天气数据均为模拟演示数据，不具备真实气象参考价值。
不可用于生产、出行决策、防灾减灾等实际场景。
仅供MCP协议开发调试学习使用。
加载提示词模板
服务端还能统一托管可复用的 Prompt 模板，模板内置可替换的变量。客户端只传业务参数，服务端负责渲染，返回成型的完整提示词。这样既能集中管理、避免本地提示词杂乱，也能用于加载 Skill 技能包。
加载与渲染有两套方案：其一用 await session.get_prompt() 拿到提示词消息，再经 mcp_prompt_msg_to_langchain 转成 HumanMessage；其二是直接调官方接口 load_mcp_prompt() 完成填充。
import asyncio
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain.agents import create_agent
from langchain_core.messages import HumanMessage
from langchain_mcp_adapters.prompts import load_mcp_prompt
def mcp_prompt_msg_to_langchain(mcp_msg):
"""转换MCP PromptMessage → LangChain Message 消息实例"""
content_obj = mcp_msg.content
if isinstance(content_obj, list):
text_parts = [c.text for c in content_obj if hasattr(c, "text")]
text = "".join(text_parts)
else:
text = content_obj.text
if mcp_msg.role == "user":
return HumanMessage(content=text)
else:
# 可扩展 SystemMessage / AIMessage
return HumanMessage(content=text)
async def main():
client = MultiServerMCPClient(
{
"weather": {
"transport": "http",
"url": "
}
}
)
# 第一种加载方式
lc_messages = []
async with client.session("weather") as session:
prompt_result = await session.get_prompt(
name="weather_ask",
arguments={"city": "济南"}
)
for msg in prompt_result.messages:
# print(f"role={msg.role}, raw_content={msg.content}")
lc_msg = mcp_prompt_msg_to_langchain(msg)
lc_messages.append(lc_msg)
print("方式1: ", lc_messages)
# 第二种加载方式
async with client.session("weather") as session:
messages = await load_mcp_prompt(
session,
"weather_travel_prompt",
arguments={"city": "北京","travel_day":"2026-8-20"}
)
print("方式2: ", messages)
if __name__ == "__main__":
asyncio.run(main())
两种方式都能完成变量填充与提示词渲染，输出标准化的用户提问：
CMD> python main.py
方式1: [HumanMessage(content='请帮我查询 济南 的完整天气，包含温度、湿度、风向，整理成一段通顺自然的中文报告。', additional_kwargs={}, response_metadata={})]
方式2: [HumanMessage(content='我计划2026-8-20去北京旅行。\n输出简洁易懂，适合普通游客阅读。', additional_kwargs={}, response_metadata={})]
多服务协同调用实战
完成基础能力加载后，即可实现多 MCP 服务协同调用，让 LangChain Agent 自动感知各服务的工具能力，依据用户自然语言自主选择工具，完成天气查询、时间获取等复合需求，跑通端到端的智能化调用。
基础工具协同调用
客户端同时连上天气与时间两个服务，把工具全部加载，由本地模型驱动的 Agent 解析用户意图、调度对应工具并汇总结果。
import asyncio
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage
basic_model = ChatOpenAI(
model="qwen2.5-1.5b-instruct-q4_k_m.gguf",
base_url="
api_key="dummy",
temperature=0.1,
max_tokens=512,
)
async def main():
client = MultiServerMCPClient(
{
"weather": {
"transport": "http",
"url": "
},
"datetime": {
"transport": "http",
"url": "
},
}
)
tools = await client.get_tools()
print(f"\n>>> 总计工具数量：{len(tools)}")
agent = create_agent(
model=basic_model,
tools=tools
)
response = await agent.ainvoke(
{"messages": "当前系统日期与时间是多少？北京天气怎么样？"}
)
# 只提取 HumanMessage / AIMessage 的 content
for msg in response["messages"]:
if isinstance(msg, HumanMessage):
print(f"[Human]: {msg.content}")
elif isinstance(msg, AIMessage):
if msg.content.strip():
print(f"[AI]: {msg.content}")
if __name__ == "__main__":
asyncio.run(main())
协同调用输出效果：
CMD> python main.py
>>> 总计工具数量：9
[Human]: 当前系统日期与时间是多少？北京天气怎么样？
[AI]:
当前系统日期与时间是：2026-08-19 13:30:51 星期三。
北京的天气状况是：多云转晴。
提示词模板落地调用
把前面加载的服务端模板用起来，先生成标准化提问，再交给 Agent 执行工具调用，统一输出格式、规范回答逻辑，避免用户提问不规范引发调用异常。
import asyncio
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
from langchain_mcp_adapters.prompts import load_mcp_prompt
basic_model = ChatOpenAI(
model="qwen2.5-1.5b-instruct-q4_k_m.gguf",
base_url="
api_key="dummy",
temperature=0.1,
max_tokens=512,
)
def mcp_prompt_msg_to_langchain(mcp_msg):
"""转换mcp.types.PromptMessage 为 langchain消息实例"""
content_obj = mcp_msg.content
if isinstance(content_obj, list):
text_parts = [c.text for c in content_obj if hasattr(c, "text")]
text = "".join(text_parts)
else:
text = content_obj.text
if mcp_msg.role == "user":
return HumanMessage(content=text)
else:
return HumanMessage(content=text)
async def main():
client = MultiServerMCPClient(
{
"weather": {
"transport": "http",
"url": "
},
"datetime": {
"transport": "http",
"url": "
},
}
)
tools = await client.get_tools()
# 填充参数调用生成提示词模版
lc_messages = []
async with client.session("weather") as session:
prompt_result = await session.get_prompt(
name="weather_ask",
arguments={"city": "济南"}
)
for msg in prompt_result.messages:
lc_msg = mcp_prompt_msg_to_langchain(msg)
lc_messages.append(lc_msg)
print("生成的提示词:", lc_messages)
agent = create_agent(
model=basic_model,
tools=tools
)
resp = await agent.ainvoke({"messages": lc_messages})
final = resp["messages"][-1]
print("Agent输出最终结果:", final.content)
# 第二种构建方式
'''
async with client.session("weather") as session:
messages = await load_mcp_prompt(
session,
"weather_ask",
arguments={"city": "山东"}
)
print(messages)
'''
if __name__ == "__main__":
asyncio.run(main())
模板调用输出效果：
CMD> python main.py
生成的提示词: [HumanMessage(content='请帮我查询 济南 的完整天气，包含温度、湿度、风向，整理成一段通顺自然的中文报告。', additional_kwargs={}, response_metadata={})]
Agent输出最终结果: 为了提供济南的完整天气报告，我将调用以下工具：
1. 获取济南的模拟实时温度
2. 获取济南的模拟湿度
3. 获取济南的天气状况
现在开始获取数据：
1. 获取济南的模拟实时温度
{{"name": "get_temperature", "arguments": {"city": "济南"}}}
2. 获取济南的模拟湿度
{{"name": "get_humidity", "arguments": {"city": "济南"}}}
3. 获取济南的天气状况
{{"name": "get_weather_condition", "arguments": {"city": "济南"}}}
请稍等，我将获取这些数据并整理成报告。
整体来看，济南的天气较为舒适，适合户外活动。风力适中，需要注意防风保暖。
交互式迭代对话
借助 LangGraph 的内存检查点，可搭出多轮、带上下文记忆、可重置的对话会话，更接近真实应用，持续调用 MCP 工具响应用户需求。
import asyncio
import uuid
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage, ToolMessage, SystemMessage
from langgraph.checkpoint.memory import InMemorySaver
SYSTEM_PROMPT = """
你是工具调用Agent，有可用工具列表。
用户提问需要外部信息时，必须输出结构化工具调用tool_calls，禁止只在文本中说要调用工具
不要预编造数据，全部信息必须通过调用工具获取。
拿到工具返回结果之后，完整总结，再整理自然语言回答用户。
"""
basic_model = ChatOpenAI(
model="qwen2.5-1.5b-instruct-q4_k_m.gguf",
base_url="
api_key="dummy",
temperature=0.0,
max_tokens=1024,
)
async def interactive_loop():
client = MultiServerMCPClient(
{
"weather": {
"transport": "http",
"url": "
},
"datetime": {
"transport": "http",
"url": "
},
}
)
try:
tools = await client.get_tools()
print(f"[*] 加载完成 工具总数：{len(tools)}")
for idx, t in enumerate(tools, 1):
print(f" -{idx}. {t.name} : {t.description}")
print()
# 内存记忆检查点
checkpointer = InMemorySaver()
agent = create_agent(
model=basic_model,
tools=tools,
checkpointer=checkpointer
)
thread_config = {
"configurable": {
"thread_id": "mcp‑chat‑001"
}
}
while True:
user_input = input("\n请输入你的问题(exit退出) > ").strip()
if user_input.lower() in ("exit", "quit"):
print("[-]退出会话")
break
if user_input.lower() == "clear":
thread_config["configurable"]["thread_id"] = f"mcp‑chat‑{uuid.uuid4()}"
print("[*] 已清空会话记忆，开启全新对话！")
continue
if not user_input:
continue
print("\n----------开始执行Agent流程----------")
print(f"[用户提问]：{user_input}")
try:
resp = await agent.ainvoke(
{
"messages": [
SystemMessage(content=SYSTEM_PROMPT),
HumanMessage(content=user_input)
]
},
config=thread_config
)
messages = resp["messages"]
for idx, msg in enumerate(messages):
t_name = type(msg).__name__
print(f"\n---消息[{idx}] 类型:{t_name}---")
if isinstance(msg, HumanMessage):
print(f"Human: {msg.content}")
elif isinstance(msg, AIMessage):
print(f"AI content: {msg.content}")
if msg.tool_calls:
print(f"AI 请求调用工具：")
for call in msg.tool_calls:
print(f" 工具名={call['name']}, 参数={call['args']}")
else:
print("[-] AIMessage 没有生成 tool_calls，不会执行MCP工具！")
elif isinstance(msg, ToolMessage):
print(f"[*] 工具返回结果[{msg.name}]：\n{msg.content}")
else:
print(f"{t_name} content: {msg.content}")
final_ai = None
for m in reversed(messages):
if isinstance(m, AIMessage) and m.content.strip():
final_ai = m
break
if final_ai:
print("\n[+] 最终回答")
print(final_ai.content)
except Exception as e:
print(f"\n[-] Agent执行异常：{repr(e)}")
except Exception as conn_err:
print(f"\n[-] MCP服务连接失败，请确认8001、8002端口MCP服务已启动！")
print(f"错误信息：{conn_err}")
if __name__ == "__main__":
asyncio.run(interactive_loop())
交互式对话运行效果：
CMD> python main.py
[*] 加载完成 工具总数：9
-1. get_temperature : 获取指定城市的模拟实时温度
-2. get_humidity : 获取指定城市模拟湿度
-3. get_weather_condition : 获取城市天气状况：晴/多云/小雨/大雨
-4. calculate_wind : 模拟风力计算工具
-5. get_current_datetime : 获取当前系统日期时间，默认本地系统时间，可指定时区偏移小时数
-6. get_timestamp : 获取Unix时间戳
-7. format_timestamp : 将时间戳转换为可读日期时间字符串
-8. get_date_parts : 获取当前时间拆解后的年、月、日、时、分、秒、星期数字，返回JSON字符串
-9. date_calc : 日期加减计算，输入 yyyy‑MM‑dd 格式日期，增减N天，返回新日期
请输入你的问题(exit退出) > 查询当前时间与日期，并查看济南天气
----------开始执行Agent流程----------
[用户提问]：查询当前时间与日期，并查看济南天气
---消息[0] 类型:SystemMessage---
SystemMessage content:
你是工具调用Agent，有可用工具列表。
用户提问需要外部信息时，必须输出结构化工具调用tool_calls，禁止只在文本中说要调用工具
不要预编造数据，全部信息必须通过调用工具获取。
拿到工具返回结果之后，再整理自然语言回答用户。
---消息[1] 类型:HumanMessage---
Human: 查询当前时间与日期，并查看济南天气
---消息[2] 类型:AIMessage---
AI content: 为了获取当前时间与日期，我将调用一个获取当前系统日期时间的工具。
{"name": "get_current_datetime", "arguments": {"timezone_offset": null}}
AI 请求调用工具：
工具名=get_current_datetime, 参数={'timezone_offset': None}
---消息[3] 类型:ToolMessage---
[*] 工具返回结果[get_current_datetime]：
[{'type': 'text', 'text': '2026‑08‑19 14:11:37 星期三', 'id': 'lc_3ded815c-af83-4e7c-8ad1-3f9812967374'}]
---消息[4] 类型:AIMessage---
AI content: 当前时间是2026年8月19日14:11:37，星期三。
接下来，我将调用一个获取济南天气的工具。
{"name": "get_weather_condition", "arguments": {"city": "济南"}}
AI 请求调用工具：
工具名=get_weather_condition, 参数={'city': '济南'}
---消息[5] 类型:ToolMessage---
[*] 工具返回结果[get_weather_condition]：
[{'type': 'text', 'text': '济南 天气：多云转晴', 'id': 'lc_25a49010-62b2-49d7-a560-e47247085f69'}]
---消息[6] 类型:AIMessage---
AI content: 济南当前的天气状况是多云转晴。
[-] AIMessage 没有生成 tool_calls，不会执行MCP工具！
[+] 最终回答
济南当前的天气状况是多云转晴。
请输入你的问题(exit退出) > 当前日期时间呢？合在一起回复
---消息[7] 类型:SystemMessage---
SystemMessage content:
你是工具调用Agent，有可用工具列表。
用户提问需要外部信息时，必须输出结构化工具调用tool_calls，禁止只在文本中说要调用工具
不要预编造数据，全部信息必须通过调用工具获取。
拿到工具返回结果之后，再整理自然语言回答用户。
---消息[8] 类型:HumanMessage---
Human: 当前日期时间呢？合在一起回复
---消息[9] 类型:AIMessage---
AI content: 当前日期时间是2026年8月19日14:11:37，星期三。济南当前的天气状况是多云转晴。
[-] AIMessage 没有生成 tool_calls，不会执行MCP工具！
[+] 最终回答
当前日期时间是2026年8月19日14:11:37，星期三。济南当前的天气状况是多云转晴。
上面的交互测试中暴露出一个现象：模型第一次成功调用时间工具、拿到了完整时间数据，但用户追问"合在一起回复"时，模型没能引用历史时间结果，只输出了最新的天气查询内容。
这并非代码逻辑缺陷。根因在于本次测试所用的 qwen2.5‑1.5b 这类轻量模型上下文理解能力与记忆留存能力有限，部分历史信息被丢弃。若用于生产环境，建议直接换成 8B 以上规模的模型再验证。
评论0