news 2026/10/30 19:14:50

LangGraph 接入 MCP 协议实战:从工具解耦到多 Server 编排的完整指南

作者头像

张小明

前端开发工程师

1.2k 24
文章封面图
LangGraph 接入 MCP 协议实战:从工具解耦到多 Server 编排的完整指南

1. 为什么要在 LangGraph 里接入 MCP,而不是继续手写工具函数

如果你已经用 LangGraph 搭过几个 Agent,大概率经历过这个阶段:一开始觉得ToolNode加几个@tool装饰器就够了,后来工具越加越多,每个工具都要自己写参数校验、错误处理、鉴权逻辑,再后来发现隔壁团队用另一套框架写的工具,你想直接拿过来复用,结果发现接口对不上,只能重写一遍。这种重复劳动做多了,人就会开始想:有没有一种标准化的方式,让工具的定义和调用跟框架解耦?

MCP 协议就是冲着这个问题来的。它的全称是 Model Context Protocol,核心思路是把"模型能调用的外部能力"抽象成一套统一的协议,工具提供方只需要按照协议暴露自己的能力,调用方(也就是你的 Agent 框架)按照协议去发现和调用,双方不需要知道对方内部怎么实现。你可以把它理解成工具界的 USB-C 接口——以前每个设备一个充电口,现在统一了,插上就能用。

那为什么要在 LangGraph 里接 MCP,而不是继续手写工具函数?我总结了三个实际的理由。

第一个理由是工具复用成本。手写工具函数意味着你的工具和 LangGraph 是强绑定的,一旦你想换个框架,或者想把工具分享给用其他框架的同事,就得重写。而 MCP 工具是协议层面的,理论上任何支持 MCP 的客户端都能调用。我实测过一个场景:同一个 MCP 工具服务,既可以被 LangGraph 的 Agent 调用,也可以被其他支持 MCP 的客户端调用,切换成本几乎为零。

第二个理由是动态发现能力。手写工具的时候,工具列表是写死在代码里的,你必须在启动前就知道有哪些工具。但 MCP 支持运行时发现——客户端连上服务端之后,可以查询当前有哪些工具可用、每个工具需要什么参数。这对于工具会动态增减的场景特别有用,比如你有一个工具服务会根据用户权限动态暴露不同的工具集。

第三个理由是职责分离。工具的实现细节(比如数据库连接、API 密钥、重试逻辑)全部封装在 MCP 服务端,LangGraph 这边只负责编排和调用。这意味着工具服务可以独立部署、独立升级,Agent 这边不需要跟着改。我在一个项目里就是这么做的:工具服务由另一个团队维护,我们这边只关心怎么把工具编排进工作流,两边通过 MCP 协议对接,协作效率高了很多。

不过话说回来,MCP 也不是银弹。如果你的工具就三五个,而且短期内不会变,那手写工具函数反而更简单直接,引入 MCP 会增加一层网络通信和协议解析的开销。我的建议是:工具数量超过十个,或者有跨团队复用需求,或者工具需要动态发现,这时候上 MCP 才划算。

提示:MCP 目前还在快速演进中,协议版本之间可能有 breaking change。生产环境使用前,建议锁定协议版本,并关注官方仓库的更新日志。

2. MCP 协议的核心概念拆解:Host、Client、Server 到底谁是谁

很多人第一次看 MCP 文档的时候会被 Host、Client、Server 这三个角色绕晕,我当初也是。这里我用一个生活化的类比帮你理清楚:把 MCP 想象成餐厅点餐。Host 是餐厅(整个应用),Client 是服务员(负责传话),Server 是厨房(真正做菜的)。你(用户)跟餐厅说想吃什么,服务员把需求传给厨房,厨房做好菜,服务员再端回来给你。

具体到 LangGraph 的场景:

Host就是你的 LangGraph 应用本身。它是整个系统的入口,负责接收用户输入、管理对话状态、决定什么时候调用工具。Host 不直接跟 Server 通信,而是通过 Client 来中转。

