MCP协议

1、MCP技术体系介绍

1_MCP介绍

MCP（Model Context Protocol，模型上下文协议），旨在为大语言模型（LLM）提供统一的、 标准化的方式与外部数据源和工具之间进行通信。

从本质上来说，MCP 是一种技术协议，一种智能体 Agent 开发过程中共同约定的一种规范。

这就好比秦始皇的书同文、车同轨，在统一的规范下，大家的协作效率就能大幅提高，最终提升智能体 Agent 的开发效率。

传统AI集成的问题：这种为每个数据源构建独立连接的方式，可以被视为一个 M*N 问题。

问题：架构碎片化，难以扩展，限制了AI获取必要上下文信息的能力

MCP解决方案：提供统一且可靠的方式来访问所需数据，克服了以往集成方法的局限性。

截止目前，已上千种 MCP 工具诞生，在强悍的 MCP 生态加持下， 人人手搓 Manus 的时代即将到来。

特征	MCP	TCP/IP、HTTPS
本质	协议（Protocol）	协议（Protocol）
作用	标准化 AI 模型与上下文来源/工具之间的数据交互方式	标准化设备之间的网络通信方式
目标	让不同模型应用可以用统一方式访问资源/工具	让不同设备、系统可以互通数据
好处	消除碎片化集成、形成生态闭环	解决设备互联、实现互联网基础

2_MCP推广时间线

2024年11月底，Anthropic 推出了 MCP，刚开始不温不火。目标就是能在 Agent 的开发过程中，让大模型更加便捷地调用外部工具。

2025年2月份， Cursor 正式宣布加入 MCP 功能支持， 一举将 MCP 推到了全体开发人员面前！

2025年3月27日，OpenAI 智能体支持 MCP。 OpenAI 联合创始人兼首席执行官 Sam Altman 也特意发文大赞 MCP，可见其对 Agent 的重要性。

3_Function calling到MCP

能调用外部工具，是大模型进化为智能体 Agent 的关键，如果不能使用外部工具，大模型就只能是个简单的聊天机器人，甚至连查询天气都做不到。

由于底层技术限制啊，大模型本身是无法和外部工具直接通信的，因此 Function calling 的思路，就是创建一个外部函数（function）作为中介，一边传递大模型的请求，另一边调用外部工具达成某一目的，最终让大模型能够间接的调用外部工具。

但唯一的问题就是，编写这个外部函数的工作量太大了，一个简单的外部函数往往就得上百行代码，而且，为了让大模型“认识”这些外部函数，还要额外为每个外部函数编写一个 JSON Schema 格式的功能说明，此外，我们还需要精心设计一个提示词模版，才能提高 Function calling 响应的准确率。

而 MCP 的目标，就是能在 Agent 开发过程中，让大模型更加便捷的调用外部工具，统一 Function calling 的运行规范。

首先是先统一名称，MCP 把大模型运行环境称作 MCP Client，也就是 MCP 客户端，同时，把外部函数运行环境称作 MCP Server，也就是 MCP 服务器，

接下来，统一 MCP 客户端和服务器的运行规范，并且要求 MCP 客户端和服务器之间，也统一按照某个既定的提示词模板进行通信。

这样的好处就是：可以避免 MCP 服务器的重复开发，也就是避免外部函数重复编写。

例如，像查询天气、网页爬取、查询本地 MySQL 数据库这种通用的需求，大家有一个人开发了一个服务器就好，开发完大家都能复制到自己的项目里来使用，不用每个人每次都单独写一套。

4_支持MCP的平台

Github查看：

• MCP官方资源：https://github.com/modelcontextprotocol/servers
• MCP热门资源： https://github.com/punkpeye/awesome-mcp-servers

其它平台：

• Glama：https://glama.ai/mcp/servers
• Smithery：https://smithery.ai
• cursor：https://cursor.directory
• MCP.so：https://mcp.so/zh
• 阿里云百炼：https://bailian.console.aliyun.com/?tab=mcp#/mcp-market

