MCP （Model Context Protocol，模型上下文协议），它是一个协议，由 Anthropic 在 2024 年 11 月发布。它旨在实现大型语言模型（LLM）与外部数据源和工具之间的无缝集成。。MCP 就像 USB-C 一样，可以让不同设备能够通过相同的接口连接在一起。在没有 MCP 时，开发者需为每个数据源单独开发定制整合方式和 API，过程耗时且难以扩展。而 MCP 使 AI 应用能通过统一协议访问文件系统、数据库等资源，简化整合过程，提升 LLM 的使用体验。

image.png

核心功能

• 统一接口：MCP 提供了一种标准化的方式，使 LLM 能够以一致的方式连接各种数据源和工具。它类似于 AI 世界的“USB-C”接口。
• 上下文管理：通过定义标准化的接口和协议，MCP 能够动态维护 LLM 的对话上下文，确保多轮对话的连贯性和一致性。
• 安全性和灵活性：MCP 内置了安全机制，允许数据和工具的提供者控制资源访问，而无需将 API 密钥等敏感信息暴露给 LLM 提供商。

架构

image.png

• MCP主机（MCP Hosts）：发起请求的LLM应用程序，如Claude Desktop、IDE或AI工具。

• MCP客户端（MCP Clients）：在主机程序内部，与MCP server保持1:1的连接，负责发送请求和接收响应。

image.png

• MCP服务器（MCP Servers）：为MCP client提供上下文、工具和prompt信息，执行具体任务。

  • Tools：使大语言模型能够通过你的 Server 执行操作。
  • Resources：将 Server 上的数据和内容开放给大语言模型。
  • Prompts：创建可复用的提示词模板和工作流程。

• 本地资源（Local Resources）：本地计算机中可供MCP server安全访问的资源，如文件、数据库。

• 远程资源（Remote Resources）：MCP server可以连接到的远程资源，如通过API访问的外部服务。

通信协议

stdio：基于标准输入输出的本地通信：通过stdio传输数据，适用于在同一台机器上运行的客户端和服务器之间的通信。

SSE：基于SSE（Server-Sent Events）的远程通信：利用SSE与HTTP结合，实现跨网络的实时数据传输，适用于需要访问远程资源或分布式部署的场景。（协议版本 2024-11-05 开始支持，即将废弃）

Streamble HTTP：通过 HTTP 进行通信，支持流式传输。（协议版本 2025-03-26 开始支持，用于替代 SSE）MCP 协议要求客户端应尽可能支持 stdio。

STDIO

stdio 传输stdio 即 standard input & output（标准输入 / 输出），是 MCP 协议推荐使用的一种传输机制，主要用于本地进程通信，在 stdio 传输中：客户端以子进程的形式启动 MCP 服务器。 通信流程如下：

image.png

• 客户端以子进程的方式启动服务器
• 客户端往服务器的 stdin 写入消息
• 服务器从自身的 stdin 读取消息
• 服务端往自身的 stdout 写入消息
• 客户端从服务器的 stdout 读取消息
• 客户端终止子进程，关闭服务器的 stdin
• 服务器关闭自身的 stdout

优势：无外部依赖，实现简单、无网络传输，通信速度快、本地通信，安全性高

局限性：单进程通信，无法并行处理多个客户端请求进程通信的资源开销大，很难在本地运行非常多的服务

SSE

MCP 协议使用 SSE（Server-Sent Events） 传输来解决远程资源访问的问题。底层是基于 HTTP 通信，通过类似 API 的方式，让 MCP 客户端直接访问远程资源，而不用通过 stdio 传输做中转。在 SSE 传输中，服务器作为一个独立进程运行，可以处理多个客户端连接。

服务器必须提供两个端点：一个 SSE 端点，供客户端建立连接并从服务器接收消息；一个常规 HTTP POST 端点，供客户端向服务器发送消息

客户端连接时，服务器必须发送一个包含客户端用于发送消息的 URL 的端点事件。所有后续客户端消息必须作为 HTTP POST 请求发送到该端点。

8a53fbf6c096931efcb7d85565f7a24b.png

