MCP技术体系介绍

MCP，全称时Model Context Protocol，模型上下文协议，由Claude母公司Anthropic于2014年11月正式提出。

MCP的核心作用是统一了Agent开发过程中大模型调用外部工具的技术实现流程，从而大幅提高Agent的开发效率。在MCP诞生之前，不同外部工具各有不同的调用方法。

 要连接这些外部工具，就必须一把锁配一把钥匙，开发过程相当繁琐。

而MCP的诞生则统一了这些外部工具的调用流程。使得无论是什么外部工具，都可以借助MCP技术按照统一的流程来快速接入到大模型中，从而大幅加快Agent开发效率。这叫好比现在很多设备都可以使用Type-C和电脑进行连接类似。而未来要做到这一点，MCP提供了三个方面的技术支持。第一是抽象的技术理论的支持，第二是一整套的MCP开发工具，也就是一些库，方便大家来进行MCP工具（MCP Server）的开发和智能体的构建。第三MCP还提供一整套非常丰富的开发生态。由于MCP是一套标准协议，任何MCP工具（MCP Server）都可以无缝接入到任何MCP智能体中。也就是你开发的工具我也能用，我开发的工具你也能用。目前开源社区上已经有上个个MCP工具了。如连接数据库工具、浏览器工具、网页搜索工具、文本编写提示词工具等等。而在MCP提携中，这些工具仅需要几行代码就可以嵌入到你的智能体中，从而低门槛大幅扩展当前智能体的能力边界。

MCP解决的最大痛点，就是Agent开发中调用外部工具的技术门槛过高的问题。

能调用外部工具，是大模型进化为智能体Agent的关键。如果不能使用外部工具。大模型就只能是个简单的聊天机器人。甚至连查询天气都做不到。由于底层技术限制，大模型本身是无法和外部工具直接通信的。因此Function Calling的思路，就是创建一个外部函数（function）作为中介，以变传递大模型的请求，另一边调用外部工具，最终让大模型能够间接调用外部工具。

 例如，当要查询当前天气时，让大模型调用外部工具的function calling的过程如下：

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

MCP的目标，就是在Agent开发过程中，让大模型更加便捷的调用外部工具。为此，MCP提出了两个方案。其一，统一function Calling的运行规范。

首先是统一名称，MCP把大模型运行环境称作MCP client.也就是MCP客户端，同时把MCP运行环境称作MCP Server。

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

这样好的好处在于可以避免MCP服务的重复开发，也就是避免外部函数的重复编写。例如像查询天气、网页爬取、查询本地MySQL数据库这种通用需求，大家有一个开发一个服务器就好，开发完大家都可以复制到自己的项目使用，不用每个人每次都单独写一套。

这极大的促进了全球AI开发者共同协作。目前GitHub上已经出现了海量的已经开发好的MCPServer，从SQL数据库检索到网页浏览信息爬取，从命令行操作电脑到数据分析机器学习建模等等不一而足。

现在只要你本地运行的大模型支持MCP协议，只需要安装相关的库，仅需几行代码就可以接入这些海量的外部工具。MCP协议的目的是希望通过提高协作效率来提升开发效率，例如http。

为了普及MCP协议，anthropic还提供了一整套MCP客户端、服务器端开发的SDK，并且支持python、TS和java等多种语言，借助SDK，仅需几行代码，就可以快速开发一个MCP服务器。

然后你就可以把它接入任意一个MCP客户端来构建智能体。如果愿意，还可以把MCP服务器分享到社区，给有需求的开发者使用，甚至你还可以把你的MCP服务器放到线上运行，让用户付费使用。

而MCP的客户端，不仅支持Claude模型(美国人工智能初创公司Anthropic发布的大型语言模型家族，拥有高级推理、视觉分析、代码生成、多语言处理、多模态等能力，该模型对标ChatGPT、Gemini等产品)，也支持任意本地模型或者在线大模型，或者是一些IDE。例如，现在Cursor正式接入MCP，代表着Cursor正式成为MCP客户端，在Cursor中，我们不仅快速编写MCP Server。更能借助Cursor一键连接上成百上千的开源MCP服务器。让大模型快速接入海量工具，从而大幅加快Agent开发进度。