5_MCP的架构

MCP 采用客户端-服务器架构，其中 MCP 主机与一个或多个 MCP 服务器建立连接。

MCP 主机通过为每个 MCP 服务器创建一个 MCP 客户端来实现这一点。

每个 MCP 客户端与其对应的 MCP 服务器保持一对一的专用连接。

MCP架构的主要参与者包括：

• MCP Host：协调和管理一个或多个 MCP 客户端的 AI 应用程序
• MCP 客户端：维护与 MCP 服务器的连接并从 MCP 服务器获取上下文以供 MCP 主机使用的组件
• MCP 服务器：为 MCP 客户端提供上下文的程序
• Resources：向 AI 应用程序提供上下文信息的数据源（例如文件内容、数据库记录、API 响应）

每个 Agent 可以有多个 MCP Client，MCP 客户端与 MCP 服务器之间是一对一关系。

6_MCP通信方式

MCP由两层组成：

• 数据层：定义基于 JSON-RPC 的客户端-服务器通信协议，包括生命周期管理，以及核心原语，如工具、资源（文件或API响应内容等）、提示和通知。
• 传输层：定义实现客户端和服务器之间数据交换的通信机制和渠道，包括传输特定的连接建立、消息框架和授权。

从概念上讲，数据层是内层，而传输层是外层。

MCP 支持的三种传输机制：

• Stdio（本地操作，默认）：使用标准输入/输出流与同一台机器上的本地进程之间直接进行通信，提供最佳性能且无网络开销。
• Server-Sent Events（SSE）：基于 HTTP 协议的流式传输机制，它允许服务器通过 HTTP 单向推送事件到客户端（即将被废弃）。
• Streamable HTTP：基于 HTTP 协议的双向流式传输，可选用服务器发送事件来实现流式传输功能（HTTP 及 流式）。

stdio方式：

• 简单、成本低：适用于客户端和服务器同机运行，无需外部服务器。
• 通信速度快：无网络依赖，适合本地快速响应应用。
• 可靠性高：本地环境运行，受网络干扰小，易调试。
• 配置复杂：需提前安装命令行工具。
• 升级不便：不支持热更新，更新需重启客户端。
• 并发能力弱：单进程一对一通信，无法并行处理多个客户端请求（不支持分布式）。
• 资源消耗高：本地运行大量服务时开销大，不适合复杂分布式场景。
• 适用于本地开发、命令行工具、调试环境，或者模型和工具服务在同一进程内运行的情况。

SSE 方式（也是HTTP）：

• 通过 HTTP GET 请求建立与服务器的连接，服务器以流式方式持续向客户端发送数据，客户端通过解析流数据来获取实时信息。
• 这种方式即将被废弃，因为当 Client 和 Server 之间存在网络的中断时，无法感知到消息的丢失，发送失败的消息直接会被丢弃。
• 适合实时推送和客户端/浏览器的单向通知，但无法满足双向复杂交互需求。
• 既没有多通道并发同时也不够稳定，很难真正的适用于企业级应用场景。
• 所以在新版本中被 Streamable HTTP 替代，两者之间其它部分差别不大。
• 适用于需要服务器主动推送数据的场景，如实时聊天、天气预报、新闻更新等。

Streamable HTTP 方式：

• 客户端通过 HTTP POST 向服务器发送请求，并可以接收流式响应（如 JSON-RPC 响应或 SSE 流）
• 当请求数据较多或需要多次交互时，服务器可以通过长连接和分批推送的方式进行数据传输。
• 配置简单：基本上配置就一个链接地址
• 面向服务的：采用服务化设计，接口和功能模块独立，方便集成和复用，符合微服务架构理念。
• 支持高并发：能够处理大量并发请求，保证在高访问量场景下仍有稳定响应。
• OAuth2 权限控制：兼容 HTTP OAuth2 认证授权机制，支持细粒度的用户访问权限管理，保证数据安全。
• 分布式云服务：支持部署在云端和分布式环境中，实现跨节点的负载均衡和高可用。
• 热更新：服务端进行升级、客户端不必重启。
• 适用于需要支持高并发、低延迟通信的分布式系统，尤其是跨服务或跨网络的应用。