• 客户端向服务器的 /sse 端点发送请求（一般是 GET 请求），建立 SSE 连接
• 服务器给客户端返回一个包含消息端点地址的事件消息
• 客户端给消息端点发送消息
• 服务器给客户端响应消息已接收状态码
• 服务器给双方建立的 SSE 连接推送事件消息
• 客户端从 SSE 连接读取服务器发送的事件消息
• 客户端关闭 SSE 连接

SSE 传输主要解决远程资源访问的问题，依靠 HTTP 协议实现底层通信。

优势

• 支持远程资源访问，让 MCP 客户端可以直接访问远程服务，解决了 stdio 传输仅适用于本地资源的局限
• 基于标准 HTTP 协议实现，兼容性好，便于与现有 Web 基础设施集成
• 服务器可作为独立进程运行，支持处理多个客户端连接

劣势:

• 扩展性挑战：SSE 不是为云原生架构设计的，在扩展平台时会遇到瓶颈
• 浏览器连接限制：每个浏览器和域名的最大打开连接数很低(6 个)，当用户打开多个标签页时会出现问题
• 复杂的双通道响应机制：MCP 中的 SSE 实现要求服务器在接收客户端消息后，既要给当前请求响应，也要给之前建立的 SSE 连接发送响应消息
• 无法支持长期的无服务器部署：无服务器架构通常自动扩缩容，不适合长时间连接，而 SSE 需要维持持久连接
• 需要大量会话管理：需要为每个 SSE 连接分配唯一标识(sessionId)来防止数据混淆，增加了实现复杂度

Streamable HTTP

Streamable HTTP 传输是 MCP 协议在 2025-03-26 版本中引入的新传输机制，用于替代之前的 SSE 传输。

• 客户端给服务器的通信端点发消息
• 服务器给客户端响应消息
• 客户端根据服务器的响应类型，继续给服务器发消息
• 服务器继续响应客户端消息

跟 SSE 传输不同的点在于，Streamable HTTP 传输中，客户端与服务器的消息交互，基本上是“一来一回”的（单通道响应）。

工作流程

11f573385244698f1b5b572801e78c2d.png

1、用户输入处理：用户提问首先被应用接收

2、意图识别与实体提取：LLM 判断用户需求属于“”，并识别出需要外部数据支持（天气数据需通过 MCP Server 获取）

3、工具选择：基于识别的意图，选择合适的工具

4、调用MCP客户端：使用MCP客户端调用相应的工具

意图识别

现在的客户端在意图层面包括以下几种：

• 基于规则的方法：使用关键词匹配或预定义规则来识别需要调用工具的情况
• 基于LLM的方法：使用LLM分析用户问题，决定是否需要调用工具
• 混合方法：结合规则和LLM的方法

基于LLM的方法层面，Anthropic 在官网为我们提供了一个 简单的解释 ：

当用户提出一个问题时：

（1）客户端（Claude Desktop / Cursor）将你的问题发送给大模型。

（2）Claude 分析可用的工具，并决定使用哪一个（或多个）。

（3）MCP Client 客户端 通过 MCP Server 执行所选的工具。

（4）工具的执行结果被送回给大模型。

（5）Claude 结合执行结果构造最终的 prompt 并生成自然语言的回应。

（6）回应最终展示给用户！

可以看出MCP Server 是由大模型主动选择并调用的。有意思的是大模型具体是如何确定该使用哪些工具呢？以及是否会使用一些不存在的工具呢（幻觉）？

调用过程可以分为两个步骤：

• 由 LLM确定使用哪些 MCP Server。
• 执行对应的 MCP Server 并对执行结果进行重新处理。

42d32a7c9d2ce5b777408315bc875ae2.png

第一步模型如何确定该使用哪些工具