Client是 Host 内部的一个组件,负责跟 Server 建立连接、发送请求、接收响应。在 LangGraph 里,Client 通常由 MCP 适配器来充当,你不需要自己实现协议细节,只需要配置好连接参数。一个 Host 可以同时连接多个 Server,每个 Server 对应一个 Client 实例。

Server是工具的实际提供方。它可以是本地进程(通过 stdio 通信),也可以是远程服务(通过 HTTP 或 SSE 通信)。Server 负责暴露工具列表、处理工具调用请求、返回执行结果。Server 内部怎么实现工具,Host 完全不关心。

这三者的关系可以用一个表格来对照:

角色职责在 LangGraph 中的对应通信方式
Host应用入口,管理状态和编排LangGraph 的 StateGraph不直接通信
Client协议适配,请求转发MCP 适配器组件与 Server 通信
Server工具提供,请求处理独立的工具服务进程与 Client 通信

理解了这三个角色,你就能明白为什么 MCP 能做到工具和框架解耦——因为工具的实现被隔离在 Server 里,Host 只通过 Client 这个标准接口跟 Server 交互,双方不需要知道对方的内部细节。

还有一个容易混淆的概念是Transport(传输层)。MCP 支持多种传输方式,最常见的是 stdio 和 SSE。stdio 适合本地进程通信,Server 作为子进程启动,通过标准输入输出跟 Client 交换数据。SSE 适合远程通信,Server 作为一个 HTTP 服务运行,Client 通过 Server-Sent Events 接收推送。选哪种传输方式取决于你的部署架构:本地工具用 stdio 更简单,远程工具用 SSE 更灵活。

我在实际项目中两种都用过。本地文件操作类的工具用 stdio,启动快、延迟低;需要跨机器调用的工具用 SSE,虽然多了一层网络开销,但部署和扩展方便很多。这里有个经验:如果你用 stdio,要注意 Server 进程的生命周期管理,Host 退出时要确保子进程也被正确清理,否则会留下僵尸进程。

2.1 工具发现与调用的完整生命周期

MCP 的工具调用不是一步到位的,它有一个完整的生命周期。理解这个生命周期,对排查问题特别有帮助。

第一步是初始化握手。Client 连上 Server 之后,双方会交换协议版本和 capabilities 信息。这一步很关键,如果版本不匹配,后续调用会直接失败。我踩过一次坑:Client 用的是较新的协议版本,Server 还是旧版本,握手阶段没报错,但调用工具时参数格式对不上,排查了半天才发现是版本问题。

第二步是工具发现。Client 向 Server 发送tools/list请求,Server 返回当前可用的工具列表,每个工具包含名称、描述、参数 schema。这个 schema 是 JSON Schema 格式的,Client 拿到之后可以动态生成调用界面或者做参数校验。

第三步是工具调用。Client 发送tools/call请求,带上工具名称和参数,Server 执行工具逻辑,返回结果。结果可以是文本、图片、结构化数据等多种格式。

第四步是错误处理。如果工具执行失败,Server 会返回错误信息,Client 需要把这个错误传递给 Host,Host 再决定是重试、降级还是直接报错给用户。

这个生命周期看起来简单,但实际用起来有几个细节要注意。比如工具发现的结果要不要缓存?我的做法是启动时发现一次,缓存起来,同时监听 Server 的通知(如果 Server 支持tools/list_changed通知的话),有变化时更新缓存。这样既避免了每次调用都去查询工具列表的开销,又能及时感知工具变化。

2.2 LangGraph 的 ToolNode 和 MCP 工具怎么对接

LangGraph 原生的ToolNode是围绕 LangChain 的BaseTool设计的,而 MCP 工具是协议层面的抽象,两者需要一个适配层。这个适配层的核心工作是把 MCP 工具的描述转换成 LangChain Tool 的格式,让ToolNode能识别和调用。

具体来说,适配层需要做三件事:

第一,把 MCP 工具的 JSON Schema 参数定义转换成 LangChain 的 args_schema。这一步大部分情况下可以自动转换,但有些边界情况需要手动处理,比如 MCP 工具用了 JSON Schema 的oneOf或anyOf,LangChain 的 schema 可能不支持,需要降级处理。

