在LLM应用的快速发展中,一个核心挑战始终存在:如何让模型获取最新、最准确的外部知识并有效利用工具?
背景其实很简单:大模型(LLM)再强,也总有不知道的东西,怎么办?让它“查资料”“调工具”成了近两年最热的技术方向。从最早的 RAG(Retrieval-Augmented Generation),到 OpenAI 引领的 Function Call,再到现在 Anthropic 抛出的 MCP(Model Context Protocol),每一代方案都在试图解答一个问题:模型如何以更自然的方式获得外部世界的帮助?
MCP 主打的是统一标准和跨模型兼容性。虽然协议本身尚处于早期阶段,设计也远称不上完美,但出现的时机十分巧妙。。就像当年 OpenAI 的 API,一旦形成事实标准,后面哪怕有点毛病,也可以很快改进,毕竟生态具有滚雪球效应,一旦用户基数形成规模,自然而然就成为事实标准。
本篇文章将结合 MCP 官方 SDK,通过代码和流程图模拟一次带 Tool 调用的完整交互过程,了解并看清 MCP 的全生命周期。
整体流程
一次MCP完整的调用流程如下:
图1. 一次包含MCP调用的完整流程
图1省略了第一步与第二步之间,list_tools()或resource()的步骤,也就是最开始MCP Host知道有哪些可用的工具与资源,我们在本 DEMO 中使用了硬编码的方式将资源信息构建在提示词中。
这里需要注意的是MCP Client与MCP Host(主机)并不是分离的部分,但为了时序图清晰,这里将其逻辑上拆分为不同的部分,实际上MCP Host可以理解为我们需要嵌入AI的应用程序,例如 CRM 系统或 SaaS 服务,实际上Host中是包含MCP Client的代码。实际的 MCP Host 与 Client 结构如下图所示:
整体示例代码
MCP Server
mcp server的代码使用最简单的方式启动,并通过Python装饰器注册最简单的两个工具,为了DEMO简单,hard code两个工具(函数)返回值,代码如下:
#mcp_server_demo.pyfrom mcp.server.fastmcp import FastMCPimport asynciomcp = FastMCP(name="weather-demo", host="0.0.0.0", port=1234)@mcp.tool(name="get_weather", description="获取指定城市的天气信息")async def get_weather(city: str) -> str: """ 获取指定城市的天气信息 """ weather_data = { "北京": "北京:晴,25°C", "上海": "上海:多云,27°C" } return weather_data.get(city, f"{city}:天气信息未知")@mcp.tool(name="suggest_activity", description="根据天气描述推荐适合的活动")async def suggest_activity(condition: str) -> str: """ 根据天气描述推荐适合的活动 """ if "晴" in condition: return "天气晴朗,推荐你去户外散步或运动。" elif "多云" in condition: return "多云天气适合逛公园或咖啡馆。" elif "雨" in condition: return "下雨了,建议你在家阅读或看电影。" else: return "建议进行室内活动。"async def main(): print("✅ 启动 MCP Server: http://127.***0.0.1:1234") await mcp.run_sse_async()if __name__ == "__main__": asyncio.run(main())
大模型调用代码
大模型调用选择使用openrouter这个LLM的聚合网站,主要是因为该网站方便调用与测试不同的模型,同时网络环境可以直接连接而不用其他手段。
代码如下:
# llm_router.pyimport jsonimport requests# OpenRouter 配置OPENROUTER_API_KEY = '这里写入使用的Key'OPENROUTER_API_URL = "https://o**penrou*ter.ai/api/v1/chat/completions"OPENROUTER_HEADERS = { "Authorization": f"Bearer {OPENROUTER_API_KEY}", "Content-Type": "application/json", "HTTP-Referer": "http://local*ho*s*t", "X-Title": "MCP Demo Server"}class OpenRouterLLM: """ 自定义 LLM 类,使用 OpenRouter API 来生成回复 """ def __init__(self, model: str = LLM_MODEL): self.model = model def generate(self, messages): """ 发送对话消息给 OpenRouter API 并返回 LLM 的回复文本 参数: messages: 一个 list,每个元素都是形如 {'role': role, 'content': content} 的字典 返回: LLM 返回的回复文本 """ request_body = { "model": self.model, "messages": messages } print(f"发送请求到 OpenRouter: {json.dumps(request_body, ensure_ascii=False)}") response = requests.post( OPENROUTER_API_URL, headers=OPENROUTER_HEADERS, json=request_body ) if response.status_code != 200: print(f"OpenRouter API 错误: {response.status_code}") print(f"错误详情: {response.text}") raise Exception(f"OpenRouter API 返回错误: {response.status_code}") response_json = response.json() print(f"OpenRouter API 响应: {json.dumps(response_json, ensure_ascii=False)}") # 提取 LLM 响应文本 try: content = response_json['choices'][0]['message']['content'] return content except KeyError: raise Exception("无法从 OpenRouter 响应中提取内容")# 如果需要独立测试该模块,可以在此进行简单的测试if __name__ == "__main__": # 示例系统提示和用户输入 messages = [ {"role": "system", "content": "你是一个智能助手,可以帮助查询天气信息。"}, {"role": "user", "content": "请告诉我北京今天的天气情况。"} ] llm = OpenRouterLLM() try: result = llm.generate(messages) print("LLM 返回结果:") print(result) except Exception as e: print(f"调用 OpenRouter 时发生异常: {e}")
MCP Client
这里的MCP Client,使用Server-Side Event(SSE)方式进行连接(题外话,MCP协议使用SSE协议作为默认远程协议稍微有点奇怪,听说后续迭代会考虑HTTP Streaming以及JSONRPC over HTTP2的方式)。
这里我们在main测试代码中,尝试列出所有可用的Tool与Resource,并尝试调用Tool,结果如图,可以看到能够展示出MCP Server中定义的Tool。
# mcp_client_demo.pyimport asynciofrom mcp.client.session import ClientSessionfrom mcp.client.sse import sse_clientclass WeatherMCPClient: def __init__(self, server_url="http://127.***0.0.1:1234/sse"): self.server_url = server_url self._sse_context = None self._session = None async def __aenter__(self): # 创建 SSE 通道 self._sse_context = sse_client(self.server_url) self.read, self.write = await self._sse_context.__aenter__() # 创建 MCP 会话 self._session = ClientSession(self.read, self.write) await self._session.__aenter__() await self._session.initialize() return self async def __aexit__(self, exc_type, exc_val, exc_tb): if self._session: await self._session.__aexit__(exc_type, exc_val, exc_tb) if self._sse_context: await self._sse_context.__aexit__(exc_type, exc_val, exc_tb) async def list_tools(self): return await self._session.list_tools() async def list_resources(self): return await self._session.list_resources() async def call_tool(self, name, arguments): return await self._session.call_tool(name, arguments)async def main(): async with WeatherMCPClient() as client: print("✅ 成功连接 MCP Server") tools = await client.list_tools() print("n? 可用工具:") print(tools) resources = await client.list_resources() print("n? 可用资源:") print(resources) print("n? 调用 WeatherTool 工具(city=北京)...") result = await client.call_tool("get_weather", {"city": "北京"}) print("n? 工具返回:") for item in result.content: print(" -", item.text)if __name__ == "__main__": asyncio.run(main())
MCP Host
MCP host的角色也就是我们需要嵌入AI的应用,可以是一个程序,可以是一个CRM系统,可以是一个OA,MCP Host包含MCP Client,用于集成LLM与Tool,MCP Host之外+Tool+大模型,共同构成了一套基于AI的系统,现在流行的说法是AI Agent(中文翻译:AI智能体?)
MCP Host代码中步骤注释,与图1中的整体MCP流程对齐。
import asyncioimport jsonimport refrom llm_router import OpenRouterLLMfrom mcp_client_demo import WeatherMCPClientdef extract_json_from_reply(reply: str): """ 提取 LLM 返回的 JSON 内容,自动处理 markdown 包裹、多余引号、嵌套等。 支持 string 或 dict 格式。 如果无法解出 dict,则返回原始 string。 """ # 如果已经是 dict,直接返回 if isinstance(reply, dict): return reply # 清除 markdown ```json ``` 包裹 if isinstance(reply, str): reply = re.sub(r"^```(?:json)?|```$", "", reply.strip(), flags=re.IGNORECASE).strip() # 最多尝试 3 层 json.loads 解码 for _ in range(3): try: parsed = json.loads(reply) if isinstance(parsed, dict): return parsed else: reply = parsed # 如果解出来还是 str,继续下一层 except Exception: break # 如果最终不是 dict,返回原始字符串(表示是普通答复) return replyllm = OpenRouterLLM()async def main(): # === 初始化 MCP 客户端 === client = WeatherMCPClient() await client.__aenter__() tools = await client.list_tools() resources = await client.list_resources() tool_names = [t.name for t in tools.tools] tool_descriptions = "n".join(f"- {t.name}: {t.description}" for t in tools.tools) resource_descriptions = "n".join(f"- {r.uri}" for r in resources.resources) while True: # === Step 1. 用户 → MCP主机:提出问题 === user_input = input("n请输入你的问题(输入 exit 退出):n> ") if user_input.lower() in ("exit", "退出"): break # 构造系统提示 + 工具说明 system_prompt = ( "你是一个智能助手,拥有以下工具和资源可以调用:nn" f"? 工具列表:n{tool_descriptions or '(无)'}nn" f"? 资源列表:n{resource_descriptions or '(无)'}nn" "请优先调用可用的Tool或Resource,而不是llm内部生成。仅根据上下文调用工具,不传入不需要的参数进行调用n" "如果需要,请以 JSON 返回 tool_calls,格式如下:n" '{"tool_calls": [{"name": "get_weather", "arguments": {"city": "北京"}}]}n' "如无需调用工具,返回:{"tool_calls": null}" ) # === 构造 LLM 上下文消息 === messages = [ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_input} ] final_reply = "" # === 循环处理 tool_calls,直到 LLM 给出最终 content 为止 === while True: # === Step 2. MCP主机 → LLM:转发上下文 === reply = llm.generate(messages) print(f"n? LLM 回复:n{reply}") # === Step 3. 解析 JSON 格式回复(或普通字符串) === parsed = extract_json_from_reply(reply) # === 如果是普通自然语言字符串,说明 LLM 已直接答复用户 === if isinstance(parsed, str): final_reply = parsed break # === 如果是字典,判断是否包含工具调用 === tool_calls = parsed.get("tool_calls") if not tool_calls: # LLM 给出普通答复结构(带 content 字段) final_reply = parsed.get("content", "") break # === 遍历 LLM 请求的工具调用列表 === for tool_call in tool_calls: # === Step 4. LLM → MCP客户端:请求使用工具 === tool_name = tool_call["name"] arguments = tool_call["arguments"] if tool_name not in tool_names: raise ValueError(f"❌ 工具 {tool_name} 未注册") # === Step 5. MCP客户端 → MCP服务器:调用工具 === print(f"? 调用工具 {tool_name} 参数: {arguments}") result = await client.call_tool(tool_name, arguments) # === Step 8. MCP服务器 → MCP客户端:返回结果 === tool_output = result.content[0].text print(f"? 工具 {tool_name} 返回:{tool_output}") # === Step 9. MCP客户端 → LLM:提供工具结果 === messages.append({ "role": "tool", "name": tool_name, "content": tool_output }) # Step 10: 再次调用 LLM,进入下一轮(可能再次产生 tool_calls) # === Step 11. MCP主机 → 用户:最终结果答复 === print(f"n? 最终答复:{final_reply}") await client.__aexit__(None, None, None)if __name__ == "__main__": asyncio.run(main())
用户提问
DEMO的交互方式是一个简单的Chatbox。假设用户在聊天界面的输入框里敲下:“上海的天气如何” 。此时,用户的问题通过 MCP 主机(MCP Host) 被发送给大模型。
MCP Host 可以是一个浏览器前端、桌面应用,也可以只是后端的一段代码。在这个场景里,它主要负责收集用户输入并与LLM通信。
对应流程图1中的步骤1:提出问题 与步骤2:转发问题。
LLM 推理:是否需要外部Tool配合
收到用户提问后,MCP 主机(Host)负责将用户提问解析并附加上下文后转发给大模型。主要取决于系统设计的智能程度、工具丰富度,以及 LLM 的能力边界。通常可以是一段静态的提示词,或者从上下文中获取动态的提示词,也可以是通过一些外部API获取数据生成提示词,这并不是本文的重点,本文通过简单的静态提示词进行。
本DEMO的静态提示词如下:
# 构造系统提示 + 工具说明 system_prompt = ( "你是一个智能助手,拥有以下工具和资源可以调用:nn" f" 工具列表:n{tool_descriptions or '(无)'}nn" f" 资源列表:n{resource_descriptions or '(无)'}nn" "请优先调用可用的Tool或Resource,而不是llm内部生成。仅根据上下文调用工具,不传入不需要的参数进行调用n" "如果需要,请以 JSON 返回 tool_calls,格式如下:n" '{"tool_calls": [{"name": "get_weather", "arguments": {"city": "北京"}}]}n' "如无需调用工具,返回:{"tool_calls": null}" )
注意:MCP 协议与传统 Function Calling 最大的区别在于:工具调用的时机、选择和参数完全由大模型基于上下文和系统提示词自主推理决策,而不是由应用层预先决定调用哪个工具。这种模型主导的调用方式(model-driven invocation)体现了 Agent 思维,MCP 由此成为构建AI Agent 的关键协议基础之一。
LLM 此时会分析用户的问题:“上海的天气如何?” 如果这是一个普通常识性问题,LLM 也许可以直接作答;但这里问的是实时天气,超出了模型自身知识(训练数据可能并不包含最新天气)。此时的 LLM 就像进入一个未知领域,它明确知道需要外部信息的帮助来解答问题。在DEMO的会话开始时,MCP 主机已经通告诉 LLM 可以使用哪些工具(例如提供天气的工具叫 “get_weather”)。因此 LLM 判断:需要触发一次 Tool Call 来获取答案。
在代码实现上,LLM 模型被提示可以调用工具。当模型决定调用时(对应图1中的步骤4),会生成一段特殊的结构化信息(通常是 JSON)。比如我们的 LLM 可能返回如下内容而不是直接答案:
{ "tool_calls": [ { "name": "get_weather", "arguments": { "city": "上海" } } ]}
上面 JSON 表示:LLM请求使用名为“get_weather”的工具,并传递参数城市为“上海”。MCP 主机的 模块会检测到模型输出的是一个 Tool Call 请求 而非普通文本答案——通常通过判断返回是否是合法的 JSON、且包含预期的字段来确认。这一刻,LLM 相当于对主机说:“我需要用一下get_weather工具帮忙查一下上海天气的天气!”
日志中可以看到这一决策过程:
? LLM 回复:{"tool_calls": [{"name": "get_weather", "arguments": {"city": "上海"}}]}? 调用工具 get_weather 参数: {'city': '上海'}
如果 LLM 能直接回答问题(不需要工具),那么它会返回纯文本,MCP 主机则会直接将该回答返回给客户端,完成整个流程。而在本例中, 需要外部Tool获取数据。
Tool Call 发起与数据获取
LLM 向MCP Host发起 Tool Call 请求(对应图1中的步骤5),MCP 主机现在扮演起“信使”的角色,通过MCP Client将这个请求转交给对应的 MCP 服务器。MCP 服务器可以看作提供特定工具或服务的后端,比如一个天气信息服务。我们在示例代码 mcp_host_demo.py 中,会调用 MCP 客户端模块(与 MCP Server 通信的组件)发送请求,例如:result = mcp_client.call_tool(tool_name, args)。
此时日志可能会出现:
? 调用工具 get_weather 参数: {'city': '上海'}
MCP 服务器收到请求后,开始处理实际的数据查询。在我们的例子中,MCP Server 内部知道 get_weather如何获取天气数据(本例中是硬编码,但通常应该是一个外部API接口)。它会向数据源(可能是一个实时天气数据库或API)请求上海当前的天气。示例代码 mcp_server_demo.py 中定义了 硬编码的get_weather 工具的实现(因此也就忽略了图1中从mcp server与后端数据源的交互,步骤6与步骤7)
接下来,MCP 服务器将拿到的数据打包成结果返回。根据 MCP 协议规范,结果通常也用 JSON 表示,这里使用MCP Python SDK解析后的字符串结果:
result = await client.call_tool(tool_name, arguments)# === Step 8. MCP服务器 → MCP客户端:返回结果 ===tool_output = result.content[0].textprint(f"? 工具 {tool_name} 返回:{tool_output}")
在控制台日志里,我们可以看到:
工具 get_weather 返回:上海:多云,27°C
可以看到,MCP 服务器既完成了实际的数据获取,又把结果封装成统一格式返回给MCP Host。整个过程对于 LLM 和客户端来说是透明的:他们不需要关心天气数据具体来自哪个数据库或API,只需通过 MCP 协议与服务器交互即可。这体现了 MCP 模块化的设计理念——Tool的实现细节被封装在MCP Server中,对外提供标准接口。
结果返回与答案生成
现在MCP 主机从 MCP 服务器拿到了工具调用结果,接下来要做的是把结果交还给最初发起请求的 LLM,让它完成最终答案生成。
在我们的示例中,MCP 主机收到了 get_weather 的结果 JSON。MCP 主机会将该结果作为新的输入提供给 LLM。常见做法是将工具返回的结果附加到发送给LLM的对话中:
{ "model": "qwen/qwen2.5-vl-32b-instruct:free", "messages": [ { "role": "system", "content": "你是一个智能助手,拥有以下工具和资源可以调用:nn? 工具列表:n- get_weather: 获取指定城市的天气信息n- suggest_activity: 根据天气描述推荐适合的活动nn? 资源列表:n(无)nn请优先调用可用的Tool或Resource,而不是llm内部生成。仅根据上下文调用工具,不传入不需要的参数进行调用"北京"}}]}n如无需调用工具,返回:{"tool_calls": null}" }, { "role": "user", "content": "上海的天气如何?" }, { "role": "tool", "name": "get_weather", "content": "上海:多云,27°C" } ]}
注意到新增的role: tool,这代表工具返回的信息作为上下文提供给LLM。
现在LLM 得到真实的天气数据后,拥有足够的数据,可以给出用户想要的答复了。对于用户问的“现在上海的天气怎么样?”,模型现在知道上海天气晴朗,27°C左右。它组织语言,将信息融入自然的回答中。例如,模型产出:“上海的天气是多云,温度为27°C。根据当前天气条件,建议您进行室内活动。”
MCP 主机接收到来自 LLM 的最终回答文本后,会将其发送回先前等待的 MCP 客户端。客户端则将答案显示给用户。至此,一次完整的问答闭环结束,用户收到满意的答复,而背后经过的一系列 Tool Call 流程对用户来说几乎无感
在用户看来,聊天对话可能长这样:
用户:上海的助手:上海的天气是多云,温度为27°C。根据当前天气条件,建议您进行室内活动。
小结
通过上述实例,我们能直观感受到 MCP 架构在设计上的独特优势。它明确了 LLM 应用中的职责划分,让语言理解与工具调用两个不同的职责有效解耦,实现了更高的系统灵活性:
- 模块化易扩展:添加新的工具服务只需实现一个独立的 MCP Server 即可,完全不需要改动 LLM 本身代码。无论是新增股票信息、日程安排或是其他功能,只需符合 MCP 协议标准,新服务即可迅速上线。
- 接口统一标准化:MCP 清晰定义了请求和响应的标准化格式,极大降低了接入新工具的成本。开发者无需再为每种工具分别设计集成逻辑,统一 JSON Schema 接口,使得调试和维护更加直观、高效。
- 实时能力增强:MCP 使 LLM 可以实时获取外部信息,突破模型训练数据的时效限制。诸如天气、新闻、股票行情甚至实时数据库查询等需求,都能轻松满足,从而大幅提升模型的实用性。
- 安全控制精细化:由于工具调用被隔离在独立的 MCP Server 中,开发者可针对具体工具执行细粒度的权限和安全管理,有效避免了 LLM 直接运行任意代码的风险。
- 故障易于追踪处理:错误消息通过标准协议明确返回,方便 LLM 做出合适的错误处理与用户反馈,有效提升用户体验及系统稳定性。
此外,MCP 未来还有许多潜在的拓展方向,例如支持多步工具链调用,使得 LLM 可以高效完成更复杂的任务;或者实现动态的工具发现与调用机制,让 LLM 能够根据实际需求自主选择工具。