通过一个DEMO理解MCP（模型上下文协议）的生命周期

黑鹰七代 發表於 2025-4-15 21:09:00

通过一个DEMO理解MCP（模型上下文协议）的生命周期

在LLM应用的快速发展中，一个核心挑战始终存在：如何让模型获取最新、最准确的外部知识并有效利用工具？
背景其实很简单：大模型（LLM）再强，也总有不知道的东西，怎么办？让它“查资料”“调工具”成了近两年最热的技术方向。从最早的 RAG（Retrieval-Augmented Generation），到 OpenAI 引领的 Function Call，再到现在 Anthropic 抛出的 MCP（Model Context Protocol），每一代方案都在试图解答一个问题：模型如何以更自然的方式获得外部世界的帮助？
MCP 主打的是统一标准和跨模型兼容性。虽然协议本身尚处于早期阶段，设计也远称不上完美，但出现的时机十分巧妙。。就像当年 OpenAI 的 API，一旦形成事实标准，后面哪怕有点毛病，也可以很快改进，毕竟生态具有滚雪球效应，一旦用户基数形成规模，自然而然就成为事实标准。
本篇文章将结合 MCP 官方 SDK，通过代码和流程图模拟一次带 Tool 调用的完整交互过程，了解并看清 MCP 的全生命周期。
<h1>整体流程</h1>
一次MCP完整的调用流程如下：
<img src="https://oss-ata.alibaba.com/article/2025/04/dab51e57-d0e5-47d8-8bac-a65d703fee79.png">
 
图1. 一次包含MCP调用的完整流程
图1省略了第一步与第二步之间，list_tools()或resource()的步骤，也就是最开始MCP Host知道有哪些可用的工具与资源，我们在本 DEMO 中使用了硬编码的方式将资源信息构建在提示词中。
这里需要注意的是MCP Client与MCP Host（主机）并不是分离的部分，但为了时序图清晰，这里将其逻辑上拆分为不同的部分，实际上MCP Host可以理解为我们需要嵌入AI的应用程序，例如 CRM 系统或 SaaS 服务，实际上Host中是包含MCP Client的代码。实际的 MCP Host 与 Client 结构如下图所示：
 
<img src="https://oss-ata.alibaba.com/article/2025/04/d2d10216-9340-49ea-bc7b-21454f5d5f3f.png">
<h1>整体示例代码</h1>
<h2>MCP Server</h2>
mcp server的代码使用最简单的方式启动，并通过Python装饰器注册最简单的两个工具，为了DEMO简单，hard code两个工具（函数）返回值，代码如下：
<pre class="highlighter-hljs"><code>#mcp_server_demo.py
from mcp.server.fastmcp import FastMCP
import asyncio

mcp = 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())</code></pre>
<img src="https://oss-ata.alibaba.com/article/2025/04/9aa5d1a3-c839-4dea-82a7-61cf0d93fe7d.png">
 
<h2>大模型调用代码</h2>
大模型调用选择使用openrouter这个LLM的聚合网站，主要是因为该网站方便调用与测试不同的模型，同时网络环境可以直接连接而不用其他手段。
代码如下：
<pre class="highlighter-hljs"><code># llm_router.py
import json
import requests

# OpenRouter 配置
OPENROUTER_API_KEY = '这里写入使用的Key'
OPENROUTER_API_URL = "https://openrouter.ai/api/v1/chat/completions"

OPENROUTER_HEADERS = {
"Authorization": f"Bearer {OPENROUTER_API_KEY}",
"Content-Type": "application/json",
"HTTP-Referer": "http://localhost",
"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']['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}")</code></pre>
 
<h2>MCP Client</h2>
这里的MCP Client，使用Server-Side Event（SSE）方式进行连接（题外话，MCP协议使用SSE协议作为默认远程协议稍微有点奇怪，听说后续迭代会考虑HTTP Streaming以及JSONRPC over HTTP2的方式）。
这里我们在main测试代码中，尝试列出所有可用的Tool与Resource，并尝试调用Tool，结果如图，可以看到能够展示出MCP Server中定义的Tool。
<pre class="highlighter-hljs"><code># mcp_client_demo.py
import asyncio
from mcp.client.session import ClientSession
from mcp.client.sse import sse_client

class 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())</code></pre>
<img src="https://oss-ata.alibaba.com/article/2025/04/cd1f471a-5114-4554-9da7-ed0e40c0a41c.png">
 
<h2>MCP Host</h2>
MCP host的角色也就是我们需要嵌入AI的应用，可以是一个程序，可以是一个CRM系统，可以是一个OA，MCP Host包含MCP Client，用于集成LLM与Tool，MCP Host之外+Tool+大模型，共同构成了一套基于AI的系统，现在流行的说法是AI Agent（中文翻译：AI智能体？）
MCP Host代码中步骤注释，与图1中的整体MCP流程对齐。
<pre class="highlighter-hljs"><code>import asyncio
import json
import re
from llm_router import OpenRouterLLM
from mcp_client_demo import WeatherMCPClient