第二,把 MCP 工具的调用逻辑包装成 LangChain Tool 的_run或_arun方法。当ToolNode调用这个 Tool 时,实际执行的是 MCP 的tools/call请求。

第三,处理返回值的格式转换。MCP 工具返回的结果可能是多种格式,需要转换成 LangChain 能处理的格式,通常是字符串或者结构化数据。

我实测下来,这个适配层用现成的库可以省很多事,但如果你有特殊需求(比如自定义错误处理、自定义重试逻辑),自己写一个适配层也不复杂,核心代码量大概一两百行。

3. 从零搭建一个可运行的 MCP 工具服务

光讲概念没意思,我们直接动手搭一个。我会用一个"天气查询"工具作为示例,虽然简单,但涵盖了 MCP 工具服务的核心要素。你把这个跑通了,换成任何其他工具都是同样的套路。

3.1 环境准备与依赖选择

先说环境。Python 版本建议 3.10 以上,因为 MCP 的 Python SDK 用了一些较新的类型语法。依赖方面,核心是两个包:MCP 的 Python SDK 和 LangGraph。如果你要用 LangChain 的适配层,还需要装 langchain-mcp-adapters。

pip install mcp langgraph langchain-mcp-adapters

这里有个坑要注意:MCP SDK 的版本更新比较快,不同版本之间的 API 可能有变化。我建议在 requirements.txt 里锁定版本,比如mcp==1.2.0,避免因为自动升级导致代码跑不起来。我吃过这个亏,某次 CI 环境自动装了最新版,结果本地能跑的代码在 CI 上直接报错,排查发现是新版本改了某个函数的签名。

另外,如果你打算用 stdio 传输方式,不需要额外装什么。但如果用 SSE,可能需要一个 ASGI 服务器,比如 uvicorn。这个后面讲到远程部署的时候再说。

3.2 用 Python SDK 定义一个 MCP Server

下面是一个完整的 MCP Server 示例,提供两个工具:查询天气和查询空气质量。

from mcp.server.fastmcp import FastMCP from pydantic import Field # 创建 MCP Server 实例 mcp = FastMCP("weather-service") @mcp.tool() async def get_weather( city: str = Field(description="城市名称,例如:北京"), unit: str = Field(default="celsius", description="温度单位,celsius 或 fahrenheit") ) -> str: """查询指定城市的当前天气""" # 实际项目中这里会调用真实的天气 API # 这里用模拟数据演示 mock_data = { "北京": {"temp": 22, "condition": "晴", "humidity": 45}, "上海": {"temp": 25, "condition": "多云", "humidity": 60}, "广州": {"temp": 28, "condition": "小雨", "humidity": 75}, } data = mock_data.get(city) if not data: return f"未找到城市 {city} 的天气数据" temp = data["temp"] if unit == "fahrenheit": temp = temp * 9 / 5 + 32 return f"{city}当前天气:{data['condition']},温度{temp}度,湿度{data['humidity']}%" @mcp.tool() async def get_air_quality( city: str = Field(description="城市名称") ) -> str: """查询指定城市的空气质量指数""" mock_data = { "北京": {"aqi": 85, "level": "良"}, "上海": {"aqi": 62, "level": "良"}, "广州": {"aqi": 45, "level": "优"}, } data = mock_data.get(city) if not data: return f"未找到城市 {city} 的空气质量数据" return f"{city}空气质量:AQI {data['aqi']},等级{data['level']}" if __name__ == "__main__": # 用 stdio 传输方式启动 mcp.run(transport="stdio")

这段代码有几个点值得说明。

FastMCP是 SDK 提供的高层封装,它帮你处理了协议细节,你只需要用@mcp.tool()装饰器标记哪些函数是工具。装饰器会自动读取函数的类型注解和 docstring,生成工具的 schema。所以类型注解一定要写清楚,这直接决定了 Client 那边能不能正确理解参数含义。

Field里的description也很重要。这个描述会出现在工具 schema 里,LLM 在决定调用哪个工具、传什么参数的时候会参考这个描述。我见过有人偷懒不写 description,结果 LLM 经常传错参数,排查半天才发现是描述缺失导致的。

