深入理解 MCP 协议：从原理到工程实践
**一、MCP 介绍** **1.1 什么是 MCP** MCP（Model Context Protocol，模型上下文协议）由 Anthropic 于去年 11 月正式提出。 从本质上来说，MCP 是一种技术协议，是智能体（Agent）开发过程中共同约定的一种规范。这就好比秦始皇的“书同文、车同轨”，在统一的规范下，各方的协作效率得以大幅提高，最终提升智能体 Agent 的整体开发效率。 **1.2 MCP 解决的核心痛点** 总的来说，MCP 解决的最大痛点，是 Agent 开发中调用外部工具的技术门槛过高的问题。 能够调用外部工具，是大模型进化为智能体 Agent 的关键。如果不能使用外部工具，大模型只能是一个简单的聊天机器人。由于底层技术限制，大模型本身无法直接与外部工具通信。因此，Function Calling 的思路是创建一个外部函数（Function）作为中介：一边传递大模型的请求，另一边调用外部工具，最终让大模型能够间接调用外部工具。 Function Calling 是一项非常出色的技术设计，自诞生以来一直被业内奉为圭臻。但其存在的问题是，编写外部函数的工作量过大。一个简单的外部函数往往需要上百行代码；为了让大模型“认识”这些函数，还需要额外为每个函数编写 JSON Schema 格式的功能说明；此外，还需精心设计提示词模板，才能提高 Function Calling 响应的准确率。 MCP 的目标，就是在 Agent 开发过程中，让大模型更加便捷地调用外部工具。 **1.3 MCP 的设计思路** 为了让大模型更便捷地调用外部工具，MCP 提出了“车同轨、书同文”的方案，统一 Function Calling 的运行规范： **统一名称：**MCP 将大模型运行环境称作 MCP Client（MCP 客户端），将外部函数运行环境称作 MCP Server（MCP 服务器）。 **统一规范：**统一 MCP 客户端和服务器的运行规范，并要求双方按照既定的提示词模板进行通信。 这种规范最大的好处在于，可以避免 MCP 服务器的重复开发。例如，查询天气、网页爬取、查询本地 MySQL 数据库等通用需求，只需一人开发好服务器，其他人即可直接复制到项目中使用，无需每次都单独编写。 **1.4 MCP 的生态与工具** “车同轨、书同文”的规范极大地促进了全球 AI 开发者的协作。很快，GitHub 上出现了海量的 MCP 服务器，涵盖 SQL 数据库检索、网页浏览与爬取、命令行操作、数据分析与机器学习建模等各个领域。 现在，只要本地运行的大模型支持 MCP 协议（即安装了相关库），仅需几行代码即可接入这些海量外部工具，Agent 的开发门槛瞬间降低。 这种旨在提高大模型 Agent 开发效率的技术协议，与 HTTP 等网络信息交换协议的目标一致，都是通过提高协作效率来提升开发效率。 MCP 客户端不仅支持 Claude 模型，也支持任意本地模型、在线大模型或 IDE。例如，Cursor 正式接入 MCP 后，不仅能快速编写 MCP 服务器，更能一键连接成百上千的开源 MCP 服务器，让大模型快速接入海量工具，大幅加快 Agent 开发进度。 **二、MCP 客户端 Client 开发流程** **2.1 编写基础 MCP 客户端** 以下是一个基础的 MCP 客户端实现代码： ```javascript import asyncio import os import json from typing import Optional from contextlib import AsyncExitStack from openai import OpenAI from dotenv import load_dotenv from 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("OPENAI_API_KEY") # 读取 OpenAI API Key self.base_url = os.getenv("BASE_URL") # 读取 BASE YRL self.model = os.getenv("MODEL") # 读取 model if not self.openai_api_key: raise ValueError("❌ 未找到 OpenAI API Key，请在 .env 文件中设置 OPENAI_API_KEY") self.client = OpenAI(api_key=self.openai_api_key, base_url=self.base_url) # 创建OpenAI client self.session: Optional[ClientSession] = None self.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_transport self.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.tools print("\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.name tool_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.content return content.message.content async def chat_loop(self): """运行交互式聊天循环""" print("\n🤖 MCP 客户端已启动！输入 'quit' 退出") while True: try: query = input("\n你: ").strip() if query.lower() == 'quit': break response = await self.process_query(query) # 发送用户输入到 OpenAI API print(f"\n🤖 OpenAI: {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) ") 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 sys asyncio.run(main()) ``` **2.2 客户端代码要点说明** **依赖管理：**使用 AsyncExitStack 来管理异步上下文，确保 MCP 会话和传输层在程序退出或异常时能够被正确清理。 **环境变量配置：**通过 dotenv 加载 .env 文件中的配置（如 OPENAI_API_KEY、BASE_URL、MODEL），避免硬编码敏感信息。 **动态工具发现：**在 process_query 中，每次处理请求前都会调用 session.list_tools() 动态获取服务器当前可用的工具列表，并自动转换为 OpenAI Function Calling 所需的 tools 格式。 **工具调用闭环：**当大模型返回 finish_reason == "tool_calls" 时，客户端会自动解析参数，调用 session.call_tool 执行工具，并将结果作为 tool 角色消息追加到上下文中，再次请求大模型生成最终回复。 **入口参数校验：**在 main 函数中校验命令行参数，确保传入了服务器脚本路径，并统一在文件顶部导入 sys 模块，避免运行时错误。 **三、MCP 服务器 Server 与使用** **3.1 MCP 服务器概念介绍** 根据 MCP 协议定义，Server 可以提供三种类型的标准能力。每个 Server 可同时提供这三种能力，或仅提供其中一种： **Resources（资源）**：类似于文件数据读取，可以是本地文件资源，也可以是 API 响应返回的内容。 **Tools（工具）**：第三方服务或功能函数，通过此能力控制 LLM 可调用哪些函数。 **Prompts（提示词）**：为用户预先定义好的、用于完成特定任务的模板。 **3.2 MCP 服务器通讯机制** Model Context Protocol（MCP）旨在将大型语言模型直接连接至数据源，实现无缝集成。根据 MCP 规范，当前支持两种传输方式： |**传输方式**|**通信机制**|**连接特性**|**适用场景**| | --- | --- | --- | --- | |stdio|客户端将服务器程序作为子进程启动，通过标准输入（stdin）和标准输出（stdout）进行数据交换。|本地进程通信|客户端和服务器在同一台机器上运行的场景。| |SSE|客户端和服务器通过 HTTP 协议通信，利用 Server-Sent Events 实现服务器向客户端的实时数据推送。|基于长连接的单向推送|客户端和服务器位于不同物理位置的场景。| **注意：**SSE 仅支持服务器向客户端的单向通信，而 WebSocket 则支持双向通信。HTTP 采用传统的请求-响应模式，每次请求通常建立新的连接；SSE 则基于长连接，适用于实时通知、股票行情更新等场景。 **3.3 服务器创建流程** 创建一个标准的 MCP 服务器通常包含以下步骤： **引入 mcp.server.fastmcp 模块并初始化 FastMCP 实例。** **定义具体的业务逻辑函数（如 API 请求、数据处理等）。** **使用 @mcp.tool() 装饰器将业务函数注册为 MCP 工具，并编写清晰的文档字符串（Docstring）作为工具描述。** **在 __main__ 中调用 mcp.run(transport='stdio') 启动服务器。** **3.4 服务器代码编写** 以下是一个完整的MCP 服务器实现： ```javascript import json import httpx from typing import Any from mcp.server.fastmcp import FastMCP # 初始化 MCP 服务器 mcp = FastMCP("WeatherServer") # OpenWeather API 配置 OPENWEATHER_API_BASE = "https://api.openweathermap.org/data/2.5/weather" API_KEY = "YOUR_API_KEY" # 请替换为你自己的 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} async with httpx.AsyncClient() as client: try: response = await client.get(OPENWEATHER_API_BASE, params=params, headers=headers, timeout=30.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: 格式化后的天气信息字符串 """ # 如果传入的是字符串，则先转换为字典 if isinstance(data, str): try: data = json.loads(data) except Exception as e: return f"无法解析天气数据: {e}" # 如果数据中包含错误信息，直接返回错误提示 if "error" in data: return f"⚠️ {data['error']}" # 提取数据时做容错处理 city = data.get("name", "未知") country = data.get("sys", {}).get("country", "未知") temp = data.get("main", {}).get("temp", "N/A") humidity = data.get("main", {}).get("humidity", "N/A") wind_speed = data.get("wind", {}).get("speed", "N/A") # weather 可能为空列表，因此用 [0] 前先提供默认字典 weather_list = data.get("weather", [{}]) description = weather_list[0].get("description", "未知") return ( f"🌍 {city}, {country}\n" f"🌡 温度: {temp}°C\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') ``` **3.5 服务器代码要点说明** **FastMCP 封装：**使用 FastMCP 可以极大地简化服务器的创建过程，无需手动处理底层的 JSON-RPC 通信协议。 **工具注册与描述：**@mcp.tool() 装饰器会自动将函数名和参数类型转换为 MCP 工具定义。函数的 Docstring（如“输入指定城市的英文名称…”）至关重要，它会被大模型读取并作为工具的功能说明，直接影响大模型调用工具的准确率。 **异步网络请求：**使用 httpx.AsyncClient 进行异步 HTTP 请求，避免阻塞 MCP 服务器的事件循环，确保高并发下的响应性能。 **健壮的错误处理：**在 fetch_weather 中捕获了 HTTPStatusError 和通用 Exception，并将错误信息封装为字典返回，防止因外部 API 异常导致整个 MCP 服务器崩溃。 **数据格式化：**将原始的 JSON 数据通过 format_weather 转换为人类可读的文本格式，这有助于大模型更好地理解工具返回的结果，从而生成更准确的回复。
