什么是 MCP？
MCP（Model Context Protocol，模型上下文协议）是 Anthropic 提出的开放协议，旨在标准化 AI 模型与外部工具、数据源之间的交互方式。
┌─────────────────────────────────────────────────────────────┐
│ MCP 架构全景图 │
├─────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ MCP Host │ │
│ │ (Claude Desktop / IDE / 自定义应用) │ │
│ └────────────┬──────────────────────┬─────────────────┘ │
│ │ │ │
│ │ stdio / SSE │ stdio / SSE │
│ │ │ │
│ ┌────────────▼────────┐ ┌────────▼────────────┐ │
│ │ MCP Client │ │ MCP Client │ │
│ │ (Protocol Layer) │ │ (Protocol Layer) │ │
│ └────────────┬────────┘ └────────┬────────────┘ │
│ │ │ │
│ │ JSON-RPC 2.0 │ JSON-RPC 2.0 │
│ │ │ │
│ ┌────────────▼────────┐ ┌────────▼────────────┐ │
│ │ MCP Server │ │ MCP Server │ │
│ │ (File System) │ │ (Database) │ │
│ └─────────────────────┘ └─────────────────────┘ │
│ │
│ 核心概念： │
│ • Host: 运行 LLM 的应用程序 │
│ • Client: 与 Server 建立和维护 1:1 连接 │
│ • Server: 向 Client 暴露特定能力的程序 │
│ │
└─────────────────────────────────────────────────────────────┘MCP 的核心价值：
- 标准化：统一的工具调用协议
- 可组合：不同 Server 可自由组合
- 安全：明确的权限控制模型
- 开放：任何人可开发 MCP Server
MCP vs Function Calling
| 维度 | MCP | Function Calling |
|---|---|---|
| 协议层级 | 应用层协议 | 模型能力 |
| 传输方式 | stdio / SSE | API 参数 |
| 生命周期 | 长连接 | 单次调用 |
| 状态管理 | 有状态 | 无状态 |
| 发现机制 | 能力协商 | 预定义 Schema |
| 安全性 | 细粒度权限 | 依赖实现 |
| 适用场景 | 桌面/IDE 集成 | API 服务 |
协议基础
1. 传输层
MCP 支持两种传输方式：
┌─────────────────────────────────────────────────────────────┐
│ MCP 传输层 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 1. stdio（标准输入输出） │
│ ┌─────────────────────────────────────────────┐ │
│ │ Host ──fork──> Server Process │ │
│ │ <──stdin/stdout──> │ │
│ │ │ │
│ │ 特点： │ │
│ │ • 本地进程间通信 │ │
│ │ • 简单可靠 │ │
│ │ • 适合本地工具 │ │
│ └─────────────────────────────────────────────┘ │
│ │
│ 2. SSE（Server-Sent Events） │
│ ┌─────────────────────────────────────────────┐ │
│ │ Host ──HTTP──> Remote Server │ │
│ │ <──SSE──> │ │
│ │ │ │
│ │ 特点： │ │
│ │ • 远程服务通信 │ │
│ │ • 支持网络隔离 │ │
│ │ • 适合云服务 │ │
│ └─────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘2. 消息格式
MCP 使用 JSON-RPC 2.0 作为消息格式：
// 请求消息
{
"jsonrpc": "2.0",
"id": 1,
"method": "tools/call",
"params": {
"name": "read_file",
"arguments": {
"path": "/path/to/file.txt"
}
}
}
// 响应消息
{
"jsonrpc": "2.0",
"id": 1,
"result": {
"content": [
{
"type": "text",
"text": "文件内容..."
}
],
"isError": false
}
}
// 错误消息
{
"jsonrpc": "2.0",
"id": 1,
"error": {
"code": -32602,
"message": "Invalid params",
"data": {
"details": "Path not found"
}
}
}
// 通知消息（无需响应）
{
"jsonrpc": "2.0",
"method": "notifications/progress",
"params": {
"progressToken": "abc123",
"progress": 50,
"total": 100
}
}协议流程
1. 初始化流程
┌─────────────────────────────────────────────────────────────┐
│ MCP 初始化流程 │
├─────────────────────────────────────────────────────────────┤
│ │
│ Client Server │
│ │ │ │
│ │ ────────── initialize ──────────────> │ │
│ │ │ │
│ │ { │ │
│ │ "protocolVersion": "2024-11-05", │ │
│ │ "capabilities": { │ │
│ │ "tools": {}, │ │
│ │ "resources": {}, │ │
│ │ "prompts": {} │ │
│ │ }, │ │
│ │ "clientInfo": { │ │
│ │ "name": "claude-desktop", │ │
│ │ "version": "1.0.0" │ │
│ │ } │ │
│ │ } │ │
│ │ │ │
│ │ <──────── initialized ──────────────── │ │
│ │ │ │
│ │ { │ │
│ │ "protocolVersion": "2024-11-05", │ │
│ │ "capabilities": { │ │
│ │ "tools": { │ │
│ │ "listChanged": true │ │
│ │ }, │ │
│ │ "resources": { │ │
│ │ "subscribe": true, │ │
│ │ "listChanged": true │ │
│ │ }, │ │
│ │ "prompts": { │ │
│ │ "listChanged": true │ │
│ │ } │ │
│ │ }, │ │
│ │ "serverInfo": { │ │
│ │ "name": "filesystem-server", │ │
│ │ "version": "1.0.0" │ │
│ │ } │ │
│ │ } │ │
│ │ │ │
│ │ ────────── initialized ─────────────> │ (确认) │
│ │ │ │
└─────────────────────────────────────────────────────────────┘2. 能力协商
# capability_negotiation.py
from typing import Dict, Any, Optional
from dataclasses import dataclass
@dataclass
class ClientCapabilities:
"""客户端能力"""
experimental: Optional[Dict[str, Any]] = None
sampling: Optional[Dict] = None
roots: Optional[Dict] = None
@dataclass
class ServerCapabilities:
"""服务端能力"""
experimental: Optional[Dict[str, Any]] = None
logging: Optional[Dict] = None
prompts: Optional[Dict] = None
resources: Optional[Dict] = None
tools: Optional[Dict] = None
class CapabilityNegotiator:
"""能力协商器"""
def __init__(self):
self.client_caps: Optional[ClientCapabilities] = None
self.server_caps: Optional[ServerCapabilities] = None
def negotiate(
self,
client_caps: Dict,
server_caps: Dict
) -> Dict:
"""执行能力协商"""
negotiated = {
"protocolVersion": self._select_version(
client_caps.get("protocolVersion"),
server_caps.get("protocolVersion")
),
"capabilities": {}
}
# 协商工具能力
if "tools" in client_caps.get("capabilities", {}):
if "tools" in server_caps.get("capabilities", {}):
negotiated["capabilities"]["tools"] = {
"listChanged": (
client_caps["capabilities"]["tools"].get("listChanged", False) and
server_caps["capabilities"]["tools"].get("listChanged", False)
)
}
# 协商资源能力
if "resources" in client_caps.get("capabilities", {}):
if "resources" in server_caps.get("capabilities", {}):
negotiated["capabilities"]["resources"] = {
"subscribe": (
client_caps["capabilities"]["resources"].get("subscribe", False) and
server_caps["capabilities"]["resources"].get("subscribe", False)
),
"listChanged": (
client_caps["capabilities"]["resources"].get("listChanged", False) and
server_caps["capabilities"]["resources"].get("listChanged", False)
)
}
return negotiated
def _select_version(self, client_version: str, server_version: str) -> str:
"""选择共同支持的协议版本"""
# 简单实现：选择较低的版本
versions = [client_version, server_version]
return min(versions)核心原语
1. Resources（资源）
资源表示 Server 向 Client 暴露的数据源。
# resources.py
from typing import Optional, List
from dataclasses import dataclass
@dataclass
class Resource:
"""资源定义"""
uri: str # 唯一标识符
name: str # 可读名称
description: Optional[str] # 描述
mimeType: Optional[str] # MIME 类型
size: Optional[int] # 大小（字节）
@dataclass
class TextResourceContents:
"""文本资源内容"""
uri: str
mimeType: Optional[str]
text: str
@dataclass
class BlobResourceContents:
"""二进制资源内容"""
uri: str
mimeType: Optional[str]
blob: str # base64 编码
class ResourceManager:
"""资源管理器"""
def __init__(self):
self.resources: Dict[str, Resource] = {}
self.subscribers: Dict[str, List[str]] = {}
def register(self, resource: Resource):
"""注册资源"""
self.resources[resource.uri] = resource
def list_resources(self) -> List[Resource]:
"""列出所有资源"""
return list(self.resources.values())
def read_resource(self, uri: str) -> Optional[Union[TextResourceContents, BlobResourceContents]]:
"""读取资源内容"""
resource = self.resources.get(uri)
if not resource:
return None
# 实际实现中根据 URI 读取内容
# ...
def subscribe(self, uri: str, client_id: str):
"""订阅资源更新"""
if uri not in self.subscribers:
self.subscribers[uri] = []
self.subscribers[uri].append(client_id)
def notify_update(self, uri: str):
"""通知订阅者资源更新"""
if uri in self.subscribers:
for client_id in self.subscribers[uri]:
# 发送通知
pass2. Tools（工具）
工具是 Server 向 Client 暴露的可执行功能。
# tools.py
from typing import Dict, Any, Optional, List, Callable
from dataclasses import dataclass
import json
@dataclass
class Tool:
"""工具定义"""
name: str
description: str
inputSchema: Dict[str, Any] # JSON Schema
@dataclass
class CallToolRequest:
"""工具调用请求"""
name: str
arguments: Optional[Dict[str, Any]]
@dataclass
class CallToolResult:
"""工具调用结果"""
content: List[Dict[str, Any]]
isError: bool = False
class ToolRegistry:
"""工具注册表"""
def __init__(self):
self.tools: Dict[str, Tool] = {}
self.handlers: Dict[str, Callable] = {}
def register(
self,
name: str,
description: str,
input_schema: Dict[str, Any],
handler: Callable
):
"""注册工具"""
self.tools[name] = Tool(
name=name,
description=description,
inputSchema=input_schema
)
self.handlers[name] = handler
def list_tools(self) -> List[Tool]:
"""列出所有工具"""
return list(self.tools.values())
def call(self, request: CallToolRequest) -> CallToolResult:
"""调用工具"""
handler = self.handlers.get(request.name)
if not handler:
return CallToolResult(
content=[{
"type": "text",
"text": f"Tool '{request.name}' not found"
}],
isError=True
)
try:
# 验证参数
self._validate_arguments(request.name, request.arguments)
# 执行工具
result = handler(**(request.arguments or {}))
# 格式化结果
return self._format_result(result)
except Exception as e:
return CallToolResult(
content=[{
"type": "text",
"text": f"Error: {str(e)}"
}],
isError=True
)
def _validate_arguments(self, tool_name: str, arguments: Optional[Dict]):
"""验证参数"""
tool = self.tools.get(tool_name)
if not tool:
raise ValueError(f"Tool '{tool_name}' not found")
schema = tool.inputSchema
required = schema.get("required", [])
for param in required:
if not arguments or param not in arguments:
raise ValueError(f"Missing required parameter: {param}")
def _format_result(self, result: Any) -> CallToolResult:
"""格式化结果"""
if isinstance(result, str):
return CallToolResult(
content=[{"type": "text", "text": result}]
)
elif isinstance(result, dict):
return CallToolResult(
content=[{"type": "text", "text": json.dumps(result, ensure_ascii=False)}]
)
elif isinstance(result, list):
content = []
for item in result:
if isinstance(item, dict) and item.get("type") in ["text", "image", "resource"]:
content.append(item)
else:
content.append({"type": "text", "text": str(item)})
return CallToolResult(content=content)
else:
return CallToolResult(
content=[{"type": "text", "text": str(result)}]
)3. Prompts（提示）
提示是 Server 向 Client 提供的预定义模板。
# prompts.py
from typing import Optional, List, Dict, Any
from dataclasses import dataclass
@dataclass
class Prompt:
"""提示定义"""
name: str
description: Optional[str]
arguments: Optional[List[Dict[str, Any]]] # 参数定义
@dataclass
class PromptMessage:
"""提示消息"""
role: str # "user" | "assistant"
content: Dict[str, Any]
class PromptManager:
"""提示管理器"""
def __init__(self):
self.prompts: Dict[str, Prompt] = {}
self.templates: Dict[str, Callable] = {}
def register(
self,
name: str,
description: str,
arguments: Optional[List[Dict]] = None,
template_func: Callable = None
):
"""注册提示"""
self.prompts[name] = Prompt(
name=name,
description=description,
arguments=arguments
)
self.templates[name] = template_func
def get_prompt(self, name: str, arguments: Optional[Dict] = None) -> List[PromptMessage]:
"""获取提示内容"""
template_func = self.templates.get(name)
if not template_func:
return []
return template_func(**(arguments or {}))安全模型
1. 权限控制
# security.py
from enum import Enum
from typing import Set, Dict, Optional
class PermissionLevel(Enum):
"""权限级别"""
NONE = 0
READ = 1
WRITE = 2
EXECUTE = 3
ADMIN = 4
class SecurityPolicy:
"""安全策略"""
def __init__(self):
self.resource_permissions: Dict[str, PermissionLevel] = {}
self.tool_permissions: Dict[str, PermissionLevel] = {}
self.allowed_roots: Set[str] = set()
def check_resource_access(self, uri: str, level: PermissionLevel) -> bool:
"""检查资源访问权限"""
# 检查是否在允许的根目录下
if not any(uri.startswith(root) for root in self.allowed_roots):
return False
# 检查具体权限
current_level = self.resource_permissions.get(uri, PermissionLevel.NONE)
return current_level.value >= level.value
def check_tool_execution(self, tool_name: str) -> bool:
"""检查工具执行权限"""
level = self.tool_permissions.get(tool_name, PermissionLevel.NONE)
return level.value >= PermissionLevel.EXECUTE.value
def configure_from_capabilities(self, capabilities: Dict):
"""从能力配置安全策略"""
# 根据协商后的能力设置权限
pass2. 请求验证
# request_validation.py
import jsonschema
from typing import Dict, Any
class RequestValidator:
"""请求验证器"""
# JSON-RPC 2.0 Schema
JSONRPC_SCHEMA = {
"type": "object",
"required": ["jsonrpc"],
"properties": {
"jsonrpc": {"const": "2.0"},
"id": {"type": ["string", "integer", "null"]},
"method": {"type": "string"},
"params": {"type": ["object", "array"]},
"result": {},
"error": {
"type": "object",
"required": ["code", "message"],
"properties": {
"code": {"type": "integer"},
"message": {"type": "string"},
"data": {}
}
}
}
}
@staticmethod
def validate_request(request: Dict[str, Any]) -> bool:
"""验证请求格式"""
try:
jsonschema.validate(request, RequestValidator.JSONRPC_SCHEMA)
# 请求必须有 method
if "method" not in request:
return False
# 不能同时有 result 和 error
if "result" in request and "error" in request:
return False
return True
except jsonschema.ValidationError:
return False
@staticmethod
def validate_tool_arguments(
arguments: Dict[str, Any],
schema: Dict[str, Any]
) -> tuple[bool, Optional[str]]:
"""验证工具参数"""
try:
jsonschema.validate(arguments, schema)
return True, None
except jsonschema.ValidationError as e:
return False, str(e)错误处理
# errors.py
from enum import IntEnum
class MCPErrorCode(IntEnum):
"""MCP 错误代码"""
# 标准 JSON-RPC 错误
PARSE_ERROR = -32700
INVALID_REQUEST = -32600
METHOD_NOT_FOUND = -32601
INVALID_PARAMS = -32602
INTERNAL_ERROR = -32603
# MCP 特定错误
RESOURCE_NOT_FOUND = -32001
RESOURCE_UPDATE_FAILED = -32002
TOOL_EXECUTION_FAILED = -32003
TOOL_NOT_FOUND = -32004
PROMPT_NOT_FOUND = -32005
PROMPT_RENDERING_FAILED = -32006
class MCPError(Exception):
"""MCP 错误"""
def __init__(
self,
code: MCPErrorCode,
message: str,
data: Optional[Dict] = None
):
self.code = code
self.message = message
self.data = data
super().__init__(message)
def to_jsonrpc_error(self) -> Dict:
"""转换为 JSON-RPC 错误格式"""
error = {
"code": self.code,
"message": self.message
}
if self.data:
error["data"] = self.data
return error最佳实践
1. Server 设计原则
# server_best_practices.py
class MCPServerDesign:
"""MCP Server 设计最佳实践"""
PRINCIPLES = {
"granularity": {
"title": "原子性原则",
"description": "工具应该小而专注，而非大而全",
"example": {
"good": ["read_file", "write_file", "list_directory"],
"bad": ["file_operations"]
}
},
"discoverability": {
"title": "可发现性原则",
"description": "提供清晰的描述和 Schema",
"tips": [
"工具名称应该自描述",
"描述应该说明用途和限制",
"参数应该有明确的描述"
]
},
"statelessness": {
"title": "无状态原则",
"description": "工具调用应该是幂等的",
"note": "除非有明确的会话管理需求"
},
"error_handling": {
"title": "错误处理原则",
"description": "提供详细的错误信息",
"tips": [
"区分用户错误和系统错误",
"提供可操作的错误消息",
"使用适当的错误代码"
]
}
}2. 性能优化
# performance_optimization.py
class MCPPerformanceTips:
"""MCP 性能优化建议"""
TIPS = {
"connection_pooling": {
"description": "使用连接池管理 Client-Server 连接",
"implementation": "保持长连接，避免频繁创建销毁"
},
"caching": {
"description": "缓存资源列表和工具列表",
"ttl": "根据数据更新频率设置合理的 TTL"
},
"streaming": {
"description": "大文件传输使用流式处理",
"benefit": "减少内存占用"
},
"batching": {
"description": "批量处理请求",
"note": "注意 MCP 协议本身不支持批量调用，需要在应用层实现"
}
}总结
MCP 的核心概念：
- 协议分层：传输层 → 协议层 → 应用层
- 三大原语：Resources、Tools、Prompts
- 能力协商：Client 和 Server 动态协商功能
- 安全模型：细粒度的权限控制
设计要点：
- 工具设计要原子化
- 资源 URI 要规范化
- 错误信息要详细
- 性能要考虑缓存
与 Function Calling 的关系：
- MCP 是协议层，Function Calling 是模型能力
- MCP 更适合桌面/IDE 场景
- Function Calling 更适合 API 服务
- 两者可以结合使用
相关资源：