mcp.run(transport="stdio")是启动 Server 的入口。stdio 模式下,Server 会监听标准输入,通过标准输出返回结果。这种模式适合本地进程,Host 启动 Server 的时候会把它作为一个子进程。

3.3 本地调试:怎么确认 Server 真的能跑

写完 Server 之后,别急着接到 LangGraph 里,先单独测一下 Server 能不能正常工作。MCP SDK 提供了一个命令行工具,可以直接跟 Server 交互。

# 启动 Server 并用 inspector 连接 mcp dev weather_server.py

这个命令会启动一个开发用的 inspector 界面,你可以在里面看到 Server 暴露的所有工具,手动调用它们,查看返回结果。这一步能帮你排除掉大部分低级错误,比如工具没注册上、参数 schema 有问题、返回值格式不对等等。

如果 inspector 用不了,也可以写一个简单的测试脚本,直接用 SDK 的 Client 连接 Server:

import asyncio from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client async def test_server(): server_params = StdioServerParameters( command="python", args=["weather_server.py"] ) async with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() # 列出所有工具 tools = await session.list_tools() print("可用工具:") for tool in tools.tools: print(f" - {tool.name}: {tool.description}") # 调用工具 result = await session.call_tool("get_weather", {"city": "北京"}) print(f"\n调用结果:{result.content}") asyncio.run(test_server())

这个脚本跑通了,说明 Server 本身没问题,接下来就可以接到 LangGraph 里了。

注意:stdio 模式下,Server 的日志输出要特别小心。如果你用print打印调试信息,会污染标准输出,导致协议解析失败。正确的做法是用sys.stderr输出日志,或者用 logging 模块配置输出到文件。

4. 把 MCP 工具接进 LangGraph 工作流

Server 跑通了,现在到了最关键的一步:怎么把它接进 LangGraph。这里我会给出两种方案,一种是用现成的适配器,一种是手写适配层,你可以根据实际需求选择。

4.1 用 langchain-mcp-adapters 快速接入

如果你不想自己处理协议细节,用langchain-mcp-adapters是最省事的。它帮你把 MCP 工具转换成 LangChain Tool,可以直接塞进ToolNode。

from langchain_mcp_adapters.client import MultiServerMCPClient from langgraph.prebuilt import create_react_agent from langchain_openai import ChatOpenAI async def build_agent(): # 配置 MCP Server 连接 client = MultiServerMCPClient({ "weather": { "command": "python", "args": ["weather_server.py"], "transport": "stdio", } }) # 获取所有工具 tools = await client.get_tools() print(f"加载了 {len(tools)} 个工具") # 创建 Agent model = ChatOpenAI(model="gpt-4o") agent = create_react_agent(model, tools) return agent async def main(): agent = await build_agent() result = await agent.ainvoke({ "messages": [{"role": "user", "content": "北京今天天气怎么样?"}] }) print(result["messages"][-1].content)

这段代码的核心是MultiServerMCPClient,它支持同时连接多个 MCP Server,把它们的工具聚合在一起。get_tools()方法会返回一个 LangChain Tool 列表,直接传给create_react_agent就行。

实测下来,这个方案在大多数场景下够用了。但有几个限制你要知道:第一,它默认的工具发现是启动时做一次,如果 Server 的工具列表动态变化,需要手动刷新。第二,错误处理用的是默认逻辑,如果你想自定义重试或者降级策略,需要自己包装。第三,返回值的格式转换是固定的,如果你需要特殊处理,也得自己来。

4.2 手写适配层:什么时候需要,怎么写

什么时候需要手写适配层?我总结了几种情况:你需要对工具调用做精细化的错误处理和重试;你需要缓存工具发现的结果并支持动态更新;你需要对返回值做自定义的格式转换;或者你用的 LangGraph 版本跟适配器不兼容。

手写适配层的核心思路是:实现一个类,继承 LangChain 的BaseTool,在_arun方法里调用 MCP 的tools/call。

