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

时间:2025-09-09 09:00:03来源:互联网

下面小编就为大家分享一篇通过一个DEMO理解MCP(模型上下文协议)的生命周期,具有很好的参考价值,希望对大家有所帮助。

在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 能够根据实际需求自主选择工具。

 

本站部分内容转载自互联网,如果有网站内容侵犯了您的权益,可直接联系我们删除,感谢支持!