【大模型理论篇】--MCP协议详解

目录

1.MCP--基础概念

2.MCP--框架解析

3.MCP--核心实现

3.1.资源

3.2.提示

3.3.工具

3.4.采样

4.MCP--实战演练

4.1.工具调用

---

1.MCP--基础概念

        MCP协议（Model Communication Protocol，模型通信协议）是大语言模型（LLM）与外部系统或其他模型交互时的一种标准化通信框架，旨在提升交互效率、安全性和可扩展性。作用：

1. 标准化模型交互

• 统一接口规范：解决不同模型（如GPT、Claude、Llama）输入/输出格式差异，降低集成复杂度。
• 多模态支持：兼容文本、图像、音频等数据格式（如通过Base64编码传输非结构化数据）。

---

2. 提升协作效率

• 多模型调度：支持异构模型协同工作（例如GPT-4生成文本，Stable Diffusion生成配图）。
• 工具动态调用：允许模型通过协议请求外部API（如数据库查询、数学计算工具）。

---

3. 优化性能与扩展性

• 流式传输：支持分块返回结果，减少高延迟场景下的等待时间。
• 负载均衡：根据模型算力动态分配请求，避免单点过载。

例如：AI 应用程序的 USB-C 端口。正如 USB-C 提供了一种将设备连接到各种外围设备和配件的标准化方式一样，MCP 也提供了一种将 AI 模型连接到不同数据源和工具的标准化方式。

2.MCP--框架解析

MCP 遵循客户端-服务器c/s架构，其中：

• 主机是发起连接的 LLM 应用程序（Claude for Desktop或其他 AI 工具）。
• 客户端在主机应用程序内部与服务器保持 1：1 连接，负责协议通信。
• 服务器供客户端访问，向客户提供上下文、工具和提示。同时由于 MCP Server 自己控制资源，不用把 API 密钥给 MCP Host，因此更加安全。

基本流程：

当使用 MCP 与 Claude Desktop 交互时，具体流程如下：

1.服务器发现:

Claude Desktop 在启动时连接到配置好的 MCP 服务器

2.握手协议:
当询问数据时，Claude Desktop：确定哪个 MCP 服务器可以提供帮助（在本例中为 sqlite）
通过协议协商能力，从 MCP 服务器请求数据或操作
3.交互流程:

4.安全保证:

• MCP 服务器仅公开特定的、受控的功能，
• MCP 服务器在本地运行，它们访问的资源不会暴露在互联网上
• 需要用户确认敏感操作

3.MCP--核心实现

3.1.资源

资源表示 MCP 服务器想要向客户端提供的任何类型的数据。

这可以包括：文件内容、数据库记录、API 响应、实时系统数据、截图和图片、日志文件等更多内容。每个资源由唯一的 URI 标识，并且可以包含文本或二进制数据。

{uri: string;           // Unique identifier for the resourcename: string;          // Human-readable namedescription?: string;  // Optional descriptionmimeType?: string;     // Optional MIME type
}

3.2.提示

MCP 中的提示是预定义的模板，可以：接受动态参数、上下文、链接多个交互 、指导特定工作流程、表面作为 UI 元素（如斜线命令）。

{name: string;              // Unique identifier for the promptdescription?: string;      // Human-readable descriptionarguments?: [              // Optional list of arguments{name: string;          // Argument identifierdescription?: string;  // Argument descriptionrequired?: boolean;    // Whether argument is required}]
}

3.3.工具

MCP 中的工具允许服务器公开可由客户端调用并由 LLM 用来执行操作的可执行函数。工具的关键方面包括：

1. 发现 tools/list：客户端可以通过端点列出可用的工具
2. 调用：使用端点调用工具 tools/call，服务器执行请求的操作并返回结果
3. 灵活性：工具范围从简单的计算到复杂的 API 交互

与资源一样，工具也由唯一名称标识，并可以包含说明来指导其使用。但是，与资源不同的是，工具表示可以修改状态或与外部系统交互的动态操作。

{name: string;          // Unique identifier for the tooldescription?: string;  // Human-readable descriptioninputSchema: {         // JSON Schema for the tool's parameterstype: "object",properties: { ... }  // Tool-specific parameters}
}

3.4.采样

采样是 MCP 的一项强大功能，允许服务器通过客户端请求 LLM 完成，从而实现复杂的代理行为，同时保持安全性和隐私性。这种人机交互设计确保用户可以控制 LLM 所看到和生成的内容。采样流程遵循以下步骤：

1. sampling/createMessage 服务器向客户端发送请求。
2. 客户审核请求并可以修改。
3. 来自 LLM 的客户样本。
4. 客户检查完成情况。
5. 客户端将结果返回给服务器。