from langchain_core.tools import BaseTool from pydantic import BaseModel, create_model from mcp import ClientSession from typing import Any, Optional, Type class MCPToolWrapper(BaseTool): """把 MCP 工具包装成 LangChain Tool""" name: str description: str session: Any # MCP ClientSession mcp_tool_name: str args_schema: Optional[Type[BaseModel]] = None def _run(self, **kwargs) -> str: raise NotImplementedError("只支持异步调用") async def _arun(self, **kwargs) -> str: try: result = await self.session.call_tool( self.mcp_tool_name, kwargs ) # 把 MCP 返回的内容转换成字符串 texts = [] for content in result.content: if hasattr(content, "text"): texts.append(content.text) return "\n".join(texts) if texts else "工具执行完成,无返回内容" except Exception as e: return f"工具调用失败:{str(e)}" def mcp_tool_to_langchain(tool_def, session) -> MCPToolWrapper: """把 MCP 工具定义转换成 LangChain Tool""" # 从 JSON Schema 构建 Pydantic 模型 properties = tool_def.inputSchema.get("properties", {}) required = tool_def.inputSchema.get("required", []) fields = {} for prop_name, prop_schema in properties.items(): prop_type = str # 简化处理,实际需要根据 schema 类型映射 default = ... if prop_name in required else None fields[prop_name] = (prop_type, default) args_model = create_model(f"{tool_def.name}_args", **fields) return MCPToolWrapper( name=tool_def.name, description=tool_def.description or "", session=session, mcp_tool_name=tool_def.name, args_schema=args_model, )

这个适配层的代码量不大,但覆盖了核心逻辑。实际项目中,你可能需要根据 JSON Schema 的类型信息做更精细的映射,比如把integer映射成int,把array映射成List,把object映射成嵌套的 Pydantic 模型。

我手写适配层的时候踩过一个坑:MCP 工具的inputSchema里,参数类型是用 JSON Schema 表示的,但 LangChain 的args_schema需要的是 Pydantic 模型。这两者之间不是一一对应的,比如 JSON Schema 的oneOf在 Pydantic 里没有直接对应,需要转换成Union。如果你的工具用了复杂的 schema,转换逻辑要仔细处理。

4.3 在 StateGraph 里编排 MCP 工具调用

如果你不用create_react_agent,而是自己搭StateGraph,那工具调用的编排需要自己写。核心是加一个ToolNode,然后把工具列表传给它。

from langgraph.graph import StateGraph, MessagesState from langgraph.prebuilt import ToolNode, tools_condition from langchain_openai import ChatOpenAI async def build_graph(): client = MultiServerMCPClient({ "weather": { "command": "python", "args": ["weather_server.py"], "transport": "stdio", } }) tools = await client.get_tools() model = ChatOpenAI(model="gpt-4o").bind_tools(tools) def call_model(state: MessagesState): response = model.invoke(state["messages"]) return {"messages": [response]} graph = StateGraph(MessagesState) graph.add_node("agent", call_model) graph.add_node("tools", ToolNode(tools)) graph.set_entry_point("agent") graph.add_conditional_edges("agent", tools_condition) graph.add_edge("tools", "agent") return graph.compile()

这个图的结构很经典:agent 节点负责调用 LLM,LLM 决定是否调用工具;如果决定调用,就转到 tools 节点执行;执行完再回到 agent 节点,让 LLM 根据工具返回结果生成最终回复。

tools_condition是 LangGraph 提供的条件函数,它会检查 LLM 的返回消息里有没有 tool_calls,有就转到 tools 节点,没有就结束。这个函数省去了你自己写条件判断的麻烦。

实测下来,这个结构在工具数量不多的时候很好用。但如果工具很多(比如超过 20 个),LLM 选择工具的准确率会下降,因为 prompt 里塞太多工具描述会稀释注意力。这时候可以考虑做工具分组,或者用路由层先筛选一轮工具。

5. 踩坑实录:MCP 对接过程中最容易翻车的几个地方

这一节我专门讲踩过的坑,因为这些东西官方文档不会写,但实际项目中一定会遇到。

5.1 stdio 模式下 Server 进程的僵尸化问题

stdio 模式下,MCP Server 是作为 Host 的子进程启动的。如果 Host 异常退出(比如被 kill 掉),子进程可能不会被正确清理,变成僵尸进程。我遇到过一次:开发阶段频繁重启 Host,结果系统里积累了几十个 Python 进程,把内存吃满了。