Function Calling技术回顾

大模型Agent开发体系

MCP的出现，主要是改进了Tools模块的实现方式。

MCP Client开发流程

uv入门介绍

     MCP开发要求借助uv进行虚拟环境创建和依赖管理。uv是一个python依赖管理工具，类似于pip 和 conda，但它更快、更高效，并且可以更好的管理python虚拟环境和依赖项。它的核心目标是替代pip、venv和pip-tools。提供更好的性能和更低的管理开销。

uv的特点：

1. 一个单一的工具，以取代pip， pip-tools, pipx，诗歌，pyenv, twine, virtualenv，等等。
2. 比pip快10-100倍。
3. 提供全面的项目管理，具有通用的锁文件。
4. 运行脚本，支持内联依赖元数据。
5. 安装和管理Python版本。
6. 运行和安装作为Python包发布的工具。
7. 包括一个pip兼容的接口，用于使用熟悉的CLI提高性能。
8. 支持可扩展项目的cargo风格工作区。
9. 磁盘空间效率高，具有用于依赖性重复数据删除的全局缓存。
10. 无需Rust或Python通过curl或pip安装。
11. 支持macOS、Linux和Windows操作系统。

uv安装流程

1.从  github  下载uv绿色版安装程序，下载后解压缩。

2.配置环境变量

UV_INSTALL_DIR: D:\ENV\uv-x86_64-pc-windows-msvc
UV_TOOL_DIR: %UV_INSTALL_DIR%\tools
UV_PYTHON_INSTALL_DIR: %UV_INSTALL_DIR%\python
UV_CACHE_DIR:%UV_INSTALL_DIR%\cache

   在PATH中添加 %UV_INSTALL_DIR% %UV_TOOL_DIR%

uv基本用法

•  创建虚拟环境

uv venv myenv

• 激活虚拟环境 

source myenv/bin/activate  #Linux/macOS
myenv\script\activate  #windows

• 安装requirements.txt

uv pip install -r requirements.txt

• 运行python项目

     如果项目中包含pyproject.tom1,可以直接运行

uv run python script#等效于以下命令pip install -r requirements.txt
python script.py

 MCP依赖的python环境可能包含多个模块，uv 通过pyproject.toml 提供更高效的管理方式，并且可以避免pip的依赖冲突问题。此外，uv的包管理速度远远超过pip。这对于MCP这样频繁管理依赖的项目来说是一个很大的优势。

MCP极简客户端搭建流程

创建MCP客户端项目

#初始化项目
uv init mcp-client
cd mcp-client#创建虚拟环境
uv venv#激活虚拟环境
.venv\Scripts\activate#安装MCP SDK
uv add mcp#执行程序
uv run client.py

基本结构

1.初始化MCP客户端

2.提供一个命令行交互界面

3.模拟MCP服务器连接

4.支持用户输入查询并返回模拟回复

5.支持安全退出

MCP客户端接入DeepSeek在线模型流程

uv add mcp openai python-dotenv -i https://pypi.tuna.tsinghua.edu.cn/simple

以下是示例的一个MCP 客户端程序