... // 省略了无关的代码
 async def start(self):
     // 初始化所有的 mcp server
     for server in self.servers:
         await server.initialize()
     // 获取所有的 tools 命名为 all_tools
     all_tools = []
     for server in self.servers:
         tools = await server.list_tools()
         all_tools.extend(tools)
     // 将所有的 tools 的功能描述格式化成字符串供 LLM 使用
     // tool.format_for_llm() 我放到了这段代码最后，方便阅读。
     tools_description = "\n".join(
         [tool.format_for_llm() for tool in all_tools]
     )
     // 这里就不简化了，以供参考，实际上就是基于 prompt 和当前所有工具的信息
     // 询问 LLM（Claude） 应该使用哪些工具。
     system_message = (
         "You are a helpful assistant with access to these tools:\n\n"
         f"{tools_description}\n"
         "Choose the appropriate tool based on the user's question. "
         "If no tool is needed, reply directly.\n\n"
         "IMPORTANT: When you need to use a tool, you must ONLY respond with "
         "the exact JSON object format below, nothing else:\n"
         "{\n"
         '    "tool": "tool-name",\n'
         '    "arguments": {\n'
         '        "argument-name": "value"\n'
         "    }\n"
         "}\n\n"
         "After receiving a tool's response:\n"
         "1\. Transform the raw data into a natural, conversational response\n"
         "2\. Keep responses concise but informative\n"
         "3\. Focus on the most relevant information\n"
         "4\. Use appropriate context from the user's question\n"
         "5\. Avoid simply repeating the raw data\n\n"
         "Please use only the tools that are explicitly defined above."
     )
     messages = [{"role": "system", "content": system_message}]
     while True:
         // Final... 假设这里已经处理了用户消息输入.
         messages.append({"role": "user", "content": user_input})
         // 将 system_message 和用户消息输入一起发送给 LLM
         llm_response = self.llm_client.get_response(messages)
     ... // 后面和确定使用哪些工具无关

 class Tool:
     """Represents a tool with its properties and formatting."""
     def init(
         self, name: str, description: str, input_schema: dict[str, Any]
     ) -> None:
         self.name: str = name
         self.description: str = description
         self.input_schema: dict[str, Any] = input_schema
     // 把工具的名字 / 工具的用途（description）和工具所需要的参数（args_desc）转化为文本
     def format_for_llm(self) -> str:
         """Format tool information for LLM.
         Returns:
             A formatted string describing the tool.
         """
         args_desc = []
         if "properties" in self.input_schema:
             for param_name, param_info in self.input_schema["properties"].items():
                 arg_desc = (
                     f"- {param_name}: {param_info.get('description', 'No description')}"
                 )
                 if param_name in self.input_schema.get("required", []):
                     arg_desc += " (required)"
                 args_desc.append(arg_desc)
         return f"""
 Tool: {self.name}
 Description: {self.description}
 Arguments:
 {chr(10).join(args_desc)}

通过进一步分析 MCP 的 Python SDK 源代码可以发现：大部分情况下，当使用装饰器 @mcp.tool() 来装饰函数时，对应的 name 和 description 等其实直接源自用户定义函数的函数名以及函数的 docstring 等。

这里仅截取一小部分片段，想了解更多请参考原始代码 。

@classmethod
 def from_function(
     cls,
     fn: Callable,
     name: str | None = None,
     description: str | None = None,
     context_kwarg: str | None = None,
 ) -> "Tool":
     """Create a Tool from a function."""
     func_name = name or fn.name # 获取函数名

     if func_name == "<lambda>":
         raise ValueError("You must provide a name for lambda functions")

     func_doc = description or fn.doc or "" # 获取函数 docstring
     is_async = inspect.iscoroutinefunction(fn)
     ...   更多请参考原始代码...

总结：模型是通过 prompt engineering，即提供所有工具的结构化描述和 few-shot 的 example 来确定该使用哪些工具。

工具执行与结果反馈机制

其实工具的执行就比较简单和直接了。承接上一步，我们把 system prompt（指令与工具调用描述）和用户消息一起发送给模型，然后接收模型的回复。当模型分析用户请求后，它会决定是否需要调用工具：

• 无需工具时：模型直接生成自然语言回复。
• 需要工具时：模型输出结构化 JSON 格式的工具调用请求。

如果回复中包含结构化 JSON 格式的工具调用请求，则客户端会根据这个 json 代码执行对应的工具。

具体的实现逻辑都在 process_llm_response 中，逻辑非常简单。

如果模型执行了 tool call，则工具执行的结果 result 会和 system prompt 和用户消息一起重新发送给模型，请求模型生成最终回复。

如果 tool call 的 json 代码存在问题或者模型产生了幻觉怎么办呢？通过阅读发现，会 skip 掉无效的调用请求。

 ... // 省略无关的代码
 async def start(self):
     ... // 上面已经介绍过了，模型如何选择工具

     while True:
         // 假设这里已经处理了用户消息输入.
         messages.append({"role": "user", "content": user_input})

         // 获取 LLM 的输出
         llm_response = self.llm_client.get_response(messages)

         // 处理 LLM 的输出（如果有 tool call 则执行对应的工具）
         result = await self.process_llm_response(llm_response)

         // 如果 result 与 llm_response 不同，说明执行了 tool call （有额外信息了）
         // 则将 tool call 的结果重新发送给 LLM 进行处理。
         if result != llm_response:
             messages.append({"role": "assistant", "content": llm_response})
             messages.append({"role": "system", "content": result})

             final_response = self.llm_client.get_response(messages)
             logging.info("\nFinal response: %s", final_response)
             messages.append(
                 {"role": "assistant", "content": final_response}
             )
         // 否则代表没有执行 tool call，则直接将 LLM 的输出返回给用户。
         else:
             messages.append({"role": "assistant", "content": llm_response})    

结合这部分原理分析：

• 工具文档至关重要 - 模型通过工具描述文本来理解和选择工具，因此精心编写工具的名称、docstring 和参数说明至关重要。

• 由于 MCP 的选择是基于 prompt 的，所以任何模型其实都适配 MCP，只要你能提供对应的工具描述。但是当你使用非 Claude 模型时，MCP 使用的效果和体验难以保证（没有做专门的训练）

  Anthropic 对 Claude 做了专门的训练，所以Claude 更能理解工具的 prompt 以及输出结构化的 tool call json 代码。

MCP的优缺点

优点

1. 功能扩展：通过集成外部资源，显著扩展LLM应用的功能。例如，通过MCP，LLM可以访问本地文件系统、数据库以及各种网络服务，从而实现更丰富的功能。
2. 灵活性：支持动态访问和集成多种数据源和工具。这种灵活性使得LLM能够根据不同的任务需求，动态调用所需的资源和服务。
3. 开放性：标准化协议支持第三方开发和集成。这为开发者提供了一个开放的平台，可以基于MCP构建各种功能强大的AI应用。
4. 安全性：MCP通过标准化的数据访问接口，大大减少了直接接触敏感数据的环节，降低了数据泄露的风险。同时，MCP内置了安全机制，确保只有经过验证的请求才能访问特定资源。
5. 上下文感知：支持动态管理对话上下文，提升多轮对话的连贯性。这对于处理复杂的多轮对话任务至关重要。

缺点

1. 复杂性：需要设计和维护与外部资源的交互逻辑。这可能会增加开发和维护的复杂性，尤其是对于一些复杂的外部系统。
2. 性能开销：访问外部资源可能引入额外的延迟。这在某些对实时性要求较高的场景下可能会成为瓶颈。
3. 兼容性问题：尽管MCP定义了标准化的接口和协议，但在实际应用中，不同的外部系统可能存在一定的兼容性问题。这需要开发者在集成过程中进行详细的测试和调试。

Function Call的局限性

1. 需要大模型本身支持：大模型（如 OpenAI 的 GPT-4、Anthropic 的 Claude 等）需要在其架构中内置对 Function Call 的支持。这意味着模型必须能够理解如何调用外部函数，并处理函数调用的结果。
2. 功能范围受限于大模型函数集合：大模型支持的 Function Call 功能范围取决于其开发团队的设计。如果模型没有内置对某个特定函数的支持，那么它就无法直接调用该函数。虽然 Function Call 提供了一定的灵活性，但它仍然受限于预定义的函数集合。如果需要调用新的工具或服务，可能需要模型开发团队扩展其支持的函数列表。

对比