解决方案是在 Host 退出时显式关闭 Client 连接。如果你用MultiServerMCPClient,它提供了close()方法,确保在 finally 块里调用。如果是自己管理连接,要确保stdio_client的上下文管理器正确退出。

client = MultiServerMCPClient({...}) try: tools = await client.get_tools() # ... 使用工具 finally: await client.close() # 确保子进程被清理

另外,在 Server 端也可以做一些防护,比如监听父进程的退出信号,收到信号后主动退出。不过这个要看具体的部署环境,不是所有场景都需要。

5.2 工具参数 schema 不匹配导致的静默失败

这个问题很隐蔽。MCP 工具的 schema 和 LangChain 的 args_schema 如果不匹配,有时候不会报错,而是静默失败——LLM 传了参数,但工具收到的参数是空的或者错的。

我遇到过一次:MCP 工具定义了一个可选参数unit,默认值是celsius。但在转换成 LangChain Tool 的时候,适配层把可选参数处理成了必填,导致 LLM 不传这个参数时调用直接失败。排查的时候看日志发现工具被调用了,但参数校验没通过,错误信息被吞掉了。

避免这个问题的办法是:在适配层里做好参数校验,把 MCP 的 schema 完整地转换成 LangChain 的 schema,包括默认值、可选性、类型约束。转换完之后,写个测试用例验证一下,确保 LLM 传各种参数组合都能正常工作。

5.3 工具返回结果过大导致的上下文溢出

MCP 工具返回的结果会作为消息塞进对话历史,如果返回结果太大,会迅速消耗 token 配额。我见过一个工具返回了一整张数据库表的数据,几万行,直接把上下文撑爆了。

解决方案有两个:一是在 Server 端做结果截断,只返回必要的信息;二是在适配层做结果压缩,比如把结构化数据转成摘要。我倾向于在 Server 端处理,因为 Server 更清楚哪些信息是必要的。

如果确实需要返回大量数据,可以考虑分页或者流式返回。MCP 协议支持流式返回,但 LangGraph 这边处理流式结果需要额外的适配工作,不是所有场景都值得做。

5.4 多个 Server 之间的工具命名冲突

如果你同时连接多个 MCP Server,不同 Server 可能有同名工具。比如两个 Server 都提供了一个叫search的工具,LangGraph 这边就会冲突。

MultiServerMCPClient的处理方式是在工具名前加 Server 名前缀,比如weather_search和news_search。但这个前缀会暴露给 LLM,可能影响 LLM 的工具选择。我的做法是在适配层做一层映射,把带前缀的名字映射回原始名字,同时在工具描述里说明这个工具属于哪个领域,帮助 LLM 做选择。

5.5 错误处理:MCP 的错误怎么传递给 LangGraph

MCP 工具执行失败时,Server 会返回一个错误响应。这个错误需要被正确地传递给 LangGraph,让 Agent 知道工具调用失败了,可以决定重试或者换一个工具。

默认情况下,适配器会把错误包装成异常抛出,LangGraph 的ToolNode会捕获这个异常,把它作为工具消息返回给 LLM。但有时候错误信息不够清晰,LLM 看不懂,就会反复重试同一个工具。

我的做法是在适配层对错误做分类处理:可重试的错误(比如网络超时)返回一个提示 LLM 重试的消息;不可重试的错误(比如参数错误)返回一个明确的错误说明,让 LLM 换一种方式。这样能显著提高 Agent 的鲁棒性。

6. 进阶玩法:动态工具发现与多 Server 编排

基础的东西讲完了,这一节聊点进阶的。如果你已经把基本的 MCP 对接跑通了,可以看看这些玩法。

6.1 运行时动态发现工具并更新 Agent

前面说的工具发现都是启动时做一次,但有些场景下工具会动态变化。比如你有一个工具服务,会根据用户的订阅级别暴露不同的工具集。用户升级了订阅,工具列表就变了,这时候 Agent 需要能感知到变化。

MCP 协议支持tools/list_changed通知,Server 可以在工具列表变化时主动通知 Client。Client 收到通知后,重新拉取工具列表,更新 Agent 的工具配置。