import asyncio  #让代码支持异步操作
from mcp import ClientSession  #客户端会话管理
from contextlib import AsyncExitStack   #资源管理（确保客户端关闭时可以正确释放资源）
import os
from dotenv import load_dotenv
from openai import OpenAIclass MCPClient:def __init__(self):""" 初始化 MCP 客户端"""load_dotenv()self.session = Noneself.exit_stack = AsyncExitStack()  #管理MCP客户端资源,确保程序退出时可以正确释放资self.api_key=os.getenv("DEEPSEEK_API_KEY");self.base_url=os.getenv("BASE_URL");self.model=os.getenv("MODEL");if not self.api_key:raise ValueError("未找到API Key,请在.env中添加DEEPSEEK_API_KEY")if not self.base_url:raise ValueError("未找到base url,请在.env中添加BASE_URL")if not self.model:raise ValueError("未找到model,请在.env中添加MODEL")self.client=OpenAI(api_key=self.api_key,project=self.model,base_url=self.base_url);
self.session:Optional[ClientSession]=Noneasync def process_query(self,query:str) -> str:""" 调用OpenAI API处理用户查询"""messages=[{"role":"system","content":"你是一个智能助手,帮助用户回答问题"},{"role":"user","content":query}] try:response = await asyncio.get_event_loop().run_in_executor(None,lambda:self.client.chat.completions.create(model=self.model,messages=messages))return response.choices[0].message.contentexcept Exception as e:return f"调用API时出错:{str(e)}"async def connect_to_mock_server(self):""" 模拟 MCP 服务器的连接"""print("MCP 客户端已初始化,但未连接到服务器")   async def chat_loop(self):"""运行交互式聊天循环"""print("\nMCP 客户端已经启动!输入'quit'退出")while True:try:query = input("\nQuery: ").strip()if query.lower() == 'quit':breakresponse = await self.process_query(query)print(f"\n[Mock Response] 你说的是: {response}" )except Exception as e:print(f"\n 发生错误:{str(e)}")async def cleanup(self):""" 清理资源"""await self.exit_stack.aclose()async def main():client= MCPClient()try:       await client.chat_loop()finally:await client.cleanup()if __name__ == "__main__":asyncio.run(main())

运行客户端程序

uv run client.py

MCP客户端接入本地ollama、vLLM模型流程

由于ollma和vLLM均支持Open AI风格调用方法，因此上述client.py并不需要进行任何修改，我们只需要启动响应的调度框架服务，然后修改.env文件即可。

ollama模型接入的.env配置样例如下

BASE_URL=http://localhost:11434/v1/
MODEL=qwq
OPENAI_API_KEY=ollama

然后可以运行MCP Client客户端

uv run client.py

调用MCPServer

MCP Server概念介绍

       根据MCP协议定义，Server可以提供三种类型的标准能力，Resource、Tools、Prompts，每个Server可同时提供三种类型能力或其中一种。

• Resources（资源）：类似于文件数据读取，可以是文件资源或者是API响应返回的内容。
• Tools（工具）：第三方服务、功能函数，通过此可控制LLM可调用哪些函数。
• Prompts（提示词）：为用户预先定义好的完成特定任务的模板

MCPServer通讯机制

      Model Context Protocol（MCP）是一种由Anthropic开源的协议，旨在将大语言模型直接连接至数据源，实现无缝集成。根据MCP的规范，支持三种传输机制。

1. stdio：通过标准输入和标准输出进行通信
2. SSE：通过 HTTP 进行通信，支持流式传输。（协议版本 2024-11-05 开始支持，即将废弃）
3. Streamble HTTP：通过 HTTP 进行通信，支持流式传输。（协议版本 2025-03-26 开始支持，用于替代 SSE）

MCP 协议的传输机制是可插拔的，也就是说，客户端和服务器不局限于 MCP 协议标准定义的这几种传输机制，也可以通过自定义的传输机制来实现通信。

Stdio

stdio 即 standard input & output（标准输入 / 输出）。

是 MCP 协议推荐使用的一种传输机制，主要用于本地进程通信。

在 stdio 传输中：

• 客户端以子进程的形式启动 MCP 服务器。
• 服务器从其标准输入（stdin）读取 JSON-RPC 消息，并将消息发送到其标准输出（stdout）。
• 消息可能是单个 JSON-RPC 请求、通知、响应，或者包含多个请求、通知、响应的 JSON-RPC 批处理。
• 消息由换行符分隔，且不得包含嵌套的换行符。
• 服务器可以将其 UTF-8 字符串写入标准错误（stderr）以进行日志记录。客户端可以捕获、转发或忽略此日志。
• 服务器不得向 stdout 写入无效的 MCP 消息内容。
• 客户端不得向服务器的 stdin 写入无效的 MCP 消息内容。

stdio 传输机制主要依靠本地进程通信实现。

主要的优势是：

1. 无外部依赖，实现简单
2. 无网络传输，通信速度快
3. 本地通信，安全性高

也有一些局限性：

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

SSE

MCP 协议使用 SSE（Server-Sent Events） 传输来解决远程资源访问的问题。底层是基于 HTTP 通信，通过类似 API 的方式，让 MCP 客户端直接访问远程资源，而不用通过 stdio 传输做中转。