MCP 传输方式优劣势对比：

特性	Stdio	SSE	Streamable HTTP
通信方向	双向（但仅限本地）	单向（服务器到客户端）	双向（适用于复杂交互）
使用场景	本地进程间通信	实时数据推送，浏览器支持	跨服务、分布式系统、大规模并发支持
支持并发连接数	低	中等	高（适合大规模并发）
适应性	局限于本地环境	支持浏览器，但单向通信	高灵活性，支持流式数据与请求批处理
实现难度	简单，适合本地调试	简单，但浏览器兼容性和长连接限制	复杂，需处理长连接和流管理
适合的业务类型	本地命令行工具，调试环境	实时推送，新闻、股票等实时更新	高并发、分布式系统，实时交互系统

注意：对于 Java 程序员来说 Spring AI 1.0.x 不支持 Streamable HTTP，目前仅有 SSE。

7_MCP工作流程

API 主要有两个

• tools/list：列出 Server 支持的所有工具。
• tools/call：Client 请求 Server 去执行某个工具， 并将结果返回。

2、MCP工具接入

目前网络上 MCP 服务 Stdio 接入方式大多数为 TypeScript 和 Python 的方式，分别对应着 uvx 和 npx 两种指令。

1_准备工作

提前下载并配置 Python 或 Node.js。

第1种：若已配置 Python 环境，可使用以下命令安装：

pip install uv

第2种：在 Windows 下可以通过 PowerShell 运行命令来安装：

powershell -ExecutionPolicy ByPass -c "irm https://astral.sh/uv/install.ps1 | iex”

npx 为 npm 自带的包管理工具，无需额外安装。

2_Cursor中使用MCP

Cursor 中国区官网：https://www.cursor.com/cn 。

在 Cursor 中添加 MCP Server 有两种配置方式：

• 全局设置：通过 Cursor Settings -> MCP -> Add new global MCP server 添加全局可用的 MCP 服务。
• 项目级别：在项目目录的 .cursor 目录中新建 mcp.json 文 件进行配置，仅对特定项目生效。

全局设置步骤如下：

1、点击右上角的齿轮按键，弹出设置栏。2、点击 Tool & Integrations 然后通过写入JSON的形式导入MCP Server。

项目级别设置如下：

1、在项目根目录创建 .cursor 文件夹（如果不存在）。2、在该文件夹中创建 mcp.json 文件。

在“支持的MCP平台”中选择一个服务添加到 mcp.json 或全局配置中（以百度地图为例）

{"mcpServers": {"baidu-map": {"command": "cmd","args": ["/c","npx","-y","@baidumap/mcp-server-baidu-map"],"env": {"BAIDU_MAP_API_KEY": "xxx"}}}
}

然后在设置中启用，如果出现绿色则配置成功

测试验证

3_Cline中使用MCP

在 VSCoed 中打开 Cline 点击 MCP Servers 中的设置

打开 MCP 配置文件 cline_mcp_settings.json 后添加如下配置：

{"mcpServers": {"baidu-map": {"command": "cmd","args": ["/c","npx","-y","@baidumap/mcp-server-baidu-map"],"env": {"BAIDU_MAP_API_KEY": "xxx"}}}
}

配置成功后即可看到如下信息：

3、手动开发MCP

创建 mcp 项目：

uv init mcp-project
cd mcp-project
# 添加依赖项
uv add mcp
# 创建环境文件
touch .env

需要在 .env 中配置模型的基本信息。

1_Stdio方式

编写 server-stdio.py 服务端代码，以加法计算为例

