MCP 协议详解
MCP（Model Context Protocol） 是 Anthropic 于 2024 年 11 月推出的标准化协议，旨在解决 AI 模型与外部工具、数据源之间的连接问题。MCP 被称为"AI 领域的 USB 协议"，有望成为工具调用的行业标准。
一、核心原理
1.1 MCP 是什么？
┌─────────────────────────────────────────────────────────────┐
│ MCP 核心概念 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 问题：AI 连接碎片化 │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ │ │
│ │ ┌─────────┐ ┌─────────┐ ┌─────────┐ │ │
│ │ │ Claude │ │ GPT-4 │ │ Gemini │ │ │
│ │ └────┬────┘ └────┬────┘ └────┬────┘ │ │
│ │ │ │ │ │ │
│ │ 不同接口 不同接口 不同接口 │ │
│ │ │ │ │ │ │
│ │ ↓ ↓ ↓ │ │
│ │ ┌─────────┐ ┌─────────┐ ┌─────────┐ │ │
│ │ │文件系统 │ │数据库 │ │API服务 │ │ │
│ │ └─────────┘ └─────────┘ └─────────┘ │ │
│ │ │ │
│ │ 每个模型 × 每个数据源 = N×M 种连接方式 │ │
│ │ │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ 解决方案：MCP 标准化协议 │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ │ │
│ │ ┌─────────┐ ┌─────────┐ ┌─────────┐ │ │
│ │ │ Claude │ │ GPT-4 │ │ Gemini │ │ │
│ │ └────┬────┘ └────┬────┘ └────┬────┘ │ │
│ │ │ │ │ │ │
│ │ └───────────────┼───────────────┘ │ │
│ │ │ │ │
│ │ ↓ │ │
│ │ ┌─────────────┐ │ │
│ │ │ MCP │ │ │
│ │ │ 标准协议 │ │ │
│ │ └──────┬──────┘ │ │
│ │ │ │ │
│ │ ┌───────────────┼───────────────┐ │ │
│ │ ↓ ↓ ↓ │ │
│ │ ┌─────────┐ ┌─────────┐ ┌─────────┐ │ │
│ │ │文件系统 │ │数据库 │ │API服务 │ │ │
│ │ │ Server │ │ Server │ │ Server │ │ │
│ │ └─────────┘ └─────────┘ └─────────┘ │ │
│ │ │ │
│ │ N + M 种连接方式（线性复杂度） │ │
│ │ │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘1.2 MCP 架构
┌─────────────────────────────────────────────────────────────┐
│ MCP 架构 │
├─────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ MCP Host │ │
│ │ (Claude Desktop / IDE) │ │
│ │ │ │
│ │ ┌──────────────────────────────────────────────┐ │ │
│ │ │ MCP Client │ │ │
│ │ │ │ │ │
│ │ │ • 管理连接 │ │ │
│ │ │ • 协议转换 │ │ │
│ │ │ • 能力协商 │ │ │
│ │ └───────────────────┬──────────────────────────┘ │ │
│ └───────────────────────┼─────────────────────────────┘ │
│ │ │
│ │ JSON-RPC 2.0 │
│ │ │
│ ┌───────────────────────┼─────────────────────────────┐ │
│ │ ↓ │ │
│ │ ┌─────────┐ ┌─────────┐ ┌─────────┐ │ │
│ │ │ MCP │ │ MCP │ │ MCP │ │ │
│ │ │ Server1 │ │ Server2 │ │ Server3 │ │ │
│ │ │ │ │ │ │ │ │ │
│ │ │ 文件系统 │ │ 数据库 │ │ 搜索API │ │ │
│ │ └─────────┘ └─────────┘ └─────────┘ │ │
│ │ │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ 核心概念： │
│ • Host: 运行 MCP Client 的应用程序（如 Claude Desktop） │
│ • Client: 与 MCP Server 通信的客户端 │
│ • Server: 提供工具、资源、提示词的服务端 │
│ • Transport: 通信传输层（stdio / HTTP / WebSocket） │
│ │
└─────────────────────────────────────────────────────────────┘1.3 MCP 核心能力
| 能力 | 说明 | 示例 |
|---|---|---|
| Resources | 暴露数据源 | 文件内容、数据库记录 |
| Tools | 暴露可执行功能 | 搜索、计算、API 调用 |
| Prompts | 暴露预定义提示词 | 代码审查模板、文档生成 |
| Sampling | 服务端请求 LLM 生成 | 智能摘要、内容分析 |
二、协议规范
2.1 通信协议
MCP 使用 JSON-RPC 2.0 作为通信协议：
// 请求格式
{
"jsonrpc": "2.0",
"id": 1,
"method": "tools/call",
"params": {
"name": "search",
"arguments": {
"query": "Python教程"
}
}
}
// 响应格式
{
"jsonrpc": "2.0",
"id": 1,
"result": {
"content": [
{
"type": "text",
"text": "搜索结果..."
}
]
}
}
// 错误格式
{
"jsonrpc": "2.0",
"id": 1,
"error": {
"code": -32600,
"message": "Invalid Request"
}
}2.2 生命周期
┌─────────────────────────────────────────────────────────────┐
│ MCP 连接生命周期 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 1. 初始化阶段 (Initialization) │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ │ │
│ │ Client Server │ │
│ │ │ │ │ │
│ │ │ ── initialize ─────────────→ │ │ │
│ │ │ {protocolVersion, │ │ │
│ │ │ capabilities} │ │ │
│ │ │ │ │ │
│ │ │ ←── InitializeResult ─────── │ │ │
│ │ │ {protocolVersion, │ │ │
│ │ │ capabilities, │ │ │
│ │ │ serverInfo} │ │ │
│ │ │ │ │ │
│ │ │ ── initialized ────────────→ │ │ │
│ │ │ │ │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ 2. 能力发现阶段 (Capability Discovery) │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ │ │
│ │ Client Server │ │
│ │ │ │ │ │
│ │ │ ── tools/list ─────────────→ │ │ │
│ │ │ │ │ │
│ │ │ ←── ListToolsResult ──────── │ │ │
│ │ │ {tools: [...]} │ │ │
│ │ │ │ │ │
│ │ │ ── resources/list ─────────→ │ │ │
│ │ │ │ │ │
│ │ │ ←── ListResourcesResult ──── │ │ │
│ │ │ │ │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ 3. 操作阶段 (Operation) │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ │ │
│ │ Client Server │ │
│ │ │ │ │ │
│ │ │ ── tools/call ─────────────→ │ │ │
│ │ │ {name, arguments} │ │ │
│ │ │ │ │ │
│ │ │ ←── CallToolResult ──────── │ │ │
│ │ │ {content, isError} │ │ │
│ │ │ │ │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘2.3 核心方法
| 方法 | 说明 | 方向 |
|---|---|---|
initialize | 初始化连接 | Client → Server |
initialized | 确认初始化完成 | Client → Server |
tools/list | 列出可用工具 | Client → Server |
tools/call | 调用工具 | Client → Server |
resources/list | 列出可用资源 | Client → Server |
resources/read | 读取资源内容 | Client → Server |
prompts/list | 列出可用提示词 | Client → Server |
prompts/get | 获取提示词内容 | Client → Server |
三、工具定义规范
3.1 工具 Schema
"""
MCP 工具定义格式
"""
# 工具定义示例
tool_definition = {
"name": "search_web",
"description": "搜索互联网获取实时信息",
"inputSchema": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "搜索关键词"
},
"num_results": {
"type": "integer",
"description": "返回结果数量",
"default": 5,
"minimum": 1,
"maximum": 20
}
},
"required": ["query"]
},
# MCP 特有的注解
"annotations": {
"title": "Web Search", # 人类可读标题
"readOnlyHint": True, # 只读操作
"destructiveHint": False, # 非破坏性操作
"idempotentHint": True, # 幂等操作
"openWorldHint": True # 访问外部世界
}
}
# 工具调用响应
tool_result = {
"content": [
{
"type": "text",
"text": "搜索结果：..."
},
{
"type": "image", # 支持图片
"data": "base64...",
"mimeType": "image/png"
},
{
"type": "resource", # 支持资源引用
"resource": {
"uri": "file:///path/to/file",
"mimeType": "text/plain"
}
}
],
"isError": False
}3.2 资源定义
"""
MCP 资源定义
"""
# 资源列表响应
resources_list = {
"resources": [
{
"uri": "file:///project/src/main.py",
"name": "main.py",
"description": "主程序入口文件",
"mimeType": "text/x-python"
},
{
"uri": "file:///project/config.json",
"name": "config.json",
"description": "配置文件",
"mimeType": "application/json"
}
]
}
# 资源读取响应
resource_read = {
"contents": [
{
"uri": "file:///project/src/main.py",
"mimeType": "text/x-python",
"text": "def main():\n print('Hello, World!')"
}
]
}四、MCP Server 实现
4.1 Python SDK 实现
"""
MCP Server 实现
使用 Python SDK
"""
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent
import json
# 创建 Server 实例
server = Server("example-server")
# 定义工具
@server.list_tools()
async def list_tools() -> list[Tool]:
"""返回可用工具列表"""
return [
Tool(
name="search",
description="搜索互联网获取信息",
inputSchema={
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "搜索关键词"
}
},
"required": ["query"]
}
),
Tool(
name="calculate",
description="执行数学计算",
inputSchema={
"type": "object",
"properties": {
"expression": {
"type": "string",
"description": "数学表达式"
}
},
"required": ["expression"]
}
)
]
@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
"""处理工具调用"""
if name == "search":
query = arguments.get("query", "")
# 实际调用搜索 API
result = f"搜索 '{query}' 的结果：..."
return [TextContent(type="text", text=result)]
elif name == "calculate":
expression = arguments.get("expression", "")
try:
result = eval(expression)
return [TextContent(type="text", text=f"计算结果：{result}")]
except Exception as e:
return [TextContent(type="text", text=f"计算错误：{e}")]
else:
return [TextContent(type="text", text=f"未知工具：{name}")]
# 启动服务器
async def main():
async with stdio_server() as (read_stream, write_stream):
await server.run(
read_stream,
write_stream,
server.create_initialization_options()
)
if __name__ == "__main__":
import asyncio
asyncio.run(main())4.2 完整 MCP Server 示例
"""
完整的 MCP Server 实现
包含工具、资源、提示词
"""
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import (
Tool,
Resource,
Prompt,
TextContent,
ResourceContent,
PromptMessage
)
from dataclasses import dataclass
from typing import List, Dict, Any
import os
import json
@dataclass
class FileInfo:
"""文件信息"""
path: str
content: str
mime_type: str
class FileSystemServer:
"""文件系统 MCP Server"""
def __init__(self, root_path: str):
self.root_path = root_path
self.server = Server("filesystem-server")
self._setup_handlers()
def _setup_handlers(self):
"""设置处理器"""
@self.server.list_tools()
async def list_tools() -> List[Tool]:
return [
Tool(
name="read_file",
description="读取文件内容",
inputSchema={
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "相对文件路径"
}
},
"required": ["path"]
}
),
Tool(
name="write_file",
description="写入文件内容",
inputSchema={
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "相对文件路径"
},
"content": {
"type": "string",
"description": "文件内容"
}
},
"required": ["path", "content"]
}
),
Tool(
name="list_directory",
description="列出目录内容",
inputSchema={
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "相对目录路径，默认为根目录"
}
}
}
)
]
@self.server.call_tool()
async def call_tool(name: str, arguments: Dict[str, Any]) -> List[TextContent]:
try:
if name == "read_file":
path = arguments.get("path", "")
full_path = os.path.join(self.root_path, path)
with open(full_path, "r", encoding="utf-8") as f:
content = f.read()
return [TextContent(type="text", text=content)]
elif name == "write_file":
path = arguments.get("path", "")
content = arguments.get("content", "")
full_path = os.path.join(self.root_path, path)
os.makedirs(os.path.dirname(full_path), exist_ok=True)
with open(full_path, "w", encoding="utf-8") as f:
f.write(content)
return [TextContent(type="text", text=f"文件已写入: {path}")]
elif name == "list_directory":
path = arguments.get("path", "")
full_path = os.path.join(self.root_path, path)
entries = os.listdir(full_path)
result = "\n".join(entries)
return [TextContent(type="text", text=result)]
else:
return [TextContent(type="text", text=f"未知工具: {name}")]
except Exception as e:
return [TextContent(type="text", text=f"错误: {str(e)}")]
@self.server.list_resources()
async def list_resources() -> List[Resource]:
"""列出可用资源"""
resources = []
for root, dirs, files in os.walk(self.root_path):
for file in files:
full_path = os.path.join(root, file)
rel_path = os.path.relpath(full_path, self.root_path)
resources.append(Resource(
uri=f"file:///{rel_path}",
name=file,
mimeType=self._get_mime_type(file)
))
return resources
@self.server.read_resource()
async def read_resource(uri: str) -> List[ResourceContent]:
"""读取资源内容"""
if uri.startswith("file:///"):
rel_path = uri[8:]
full_path = os.path.join(self.root_path, rel_path)
with open(full_path, "r", encoding="utf-8") as f:
content = f.read()
return [ResourceContent(
uri=uri,
mimeType=self._get_mime_type(rel_path),
text=content
)]
return []
@self.server.list_prompts()
async def list_prompts() -> List[Prompt]:
"""列出可用提示词"""
return [
Prompt(
name="code_review",
description="代码审查模板",
arguments=[
{
"name": "file_path",
"description": "要审查的文件路径",
"required": True
}
]
)
]
@self.server.get_prompt()
async def get_prompt(name: str, arguments: Dict[str, str]) -> List[PromptMessage]:
"""获取提示词内容"""
if name == "code_review":
file_path = arguments.get("file_path", "")
return [
PromptMessage(
role="user",
content=TextContent(
type="text",
text=f"""请审查以下代码文件：{file_path}
审查要点：
1. 代码风格和可读性
2. 潜在的 bug 和错误
3. 性能优化建议
4. 安全性问题
5. 最佳实践建议
请给出详细的审查意见。"""
)
)
]
return []
def _get_mime_type(self, filename: str) -> str:
"""获取 MIME 类型"""
ext = os.path.splitext(filename)[1].lower()
mime_map = {
".py": "text/x-python",
".js": "text/javascript",
".json": "application/json",
".md": "text/markdown",
".txt": "text/plain",
".html": "text/html",
".css": "text/css"
}
return mime_map.get(ext, "text/plain")
async def run(self):
"""运行服务器"""
async with stdio_server() as (read_stream, write_stream):
await self.server.run(
read_stream,
write_stream,
self.server.create_initialization_options()
)
# 启动服务器
if __name__ == "__main__":
import asyncio
server = FileSystemServer(root_path=".")
asyncio.run(server.run())五、MCP Client 集成
5.1 Claude Desktop 集成
// Claude Desktop 配置文件
// 位置: ~/Library/Application Support/Claude/claude_desktop_config.json (macOS)
// 或: %APPDATA%/Claude/claude_desktop_config.json (Windows)
{
"mcpServers": {
"filesystem": {
"command": "python",
"args": ["/path/to/filesystem_server.py"],
"env": {
"ROOT_PATH": "/Users/user/projects"
}
},
"database": {
"command": "node",
"args": ["/path/to/database_server.js"],
"env": {
"DATABASE_URL": "postgresql://localhost/mydb"
}
},
"web-search": {
"command": "python",
"args": ["/path/to/search_server.py"]
}
}
}5.2 Python Client 实现
"""
MCP Client 实现
"""
import asyncio
import json
from typing import Dict, Any, List, Optional
from dataclasses import dataclass
@dataclass
class MCPTool:
"""MCP 工具"""
name: str
description: str
input_schema: Dict[str, Any]
class MCPClient:
"""
MCP 客户端
与 MCP Server 通信
"""
def __init__(self):
self.process = None
self.tools: List[MCPTool] = []
self.request_id = 0
async def connect(self, command: str, args: List[str], env: Dict[str, str] = None):
"""连接到 MCP Server"""
self.process = await asyncio.create_subprocess_exec(
command,
*args,
stdin=asyncio.subprocess.PIPE,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
env=env
)
# 初始化连接
await self._initialize()
# 获取工具列表
await self._load_tools()
async def _initialize(self):
"""初始化连接"""
result = await self._send_request("initialize", {
"protocolVersion": "2024-11-05",
"capabilities": {},
"clientInfo": {
"name": "mcp-client",
"version": "1.0.0"
}
})
# 发送 initialized 通知
await self._send_notification("initialized", {})
return result
async def _load_tools(self):
"""加载工具列表"""
result = await self._send_request("tools/list", {})
self.tools = [
MCPTool(
name=tool["name"],
description=tool["description"],
input_schema=tool["inputSchema"]
)
for tool in result.get("tools", [])
]
async def call_tool(self, name: str, arguments: Dict[str, Any]) -> str:
"""调用工具"""
result = await self._send_request("tools/call", {
"name": name,
"arguments": arguments
})
# 提取文本内容
content = result.get("content", [])
text_parts = [
item["text"] for item in content
if item.get("type") == "text"
]
return "\n".join(text_parts)
async def _send_request(self, method: str, params: Dict) -> Dict:
"""发送请求"""
self.request_id += 1
request = {
"jsonrpc": "2.0",
"id": self.request_id,
"method": method,
"params": params
}
# 发送请求
request_str = json.dumps(request) + "\n"
self.process.stdin.write(request_str.encode())
await self.process.stdin.drain()
# 读取响应
response_line = await self.process.stdout.readline()
response = json.loads(response_line.decode())
if "error" in response:
raise Exception(response["error"]["message"])
return response.get("result", {})
async def _send_notification(self, method: str, params: Dict):
"""发送通知（无响应）"""
notification = {
"jsonrpc": "2.0",
"method": method,
"params": params
}
notification_str = json.dumps(notification) + "\n"
self.process.stdin.write(notification_str.encode())
await self.process.stdin.drain()
async def close(self):
"""关闭连接"""
if self.process:
self.process.terminate()
await self.process.wait()
# 使用示例
async def main():
client = MCPClient()
# 连接到 Server
await client.connect(
"python",
["/path/to/server.py"],
{"ROOT_PATH": "/Users/user/projects"}
)
# 查看可用工具
print("可用工具:")
for tool in client.tools:
print(f" - {tool.name}: {tool.description}")
# 调用工具
result = await client.call_tool("read_file", {"path": "README.md"})
print(f"\n文件内容:\n{result}")
await client.close()
if __name__ == "__main__":
asyncio.run(main())六、MCP 生态
6.1 官方 Server
| Server | 功能 | 状态 |
|---|---|---|
| Filesystem | 文件系统操作 | 官方支持 |
| PostgreSQL | 数据库操作 | 官方支持 |
| Git | Git 操作 | 官方支持 |
| Google Drive | 云盘操作 | 官方支持 |
| Slack | 消息发送 | 官方支持 |
| Brave Search | 网页搜索 | 官方支持 |
6.2 社区 Server
┌─────────────────────────────────────────────────────────────┐
│ MCP 生态发展 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 官方 Server（高质量、官方维护） │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ Filesystem | PostgreSQL | Git | Google Drive │ │
│ │ Slack | Brave Search | Memory │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ 社区 Server（快速扩展） │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ GitHub | Notion | Linear | Jira | Asana │ │
│ │ AWS | GCP | Azure | Kubernetes │ │
│ │ Elasticsearch | Redis | MongoDB │ │
│ │ Twitter/X | Discord | Telegram │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ 框架集成 │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ • LangChain MCP Adapter │ │
│ │ • LlamaIndex MCP Support │ │
│ │ • AutoGen MCP Integration │ │
│ │ • OpenAI (规划中) │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘七、面试高频问题
Q1: MCP 解决了什么问题？
答案要点：
- 连接碎片化：统一 AI 模型与数据源的连接方式
- N×M 问题：从 N×M 种连接简化为 N+M
- 标准化：提供通用的工具、资源、提示词规范
- 生态互联：一次开发，到处使用
Q2: MCP 与 Function Calling 的关系？
答案要点：
| 方面 | MCP | Function Calling |
|---|---|---|
| 层次 | 协议层 | 接口层 |
| 范围 | 跨平台标准 | 单平台特性 |
| 内容 | 工具+资源+提示词 | 仅工具 |
| 位置 | Server 端 | 模型 API |
MCP 是 Function Calling 的协议化升级，解决了跨平台兼容问题。
Q3: MCP 的核心组件有哪些？
答案要点：
- Host：运行 Client 的应用（如 Claude Desktop）
- Client：与 Server 通信的客户端
- Server：提供工具/资源/提示词的服务端
- Transport：通信传输层（stdio/HTTP/WebSocket）
Q4: 如何开发一个 MCP Server？
答案要点：
- 选择 SDK（Python/TypeScript）
- 定义工具（tools/list, tools/call）
- 可选：定义资源和提示词
- 配置传输方式（stdio 最简单）
- 在 Client 中注册
Q5: MCP 的局限性是什么？
答案要点：
| 局限 | 说明 |
|---|---|
| 生态成熟度 | 相对较新，生态正在发展 |
| 性能 | stdio 通信有开销 |
| 调试 | 分布式调试较复杂 |
| 标准化进程 | 仍在演进中 |
八、小结
| 概念 | 一句话总结 |
|---|---|
| MCP | AI 领域的 USB 协议，标准化工具调用 |
| 架构 | Host → Client → Server 三层架构 |
| 核心能力 | 工具（Tools）、资源（Resources）、提示词（Prompts） |
| 协议 | JSON-RPC 2.0 |
| 价值 | 解决 N×M 连接碎片化问题 |
一句话总结：MCP 是 AI 工具调用的标准化协议，有望成为行业统一标准，值得持续关注和学习。
最后更新：2026年3月18日