{messages: [{role: "user" | "assistant",content: {type: "text" | "image",// For text:text?: string,// For images:data?: string,             // base64 encodedmimeType?: string}}],modelPreferences?: {hints?: [{name?: string                // Suggested model name/family}],costPriority?: number,         // 0-1, importance of minimizing costspeedPriority?: number,        // 0-1, importance of low latencyintelligencePriority?: number  // 0-1, importance of capabilities},systemPrompt?: string,includeContext?: "none" | "thisServer" | "allServers",temperature?: number,maxTokens: number,stopSequences?: string[],metadata?: Record<string, unknown>
}

4.MCP--实战演练

MCP官网：

For Client Developers - Model Context Protocol

4.1.工具调用

MCP工具调用流程如下：

1. 用户发送问题 ->
2. LLM 分析可用工具 ->
3. 客户端通过 MCP 服务器来执行所选工具 ->
4. 将结果发送回 LLM ->
5. LLM根据工具返回结果和用户问题进行回答。

（流程有点像检索增强生成，把检索的部分替换成了调用工具）。

通过 mcp 构建一个获取 Path of Exile 2 游戏最新版本的工具

服务端server.py:

1、首先导入对应的 FastMCP 类（使用 字符串自动生成工具定义，从而轻松创建和维护 MCP 工具，这里的效果后面会展示）以及我们需要获取实时信息的网址（这里是 PoE2 的官网）。

from typing import Any
from mcp.server.fastmcp import FastMCP
# Initialize FastMCP server
mcp = FastMCP("Path of Exile 2 hotfix")
target_url = "https://www.pathofexile.com/forum/view-forum/2212"

2、工具的核心功能函数 - 工具helper 函数（实际上这里就是想要实现的功能，下面只是一个简单的爬虫函数用于示例）。 

async def poe2_hotfix(url: str):headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/58.0.3029.110 Safari/537.3'}response = requests.get(url, headers=headers)async with httpx.AsyncClient() as client:try:response = await client.get(url, headers=headers, timeout=30.0)soup = BeautifulSoup(response.text, 'html.parser')# 查找包含帖子内容的表格table = soup.find('table')result_text = ""if table:for row in table.find_all('tr'):cells = row.find_all('td')if cells:for cell in cells:result_text += cell.get_text(strip=True) + '\n'result_text += '-' * 50 + '\n' # 分隔线else:print('未找到表格元素')return result_textexcept Exception:return None

3、添加 mcp 类中的工具函数（执行处理程序，负责实际执行每个工具的逻辑），每个工具函数对应一个特定的工具。

@mcp.tool()
async def find_poe2_hotfix() -> str:hotfix_data = await poe2_hotfix(target_url)if not hotfix_data:return "Unable to find any hotfix in office"return hotfix_data

4、最后初始化并运行服务器，到这里就可以 server 端的工作就算完成了。

if __name__ == "__main__":    # Initialize and run the server mcp.run(transport='stdio')

然后运行：

 mcp dev server.py  

客户端client.py:

需要获取Anthropic.的密钥

基本客户结构

1.首先，让我们设置我们的导入并创建基本的客户端类:

import asyncio
from typing import Optional
from contextlib import AsyncExitStackfrom mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_clientfrom anthropic import Anthropic
from dotenv import load_dotenvload_dotenv()  # load environment variables from .envclass MCPClient:def __init__(self):# Initialize session and client objectsself.session: Optional[ClientSession] = Noneself.exit_stack = AsyncExitStack()self.anthropic = Anthropic()# methods will go here

2.服务器连接管理

接下来，我们将实现连接到MCP服务器的方法:

async def connect_to_server(self, server_script_path: str):"""Connect to an MCP serverArgs:server_script_path: Path to the server script (.py or .js)"""is_python = server_script_path.endswith('.py')is_js = server_script_path.endswith('.js')if not (is_python or is_js):raise ValueError("Server script must be a .py or .js file")command = "python" if is_python else "node"server_params = StdioServerParameters(command=command,args=[server_script_path],env=None)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()# List available toolsresponse = await self.session.list_tools()tools = response.toolsprint("\nConnected to server with tools:", [tool.name for tool in tools])

3.查询处理逻辑

现在让我们添加处理查询和工具调用的核心功能:

async def process_query(self, query: str) -> str:"""Process a query using Claude and available tools"""messages = [{"role": "user","content": query}]response = await self.session.list_tools()available_tools = [{"name": tool.name,"description": tool.description,"input_schema": tool.inputSchema} for tool in response.tools]# Initial Claude API callresponse = self.anthropic.messages.create(model="claude-3-5-sonnet-20241022",max_tokens=1000,messages=messages,tools=available_tools)# Process response and handle tool callsfinal_text = []assistant_message_content = []for content in response.content:if content.type == 'text':final_text.append(content.text)assistant_message_content.append(content)elif content.type == 'tool_use':tool_name = content.nametool_args = content.input# Execute tool callresult = await self.session.call_tool(tool_name, tool_args)final_text.append(f"[Calling tool {tool_name} with args {tool_args}]")assistant_message_content.append(content)messages.append({"role": "assistant","content": assistant_message_content})messages.append({"role": "user","content": [{"type": "tool_result","tool_use_id": content.id,"content": result.content}]})# Get next response from Clauderesponse = self.anthropic.messages.create(model="claude-3-5-sonnet-20241022",max_tokens=1000,messages=messages,tools=available_tools)final_text.append(response.content[0].text)return "\n".join(final_text)

4.交互式聊天界面

现在我们将添加聊天循环和清理功能:

async def chat_loop(self):"""Run an interactive chat loop"""print("\nMCP Client Started!")print("Type your queries or 'quit' to exit.")while True:try:query = input("\nQuery: ").strip()if query.lower() == 'quit':breakresponse = await self.process_query(query)print("\n" + response)except Exception as e:print(f"\nError: {str(e)}")async def cleanup(self):"""Clean up resources"""await self.exit_stack.aclose()

5.主入口点

最后，我们将添加主要的执行逻辑:

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())

uv run client.py server.py 

• 询问你是谁的问题时，LLM 会根据工具开头 FastMCP 类的输入分析并回答。

• 可以看到当询问最新补丁时可以准确回答 PoE2 的最新补丁版本以及时间。

优化 prompt 以及添加更多工具来实现更复杂的功能，比如使用更优的爬虫工具，以及通过深度爬虫爬取对应补丁的帖子，这样在回答的最新补丁版本号的同时返回具体内容。