在 SSE 传输中，服务器作为一个独立进程运行，可以处理多个客户端连接。

服务器必须提供两个端点：

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

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

服务器消息作为 SSE 消息事件发送，消息内容以 JSON 格式编码在事件数据中。

SSE 通信流程

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

SSE 传输的利弊

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

SSE 传输的主要优势：

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

SSE 传输的主要劣势与问题：

• 连接不稳定：在无服务器(serverless)环境中，SSE 连接会随机、频繁断开，影响 AI 代理需要的可靠持久连接
• 扩展性挑战：SSE 不是为云原生架构设计的，在扩展平台时会遇到瓶颈
• 浏览器连接限制：每个浏览器和域名的最大打开连接数很低(6 个)，当用户打开多个标签页时会出现问题
• 代理和防火墙问题：某些代理和防火墙会因为缺少 Content-Length 头而阻止 SSE 连接，在企业环境部署时造成挑战
• 复杂的双通道响应机制：MCP 中的 SSE 实现要求服务器在接收客户端消息后，既要给当前请求响应，也要给之前建立的 SSE 连接发送响应消息
• 无法支持长期的无服务器部署：无服务器架构通常自动扩缩容，不适合长时间连接，而 SSE 需要维持持久连接
• 需要大量会话管理：需要为每个 SSE 连接分配唯一标识(sessionId)来防止数据混淆，增加了实现复杂度
• 需要额外的连接检测和超时关闭机制：需要实现心跳检测和超时机制来避免资源泄露

Streamable HTTP 传输

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

在 Streamable HTTP 传输中，服务器作为一个独立进程运行，可以处理多个客户端连接。此传输使用 HTTP POST 和 GET 请求，服务器可以选择使用服务器发送事件（SSE）来流式传输多个服务器消息。

服务器必须提供一个同时支持 POST 和 GET 方法的单个 HTTP 端点。例如：https://xyz.mcp.so/mcp。

Streamable HTTP 通信流程

1. 客户端给服务器的通信端点发消息
2. 服务器给客户端响应消息
3. 客户端根据服务器的响应类型，继续给服务器发消息
4. 服务器继续响应客户端消息

Streamable HTTP 传输的利弊

Streamable HTTP 传输机制结合了 SSE 传输的远程访问能力和无状态 HTTP 的灵活性，同时解决了 SSE 传输中的许多问题。

主要优势：

1. 兼容无服务器环境，可以在短连接模式下工作
2. 灵活的连接模式，支持简单的请求-响应和流式传输
3. 会话管理更加标准化和清晰
4. 支持断开连接恢复和消息重传
5. 保留了 SSE 的流式传输能力，同时解决了其稳定性问题
6. 向后兼容，可以支持旧版客户端和服务器

主要劣势：

1. 相比单纯的 stdio 传输实现复杂度更高
2. 仍需处理网络连接断开和恢复的逻辑
3. 会话管理需要服务器引入额外的组件（比如用 Redis 来存储 Session）

Streamable HTTP 传输的适用场景

Streamable HTTP 传输适用于：

1. 需要远程访问服务的场景，特别是云环境和无服务器架构
2. 需要支持流式输出的 AI 服务
3. 需要服务器主动推送消息给客户端的场景
4. 大规模部署需要高可靠性和可扩展性的服务
5. 需要在不稳定网络环境中保持可靠通信的场景

与 SSE 传输相比，Streamable HTTP 传输是一个更全面、更灵活的解决方案，特别适合现代云原生应用和无服务器环境。

MCP Server示例

实现一个MCPServer

实现一个MCP Server，此MCP Server 通过调用OpenWeather来提供天气服务。

添加依赖

uv add mcp httpx

以下的 server.py示例代码,用于实现一个MCP Server