实现这个的难点在于 LangGraph 的图是编译后不可变的,你不能在运行时直接修改图里的工具列表。我的做法是把工具列表放在一个可变的容器里,ToolNode每次执行时从容器里读取最新的工具列表。这样图本身不变,但工具列表可以动态更新。

class DynamicToolNode: def __init__(self): self._tools = [] def update_tools(self, tools): self._tools = tools async def __call__(self, state): # 每次调用时使用最新的工具列表 tool_node = ToolNode(self._tools) return await tool_node(state)

这个方案不是最优雅的,但实测能用。更优雅的方案是等 LangGraph 官方支持动态工具,或者用子图的方式把工具节点隔离出来。

6.2 多个 MCP Server 的编排策略

当你连接多个 MCP Server 时,工具数量可能很多,LLM 选择工具的准确率会下降。这时候需要一些编排策略。

第一种策略是按领域分组。把工具按功能领域分组,每组对应一个子 Agent,主 Agent 负责路由到子 Agent。比如天气相关的工具给天气 Agent,新闻相关的给新闻 Agent。这样每个子 Agent 的工具列表都比较短,选择准确率高。

第二种策略是两阶段选择。先用一个轻量级的模型或者规则引擎筛选出候选工具,再把候选工具交给主 LLM 做最终选择。这样既保证了准确率,又控制了 prompt 长度。

第三种策略是工具描述优化。给每个工具写清晰、区分度高的描述,让 LLM 能快速判断哪个工具适合当前任务。这个听起来简单,但实际效果很好。我做过对比测试,优化工具描述之后,工具选择的准确率提升了差不多 20%。

6.3 把 MCP 工具服务部署成远程 SSE 服务

前面用的都是 stdio 模式,适合本地开发。生产环境通常需要把工具服务部署成远程服务,这时候用 SSE 模式。

SSE 模式下,Server 作为一个 HTTP 服务运行,Client 通过 URL 连接。部署方式跟普通的 Web 服务一样,可以用 uvicorn 或者 gunicorn。

# server.py from mcp.server.fastmcp import FastMCP mcp = FastMCP("weather-service") @mcp.tool() async def get_weather(city: str) -> str: # ... 工具实现 pass if __name__ == "__main__": mcp.run(transport="sse", host="0.0.0.0", port=8080)

Client 这边连接的时候,把 transport 改成 sse,并提供 URL:

client = MultiServerMCPClient({ "weather": { "url": "http://localhost:8080/sse", "transport": "sse", } })

SSE 模式的好处是部署灵活,可以独立扩缩容。但要注意几个问题:第一,网络延迟比 stdio 高,对延迟敏感的场景要谨慎;第二,需要处理连接断开和重连;第三,鉴权和限流要在 Server 端做好,因为服务暴露在网络上了。

我在生产环境用 SSE 模式的时候,加了一层 API Gateway 做鉴权和限流,Server 本身只负责工具逻辑。这样职责清晰,也方便监控和运维。

7. 一些实测有效的经验和小技巧

最后分享几个我在实际项目中总结的经验,都是踩过坑之后才明白的。

工具描述要写给 LLM 看,不是写给人看。很多人写工具描述的时候习惯用技术术语,但 LLM 不一定能理解。好的工具描述应该用自然语言说明这个工具做什么、什么时候用、参数是什么意思。我通常会写一个"使用场景"字段,明确告诉 LLM 什么情况下应该调用这个工具。

参数校验要在 Server 端做,不要依赖 Client。Client 端的校验只是第一道防线,不能保证所有请求都是合法的。Server 端一定要做完整的参数校验,包括类型、范围、必填项。我见过因为 Server 端没做校验,导致非法参数直接打到数据库,出了生产事故。

工具调用要有超时控制。MCP 工具调用默认没有超时,如果工具执行卡住了,整个 Agent 就会挂起。我的做法是在适配层加超时控制,超过阈值就中断调用,返回超时错误。超时时间根据工具的实际执行时间设定,一般设成 P99 执行时间的两倍。