def 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 reply

llm = OpenRouterLLM()

async def main():
# === 初始化 MCP 客户端 ===
client = WeatherMCPClient()
await client.__aenter__()

tools = await client.list_tools()
resources = await client.list_resources()
tool_names =

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 = (
 "你是一个智能助手，拥有以下工具和资源可以调用：\n\n"
 f"🛠 工具列表：\n{tool_descriptions or '（无）'}\n\n"
 f"📚 资源列表：\n{resource_descriptions or '（无）'}\n\n"
 "请优先调用可用的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.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())</code></pre>
 
 
<h1>用户提问</h1>
 
DEMO的交互方式是一个简单的Chatbox。假设用户在聊天界面的输入框里敲下：“上海的天气如何” 。此时，用户的问题通过 MCP 主机（MCP Host） 被发送给大模型。
MCP Host 可以是一个浏览器前端、桌面应用，也可以只是后端的一段代码。在这个场景里，它主要负责收集用户输入并与LLM通信。
对应流程图1中的步骤1：提出问题 与步骤2：转发问题。
 
<h1>LLM 推理：是否需要外部Tool配合</h1>
收到用户提问后，MCP 主机（Host）负责将用户提问解析并附加上下文后转发给大模型。主要取决于系统设计的智能程度、工具丰富度，以及 LLM 的能力边界。通常可以是一段静态的提示词，或者从上下文中获取动态的提示词，也可以是通过一些外部API获取数据生成提示词，这并不是本文的重点，本文通过简单的静态提示词进行。
本DEMO的静态提示词如下：
<pre class="highlighter-hljs"><code> # 构造系统提示 + 工具说明
 system_prompt = (
 "你是一个智能助手，拥有以下工具和资源可以调用：\n\n"
 f" 工具列表：\n{tool_descriptions or '（无）'}\n\n"
 f" 资源列表：\n{resource_descriptions or '（无）'}\n\n"
 "请优先调用可用的Tool或Resource，而不是llm内部生成。仅根据上下文调用工具，不传入不需要的参数进行调用\n"
 "如果需要，请以 JSON 返回 tool_calls，格式如下：\n"
 '{"tool_calls": [{"name": "get_weather", "arguments": {"city": "北京"}}]}\n'
 "如无需调用工具，返回：{\"tool_calls\": null}"
 )</code></pre>
注意：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 可能返回如下内容而不是直接答案：
<pre class="highlighter-hljs"><code>{
"tool_calls": [
{
 "name": "get_weather",
 "arguments": {
 "city": "上海"
 }
}
]
}</code></pre>
上面 JSON 表示：LLM请求使用名为“get_weather”的工具，并传递参数城市为“上海”。MCP 主机的  模块会检测到模型输出的是一个 Tool Call 请求 而非普通文本答案——通常通过判断返回是否是合法的 JSON、且包含预期的字段来确认。这一刻，LLM 相当于对主机说：“我需要用一下get_weather工具帮忙查一下上海天气的天气！”
日志中可以看到这一决策过程：
<pre class="highlighter-hljs"><code>🤖 LLM 回复：
{"tool_calls": [{"name": "get_weather", "arguments": {"city": "上海"}}]}
🛠 调用工具 get_weather 参数: {'city': '上海'}</code></pre>
 
如果 LLM 能直接回答问题（不需要工具），那么它会返回纯文本，MCP 主机则会直接将该回答返回给客户端，完成整个流程。而在本例中， 需要外部Tool获取数据。
 
<h1>Tool Call 发起与数据获取</h1>
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)。
此时日志可能会出现：
<pre class="highlighter-hljs"><code>🛠 调用工具 get_weather 参数: {'city': '上海'}</code></pre>
 
MCP 服务器收到请求后，开始处理实际的数据查询。在我们的例子中，MCP Server 内部知道 get_weather如何获取天气数据（本例中是硬编码，但通常应该是一个外部API接口）。它会向数据源（可能是一个实时天气数据库或API）请求上海当前的天气。示例代码 mcp_server_demo.py 中定义了 硬编码的get_weather 工具的实现（因此也就忽略了图1中从mcp server与后端数据源的交互，步骤6与步骤7）
接下来，MCP 服务器将拿到的数据打包成结果返回。根据 MCP 协议规范，结果通常也用 JSON 表示，这里使用MCP Python SDK解析后的字符串结果：
<pre class="highlighter-hljs"><code>result = await client.call_tool(tool_name, arguments)