# serve.py
from mcp.server.fastmcp import FastMCP
# create an mcp servermcp = FastMCP("Demo")# Add an addition tool
@mcp.tool()  # 类似于http 的post请求，调用后会产生一个副作用
def add(a: int, b: int)-> int:"""add two numbers"""return a * b # 错误的逻辑方便进行验证# Add a dynamic greeting resource@mcp.resource("greeting://{name}")# 类似于http 的get请求
def get_greeting(name: str)-> str:"""Get a personalized greeting"""return f"Hello, {name}!"if __name__ == "__main__":mcp.run(transport='stdio')

编写 client-stdio.py 客户端代码，提供一个命令行交互界面：

import asyncio
import os
import json
from typing import Optional
from contextlib import AsyncExitStackfrom openai import OpenAI
from dotenv import load_dotenvfrom mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client# 加载 .env 文件，确保 API Key 受到保护
load_dotenv()class MCPClient:def __init__(self):"""初始化 MCP 客户端"""self.exit_stack = AsyncExitStack()self.openai_api_key = os.getenv("DASHSCOPE_API_KEY")  # 读取 API Keyself.base_url = os.getenv("BASE_URL")  # 读取 BASE YRLself.model = os.getenv("MODEL")  # 读取 modelif not self.openai_api_key:raise ValueError("！！！！！未找到 API Key，请在 .env 文件中设置 DASHSCOPE_API_KEY")self.client = OpenAI(api_key=self.openai_api_key, base_url=self.base_url)  # 创建OpenAI clientself.session: Optional[ClientSession] = Noneself.exit_stack = AsyncExitStack()async def connect_to_server(self, server_script_path: str):"""连接到 MCP 服务器并列出可用工具"""is_python = server_script_path.endswith('.py')is_js = server_script_path.endswith('.js')if not (is_python or is_js):raise ValueError("服务器脚本必须是 .py 或 .js 文件")command = "python" if is_python else "node"server_params = StdioServerParameters(command=command,args=[server_script_path],env=None)# 启动 MCP 服务器并建立通信stdio_transport = await self.exit_stack.enter_async_context(stdio_client(server_params))self.stdio, self.write = stdio_transportself.session = await self.exit_stack.enter_async_context(ClientSession(self.stdio, self.write))await self.session.initialize()# 列出 MCP 服务器上的工具response = await self.session.list_tools()tools = response.toolsprint("\n已连接到服务器，支持以下工具:", [tool.name for tool in tools])async def process_query(self, query: str) -> str:"""使用大模型处理查询并调用可用的 MCP 工具 (Function Calling)"""messages = [{"role": "user", "content": query}]response = await self.session.list_tools()available_tools = [{"type": "function","function": {"name": tool.name,"description": tool.description,"input_schema": tool.inputSchema}} for tool in response.tools]# print(available_tools)response = self.client.chat.completions.create(model=self.model,messages=messages,tools=available_tools)# 处理返回的内容content = response.choices[0]if content.finish_reason == "tool_calls":# 如何是需要使用工具，就解析工具tool_call = content.message.tool_calls[0]tool_name = tool_call.function.nametool_args = json.loads(tool_call.function.arguments)# 执行工具result = await self.session.call_tool(tool_name, tool_args)print(f"\n\n[Calling tool {tool_name} with args {tool_args}]\n\n")# 将模型返回的调用哪个工具数据和工具执行完成后的数据都存入messages中messages.append(content.message.model_dump())messages.append({"role": "tool","content": result.content[0].text,"tool_call_id": tool_call.id,})# 将上面的结果再返回给大模型用于生产最终的结果response = self.client.chat.completions.create(model=self.model,messages=messages,)return response.choices[0].message.contentreturn content.message.contentasync def chat_loop(self):"""运行交互式聊天循环"""print("\n MCP 客户端已启动！输入 'quit' 退出")while True:try:query = input("\n你: ").strip()if query.lower() == 'quit':breakresponse = await self.process_query(query)  # 发送用户输入到 OpenAI APIprint(f"\n 模型回复: {response}")except Exception as e:print(f"\n 发生错误: {str(e)}")async def cleanup(self):"""清理资源"""await self.exit_stack.aclose()async def main():if len(sys.argv) < 2:print("Usage: python client.py <path_to_server_script>")sys.exit(1)client = MCPClient()try:await client.connect_to_server(sys.argv[1])await client.chat_loop()finally:await client.cleanup()if __name__ == "__main__":import sysasyncio.run(main())