日志要打全,但要注意脱敏。MCP 工具调用的日志对排查问题很重要,但要小心不要把敏感信息(比如 API 密钥、用户数据)打到日志里。我的做法是在适配层做日志脱敏,只记录工具名称、参数结构(不记录具体值)、执行时间、返回状态。

版本兼容性要提前测试。MCP 协议还在演进,不同版本之间可能有 breaking change。升级 SDK 之前,一定要在测试环境跑一遍完整的回归测试,确保所有工具都能正常工作。我一般会在 requirements.txt 里锁定版本,升级的时候手动改,避免自动升级带来的意外。

工具粒度要适中。工具太细,LLM 需要调用很多次才能完成一个任务,效率低;工具太粗,灵活性差,LLM 没法组合使用。我的经验是:一个工具做一件事,但这件事要有足够的业务含义。比如"查询天气"是一个好工具,"发送 HTTP 请求"就太底层了。

给工具加上幂等性保证。Agent 可能会因为各种原因重试工具调用,如果工具不是幂等的,重试会导致重复操作。比如"创建订单"这种工具,重试两次就会创建两个订单。解决方案是给工具加上幂等键,Server 端根据幂等键去重。

这些经验看起来都是细节,但实际项目中,正是这些细节决定了系统能不能稳定运行。MCP 协议本身不复杂,复杂的是怎么把它跟现有的系统集成好,怎么处理各种边界情况。希望这些经验能帮你少走一些弯路。

版权声明: 本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若内容造成侵权/违法违规/事实不符,请联系邮箱:809451989@qq.com进行投诉反馈,一经查实,立即删除!
网站建设 2026/10/30 19:14:48

Superpowers技能包:用标准化Skills提升开发效率与自动化能力

最近一直在研究怎么把手头常用的开发能力整理成一套可以随取随用的"技能弹药库",接触到一个叫 Superpowers 的开源技能包,试用了一段时间,确实把日常编码里那些重复、费神、容易出错的环节简化了不少。这篇文章就围绕 Superpowers …

作者头像 李华
网站建设 2026/10/23 13:22:26

PaperXM:AI驱动的学术论文写作工具核心技术解析

1. 论文写作工具现状与痛点分析作为一名科研工作者,我深知论文写作过程中面临的种种困扰。从文献综述到实验设计,从数据分析到结论撰写,每个环节都需要投入大量时间精力。近年来,AI写作辅助工具如雨后春笋般涌现,但真正…

作者头像 李华
网站建设 2026/10/24 6:03:42

超帧数据聚合与运动补偿:提升多传感器融合与SLAM精度的关键技术

做机器人和自动驾驶的同行,对frame这个词肯定不陌生。点云有 frame,图像有 frame,IMU 有 frame,连路径规划里的 costmap 也分 frame。这两年我一直在折腾一件事——把一系列原始数据帧组织成hyperframes再用起来,让里程…

作者头像 李华
网站建设 2026/10/23 15:39:42

SpringBoot+Vue景区民宿预约系统开发实战

1. 项目概述:景区民宿预约系统的技术价值与应用场景这个基于SpringBootVue的景区民宿预约系统管理平台,本质上是一个面向旅游行业的数字化解决方案。我在实际开发这类系统时发现,它完美解决了传统民宿经营中的三大痛点:手工登记易…

作者头像 李华
网站建设 2026/10/26 10:07:42

大模型Tool Use手写实战:三种方案与面试避坑指南

1. 面试现场:从“八股文”到“手写 Tool Use”的转向1.1 为什么 Tool Use 成了面试新宠这两年技术面试的风向变得特别快。前几年大家还在卷 Transformer 的注意力公式怎么推导、BatchNorm 的反向传播怎么算,现在面试官往桌上一靠,直接来一句&…

作者头像 李华
网站建设 2026/10/26 16:18:38

Unity WebGL数据交互:ASP.NET Core Web API连接SqlServer实战详解

我在做一款浏览器端的3D可视化项目时,遇到一个绕不开的需求:玩家的存档记录、操作日志和实时排行榜数据要落到数据库里。项目本身是Unity WebGL打包后运行在浏览器中的,数据后端选定的是ASP.NET Web API,数据库用的是SqlServer。这…

作者头像 李华