# === Step 8. MCP服务器 → MCP客户端：返回结果 ===
tool_output = result.content.text
print(f"📦 工具 {tool_name} 返回：{tool_output}")</code></pre>
在控制台日志里，我们可以看到：
<pre class="highlighter-hljs"><code>工具 get_weather 返回：上海：多云，27°C</code></pre>
可以看到，MCP 服务器既完成了实际的数据获取，又把结果封装成统一格式返回给MCP Host。整个过程对于 LLM 和客户端来说是透明的：他们不需要关心天气数据具体来自哪个数据库或API，只需通过 MCP 协议与服务器交互即可。这体现了 MCP 模块化的设计理念——Tool的实现细节被封装在MCP Server中，对外提供标准接口。
 
<h1>结果返回与答案生成</h1>
现在MCP 主机从 MCP 服务器拿到了工具调用结果，接下来要做的是把结果交还给最初发起请求的 LLM，让它完成最终答案生成。
在我们的示例中，MCP 主机收到了 get_weather 的结果 JSON。MCP 主机会将该结果作为新的输入提供给 LLM。常见做法是将工具返回的结果附加到发送给LLM的对话中：
<pre class="highlighter-hljs"><code>{
"model": "qwen/qwen2.5-vl-32b-instruct:free",
"messages": [
{
 "role": "system",
 "content": "你是一个智能助手，拥有以下工具和资源可以调用：\n\n🛠 工具列表：\n- get_weather: 获取指定城市的天气信息\n- suggest_activity: 根据天气描述推荐适合的活动\n\n📚 资源列表：\n（无）\n\n请优先调用可用的Tool或Resource，而不是llm内部生成。仅根据上下文调用工具，不传入不需要的参数进行调用\"北京\"}}]}\n如无需调用工具，返回：{\"tool_calls\": null}"
},
{
 "role": "user",
 "content": "上海的天气如何？"
},
{
 "role": "tool",
 "name": "get_weather",
 "content": "上海：多云，27°C"
}
]
}</code></pre>
注意到新增的role: tool，这代表工具返回的信息作为上下文提供给LLM。
现在LLM 得到真实的天气数据后，拥有足够的数据，可以给出用户想要的答复了。对于用户问的“现在上海的天气怎么样？”，模型现在知道上海天气晴朗，27°C左右。它组织语言，将信息融入自然的回答中。例如，模型产出：“上海的天气是多云，温度为27°C。根据当前天气条件，建议您进行室内活动。”
MCP 主机接收到来自 LLM 的最终回答文本后，会将其发送回先前等待的 MCP 客户端。客户端则将答案显示给用户。至此，一次完整的问答闭环结束，用户收到满意的答复，而背后经过的一系列 Tool Call 流程对用户来说几乎无感
在用户看来，聊天对话可能长这样：
<pre class="highlighter-hljs"><code>用户：上海的
助手：上海的天气是多云，温度为27°C。根据当前天气条件，建议您进行室内活动。</code></pre>
 
<h1>小结</h1>
通过上述实例，我们能直观感受到 MCP 架构在设计上的独特优势。它明确了 LLM 应用中的职责划分，让语言理解与工具调用两个不同的职责有效解耦，实现了更高的系统灵活性：
<ul>
<li>
<div>模块化易扩展：添加新的工具服务只需实现一个独立的 MCP Server 即可，完全不需要改动 LLM 本身代码。无论是新增股票信息、日程安排或是其他功能，只需符合 MCP 协议标准，新服务即可迅速上线。</div>
</li>
<li>
<div>接口统一标准化：MCP 清晰定义了请求和响应的标准化格式，极大降低了接入新工具的成本。开发者无需再为每种工具分别设计集成逻辑，统一 JSON Schema 接口，使得调试和维护更加直观、高效。</div>
</li>
<li>
<div>实时能力增强：MCP 使 LLM 可以实时获取外部信息，突破模型训练数据的时效限制。诸如天气、新闻、股票行情甚至实时数据库查询等需求，都能轻松满足，从而大幅提升模型的实用性。</div>
</li>
<li>
<div>安全控制精细化：由于工具调用被隔离在独立的 MCP Server 中，开发者可针对具体工具执行细粒度的权限和安全管理，有效避免了 LLM 直接运行任意代码的风险。</div>
</li>
<li>
<div>故障易于追踪处理：错误消息通过标准协议明确返回，方便 LLM 做出合适的错误处理与用户反馈，有效提升用户体验及系统稳定性。</div>
</li>
</ul>
此外，MCP 未来还有许多潜在的拓展方向，例如支持多步工具链调用，使得 LLM 可以高效完成更复杂的任务；或者实现动态的工具发现与调用机制，让 LLM 能够根据实际需求自主选择工具。
 
</article> 
来源：https://www.cnblogs.com/CareySon/p/18827525/mcp_lifecycle_via_demo

頁: [1]

琼殿技术论坛's Archiver

通过一个DEMO理解MCP（模型上下文协议）的生命周期