代码说明

代码部分	作用
MCPClient.__init__()	初始化 MCP 客户端
connect_to_mock_server()	模拟 MCP 服务器连接
chat_loop()	提供交互式聊天界面
cleanup()	释放资源
main()	启动客户端
asyncio.run(main())	运行程序

运行代码：

uv run client-stdio.py server-stdio.py

测试结果：

2_SSE方式

创建 server-sse.py，仅将传输方式修改为 sse，启动服务端：

mcp.run(transport='sse')

创建 client-sse.py，仅修改 connect_to_server 方法，使用 sse 方式建立连接：

    async def connect_to_server(self, url: str):# 启动 MCP 服务器并建立通信sse_transport = await self.exit_stack.enter_async_context(sse_client(url))self.read, self.write = sse_transportself.session = await self.exit_stack.enter_async_context(ClientSession(self.read,self.write))await self.session.initialize()# 列出 MCP 服务器上的工具response = await self.session.list_tools()tools = response.toolsprint("\n已连接到服务器，支持以下工具:", [tool.name for tool in tools])

运行代码：

uv run client_sse.py http://127.0.0.1:8000/sse

测试结果

3_Streamable HTTP方式

创建 server-http.py，将传输方式修改为 streamable-http，启动服务端：

mcp.run(transport='streamable-http')

创建 client-http.py，仅修改 connect_to_server 方法，将传输方式变为 streamable-http：

    async def connect_to_server(self, url: str):# 启动 MCP 服务器并建立通信sse_transport = await self.exit_stack.enter_async_context(streamablehttp_client(url))# 解构,这里三个参数self.read, self.write, _ = sse_transportself.session = await self.exit_stack.enter_async_context(ClientSession(self.read, self.write))await self.session.initialize()# 列出 MCP 服务器上的工具response = await self.session.list_tools()tools = response.toolsprint("\n已连接到服务器，支持以下工具:", [tool.name for tool in tools])

运行代码：

uv run client_sse.py http://127.0.0.1:8000/mcp

测试结果

4_MCP Inspector功能介绍

在实际开发 MCP 服务器的过程中，Anthropic 提供了一个非常便捷的 debug 工具：Inspector。

借助 Inspector，我们能够非常快捷的调用各类 server，并测试其功能。

首先，安装 nodejs：

curl -fsSL https://deb.nodesource.com/setup_20.x | sudo bash -
sudo apt install -y nodejs

运行 Inspector：

npx -y @modelcontextprotocol/inspector uv run server.py

然后即可在本地浏览器查看当前工具运行情况：

4、总结与补充

Streamable HTTP 请求响应顺序

HTTP 进程生命周期可以分为三个阶段：初始化、工作、连接关闭。

HTTP流式传输服务器与客户端之间的通信流程，以及外部工具信息同步格式与流程：

消息的传递格式都是 JSON-RPC 形式的，这是一种用 JSON 编写的、结构化的远程调用协议：

类型	字段	说明
请求	jsonrpc	固定为 "2.0"
	id	请求编号，用于对应请求与响应
	method	要调用的方法名（比如 "tools/call"）
	params	方法参数（可以是对象或数组）
响应	jsonrpc	也要写 "2.0"
	id	与请求的 ID 一致
	result	成功返回值（只需 result）
	error	如果出错则返回 error 对象