import json
import httpx
from typing import Any, AnyStr
from mcp.server.fastmcp import FastMCP
import asyncio
from dotenv import load_dotenv#初始化 MCP客户端
mcp=FastMCP("WeatherServer")#OpenWeather API配置
OPENWEATHER_API_BASE="https://api.openweathermap.org/data/2.5/weather"
OPENWEATHER_API_KEY=os.getenv("OPENWEATHER_API_KEY");
USER_AGENT="weather-app/1.0"async def fetch_weather(city:str) -> dict[str, Any] | None:"""从 OpenWeather API获取天气信息。:param city:城市名称(需使用英文,如Beijing):return: 天气数据字典;若出错返回包含error信息的字典"""params={"q":city,"appid":API_KEY,"units":"metric","lang":"zh_cn"}headers ={"User-Agent":USER_AGENT,"Accept": "application/geo+json"}async with httpx.AsyncClient() as client:try:response = await client.get(OPENWEATHER_API_BASE,params=params,headers=headers,timeout=100.0)response.raise_for_status()return response.json() #返回字典类型except httpx.HTTPStatusError as e:return {"error":f"HTTP错误:{e.response.status_code}"}except Exception as e:return {"error":f"请求失败:{str(e)}"}def format_weather(data:dict[str,Any] |str ) ->str:"""将天气数据格式化为易读文本。:param data :天气数据(可以是字段或JSON字符串):return: 格式化后的天气字符串"""#如果传入的是字符串，则先转换为字典print(data)if isinstance(data,str):try:data=json.loads(data)except Exception as e:return f"无法解析天气数据:{e}"if "error" in data:return data['error']#提取数据时做容错处理city =data.get("name","未知")country = data.get("sys",{}).get("country","未知")temp = data.get("main",{}).get("temp","未知")humidity = data.get("main",{}).get("humidity","N/A")wind_speed = data.get("wind",{}).get("speed","N/A")weather_list =data.get("weather",[{}])description =weather_list[0].get("description","未知")return {f"城市：{city},{country}\n"f"温度:{temp}℃\n"f"湿度：{humidity}%\n"f"风速:{wind_speed} m/s\n"f"天气:{description}\n"}@mcp.tool()
async def query_weather(city:str)->str:"""输入指定城市的英文名称,返回今日天气的查询结果。:param city:城市名称(需使用英文):return:格式化后的天气信息"""data = await fetch_weather(city)return format_weather(data)if  __name__=="__main__":#以标准I/O方式运行MCP服务器mcp.run(transport='stdio')

更新Client 程序

连接MCP Server

    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工具"""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]response = self.client.chat.completions.create(model= self.model,messages=messages,tools=available_tools)#处理返回的内容content=response.choices[0]#print(content)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.content

启动程序时，同时指定MCP工具

async def main():if len(sys.argv) < 2 :print("Usage : python client.py <path_to_server_Script>")sys.exit()client= MCPClient()try:await client.connect_to_server(sys.argv[1])await client.chat_loop()finally:await client.cleanup()

测试运行

#进入到项目目录
cd /root/mcp/mcp-client#激活虚拟环境
source .venv/bin/activate   #linux
.venc/scripts/activate   #windows#运行程序
uv run client.py server.py

MCP Inspector

anthropic提供一个一个非常便捷的debug工具：Inspector。借助Inspector，我们能够非常快捷的调用各类server，并测试其功能。Inspector具体功能实现流程如下：

1.安装nodejs

安装完毕后，运行npx -v 出现版本号，即证明安装成功

2.运行Inspector

npx @modelcontextprotocol/inspector uv run server.py

 然后即在本地浏览器查看当前工具运行情况：http://127.0.0.1:6274

在左侧配置MCP Server 连接信息，中间可以测试MCP Server 提供的各种服务。右侧可以对MCP提供的服务进行测试。

MCP进阶使用

MCP Server 进阶

    MCO标准通信协议带来的最大价值之一，就是让广大Agent开发者能够基于此进行协作。在MCP推出后的若干时间，已经诞生了数以千计的MCP服务器，允许用户直接下载并进行调用。下面是一些MCP服务器集合地址：

•  MCP官方服务器合集：https://github.com/modelcontextprotocol/server
• MCP Github 热门导航：https://github.com/punkpeye/awesome-mcp-servers
• MCP集合：https://github.com/ahujasid/blender-mcp
• MCP导航：https://mcp.so/

MCP Client进阶使用

    除了能在命令行中创建MCP客户端外，还支持各类客户端的调用：https://modelcontextprotocol.io/clients  列出了支持MCP Client的一些AI的工具。