为解决大模型外部工具调用、上下文传递标准不统一、适配混乱的痛点，本文基于 MCP 模型上下文协议，使用轻量化 FastMCP 构建 MCP 服务端，借助 langchain‑mcp‑adapters 完成与 LangChain 生态集成。文章介绍 Streamable HTTP 传输模式的适配场景，以天气查询作为实战案例，完成自定义工具、资源接口、提示词模板开发，并演示 LangChain 客户端加载工具、读取资源、解析提示词、多服务协同调用完整流程，附带可运行代码与依赖配置，提供一套轻量化标准化的大模型工具调用实践方案。
本文将以天气查询、系统时间读取为实战案例，完整实现自定义工具开发、静态资源封装、通用提示词模板配置三大核心能力。同时逐层演示 LangChain 客户端加载工具、读取资源、渲染提示词、多服务协同调用的全流程，配套完整可运行代码与依赖环境配置，为开发者提供一套轻量化、标准化、可落地的大模型工具调用实践方案。
MCP 服务核心配置 MCP（Model Context Protocol，模型上下文协议）是一套开源通用的大模型交互标准协议，核心作用是统一规范「应用向大模型交付工具能力、传递上下文信息」的交互格式，彻底解决不同工具、不同框架适配混乱的问题。而 FastMCP 是基于 Python 实现的轻量化 MCP 服务端开发框架，封装了繁杂的协议底层逻辑，支持快速定义工具、静态资源、提示词模板，兼容多种传输模式，大幅降低了 MCP 协议的落地开发门槛。
通常MCP的架构由 MCP Host、MCP Client、MCP Server 三部分组成：
在通信层面，MCP 客户端与服务端基于标准化 MCP 协议报文交互，底层支持 Stdio、SSE、Streamable HTTP 三种传输载体，可精准适配不同开发与部署场景：Stdio 适用于本地调试场景、SSE 适配简易流式交互场景、Streamable HTTP 适配生产环境远程部署场景。本文将采用Streamable HTTP模式实现接口通信，该模式也是官方推荐的生产环境首选方案，兼具稳定性与通用性。
环境依赖安装 为保障后续代码案例正常运行，需提前安装 FastMCP、langchain-mcp-adapters 等核心依赖库，推荐使用清华镜像源加速安装，安装命令及完整依赖版本清单如下：
CMD> pip install -i https://mirrors.tuna.tsinghua.edu.cn/pypi/web/simple 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 模式对外提供服务，读者可自行展开复制代码，并将其保存至本地磁盘中。
点击展开WeatherMCP代码
import jsonfrom fastmcp import FastMCPfrom 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" )
点击展开SystemDateTimeMCP代码
import jsonimport timefrom datetime import datetimefrom fastmcp import FastMCPfrom 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" )
分别独立运行两个服务端脚本，即可在本地开启对应端口的 MCP 服务，启动日志如下，代表服务监听成功、可正常接收客户端请求：
CMD> python WeatherMCP.py Started server process Waiting for application startup. Application startup complete. Uvicorn running on [http://0.0.0.0:8001] CMD> python SystemDateTimeMCP.py Started server process Waiting for application startup. Application startup complete. Uvicorn running on [http://0.0.0.0:8002]
MCP 客户端核心能力：资源加载 依托 langchain-mcp-adapters 适配器，LangChain 客户端可实现三大核心能力：工具集加载调用、静态资源读取、提示词模板加载渲染。客户端通过 Streamable HTTP 协议对接本地两个 MCP 服务端点，彻底实现工具能力、静态配置、提示词模板与业务客户端解耦，无需本地硬编码任何工具逻辑、配置参数与提示词内容，大幅提升代码复用性与标准化程度。
MCP 协议将服务端能力统一拆分为三类核心实体：Tools（可执行工具函数）、Resources（静态只读资源）、Prompts（标准化提示模板），客户端可自动发现、加载、调用所有实体能力。
加载可执行工具函数 MCP 协议中，Tool 是服务端封装的可远程调用函数，服务端会标准化声明工具名称、功能描述、JSON Schema 参数约束。客户端加载工具元数据后，可直接对接 LangChain Agent，由大模型自主判断工具选择、参数填充，再通过 MCP 协议转发请求至服务端执行，实现智能化工具调用。
下述案例基于 Streamable HTTP 通信模式，通过 client.get_tools() 读取服务端全部可调用工具，并批量打印工具详情、参数约束、必填规则，方便开发者清晰掌握可用能力，也是大模型智能调度工具的核心依据。
import asynciofrom langchain_mcp_adapters.client import MultiServerMCPClientasync def main (): client = MultiServerMCPClient( { "weather" : { "transport" : "http" , "url" : "http://127.0.0.1:8001/mcp" , } } ) 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
加载静态只读资源 MCP 协议中的 Resource 是服务端托管的静态只读数据，无业务逻辑计算能力，主要用于存储服务配置、常量参数、参考文档、声明文本等固定资源，供大模型直接引用学习。资源通过专属 URI 前缀区分类型：config:// 用于存放服务元信息、业务配置、常量数据；file:// 用于存放免责声明、说明文档等文本资源。
本文提供两种资源加载方式，适配不同业务场景，第一种使用client.get_resources("weather")方法加载，传入的参数为MultiServerMCPClient初始化的不同的MCP服务端名称，第二种方式则是直接使用load_mcp_resources一步到位加载资源，两种方式均可，开发者可按需选用。
import asyncioimport jsonfrom langchain_mcp_adapters.client import MultiServerMCPClientfrom langchain_mcp_adapters.resources import load_mcp_resourcesasync def main (): client = MultiServerMCPClient( { "weather" : { "transport" : "http" , "url" : "http://127.0.0.1:8001/mcp" , } } ) 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协议开发调试学习使用。
加载提示词模板 MCP 协议支持在服务端统一托管可复用的 Prompt 模板，模板内置变量占位符，客户端仅需传入对应业务参数，即可由服务端完成变量渲染，直接返回格式化后的完整提示词。该方式可实现提示词标准化统一管理，避免本地提示词杂乱、格式不统一的问题，同时支持利用提示词模版实现Skill技能包的加载及调用。
本文提供两种提示词加载渲染方案，适配不同开发需求，第一种直接使用await session.get_prompt()得到对应的提示词函数，传入相应的参数完成依次填充调用，并使用mcp_prompt_msg_to_langchain将数据转换为HumanMessage返回，第二种则是直接调用load_mcp_prompt()这个官方接口实现提示词填充。
import asynciofrom langchain_mcp_adapters.client import MultiServerMCPClientfrom langchain.agents import create_agentfrom langchain_core.messages import HumanMessagefrom langchain_mcp_adapters.prompts import load_mcp_promptdef 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 : return HumanMessage(content=text) async def main (): client = MultiServerMCPClient( { "weather" : { "transport" : "http" , "url" : "http://127.0.0.1:8001/mcp" , } } ) 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 ("方式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 资源协同调用实战 完成基础能力加载后，即可实现多 MCP 服务协同调用，让 LangChain Agent 自动感知多服务工具能力，根据用户自然语言提问自主选择对应工具，完成天气查询、时间获取等复合需求，实现端到端智能化工具调用。
基础工具协同调用 客户端同时对接天气、时间两个 MCP 服务，加载全部工具能力，通过本地大模型驱动 Agent 自动解析用户需求，调度对应工具完成查询并整合结果输出。
import asynciofrom langchain_mcp_adapters.client import MultiServerMCPClientfrom langchain.agents import create_agentfrom langchain_openai import ChatOpenAIfrom langchain_core.messages import HumanMessage, AIMessagebasic_model = ChatOpenAI( model="qwen2.5-1.5b-instruct-q4_k_m.gguf" , base_url="http://127.0.0.1:11433/v1" , api_key="dummy" , temperature=0.1 , max_tokens=512 , ) async def main (): client = MultiServerMCPClient( { "weather" : { "transport" : "http" , "url" : "http://127.0.0.1:8001/mcp" , }, "datetime" : { "transport" : "http" , "url" : "http://127.0.0.1:8002/mcp" , }, } ) tools = await client.get_tools() print (f"\n>>> 总计工具数量：{len (tools)} " ) agent = create_agent( model=basic_model, tools=tools ) response = await agent.ainvoke( {"messages" : "当前系统日期与时间是多少？北京天气怎么样？" } ) 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 asynciofrom langchain_mcp_adapters.client import MultiServerMCPClientfrom langchain.agents import create_agentfrom langchain_openai import ChatOpenAIfrom langchain_core.messages import HumanMessagefrom langchain_mcp_adapters.prompts import load_mcp_promptbasic_model = ChatOpenAI( model="qwen2.5-1.5b-instruct-q4_k_m.gguf" , base_url="http://127.0.0.1:11433/v1" , 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" : "http://127.0.0.1:8001/mcp" , }, "datetime" : { "transport" : "http" , "url" : "http://127.0.0.1:8002/mcp" , }, } ) 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 asyncioimport uuidfrom langchain_mcp_adapters.client import MultiServerMCPClientfrom langchain.agents import create_agentfrom langchain_openai import ChatOpenAIfrom langchain_core.messages import HumanMessage, AIMessage, ToolMessage, SystemMessagefrom langgraph.checkpoint.memory import InMemorySaverSYSTEM_PROMPT = """ 你是工具调用Agent，有可用工具列表。 用户提问需要外部信息时，必须输出结构化工具调用tool_calls，禁止只在文本中说要调用工具 不要预编造数据，全部信息必须通过调用工具获取。 拿到工具返回结果之后，完整总结，再整理自然语言回答用户。 """ basic_model = ChatOpenAI( model="qwen2.5-1.5b-instruct-q4_k_m.gguf" , base_url="http://127.0.0.1:11433/v1" , api_key="dummy" , temperature=0.0 , max_tokens=1024 , ) async def interactive_loop (): client = MultiServerMCPClient( { "weather" : { "transport" : "http" , "url" : "http://127.0.0.1:8001/mcp" , }, "datetime" : { "transport" : "http" , "url" : "http://127.0.0.1:8002/mcp" , }, } ) 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以上的模型来测试。
