MCP Server 错误处理与异常协议：构建鲁棒的分布式 AI 工具通信系统

{
  "error": {
    "code": "MCP-400-001",
    "type": "ProtocolError",
    "message": "Invalid JSON schema",
    "details": {
      "field": "params",
      "expected": "object",
      "actual": "string"
    },
    "timestamp": "2026-01-01T12:00:00Z",
    "correlation_id": "abc123",
    "trace_id": "def456"
  }
}

from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import JSONResponse

app = FastAPI()

class MCPError(Exception):
    def __init__(self, code: str, message: str, details: dict = None):
        self.code = code
        self.message = message
        self.details = details or {}
        super().__init__(self.message)

@app.exception_handler(MCPError)
async def mcp_exception_handler(request: Request, exc: MCPError):
    return JSONResponse(
        status_code=400,
        content={
            "error": {
                "code": exc.code,
                "type": "ApplicationError",
                "message": exc.message,
                "details": exc.details,
                "timestamp": datetime.now().isoformat(),
                "correlation_id": request.headers.get("X-Correlation-ID", str(uuid.uuid4())),
                "trace_id": request.headers.get("X-Trace-ID", str(uuid.uuid4()))
            }
        }
    )

@app.exception_handler(Exception)
async def general_exception_handler(request: Request, exc: Exception):
    return JSONResponse(
        status_code=500,
        content={
            "error": {
                "code": "MCP-500-001",
                "type": "InternalError",
                "message": "Internal server error",
                "details": {
                    "error_type": type(exc).__name__,
                    "error_message": str(exc)
                },
                "timestamp": datetime.now().isoformat(),
                "correlation_id": request.headers.get("X-Correlation-ID", str(uuid.uuid4())),
                "trace_id": request.headers.get("X-Trace-ID", str(uuid.uuid4()))
            }
        }
    )

class ToolExecutor:
    def __init__(self, tool_registry):
        self.tool_registry = tool_registry
    
    async def execute_tool(self, tool_name: str, params: dict, request_id: str):
        # 检查工具是否存在
        if tool_name not in self.tool_registry:
            raise MCPError(
                code="MCP-404-001",
                message=f"Tool not found: {tool_name}",
                details={"tool_name": tool_name}
            )
        
        tool = self.tool_registry[tool_name]
        
        # 验证参数
        try:
            validated_params = tool.validate_params(params)
        except ValidationError as e:
            raise MCPError(
                code="MCP-400-002",
                message="Invalid tool parameters",
                details={
                    "tool_name": tool_name,
                    "errors": e.errors()
                }
            )
        
        # 执行工具
        try:
            result = await tool.execute(validated_params, request_id)
            return result
        except Exception as e:
            # 捕获工具执行过程中的异常
            raise MCPError(
                code="MCP-500-002",
                message=f"Tool execution failed: {tool_name}",
                details={
                    "tool_name": tool_name,
                    "error_type": type(e).__name__,
                    "error_message": str(e),
                    "params": validated_params
                }
            )

class CustomMCPError(MCPError):
    """自定义 MCP 错误基类"""
    def __init__(self, code: str, message: str, details: dict = None, error_type: str = None):
        super().__init__(code, message, details)
        self.error_type = error_type or "CustomError"
    
    def to_dict(self):
        base_dict = super().to_dict()
        base_dict["error"]["type"] = self.error_type
        return base_dict

# 工具开发者定义的具体自定义错误
class FileAccessError(CustomMCPError):
    def __init__(self, file_path: str, reason: str):
        super().__init__(
            code="MCP-403-101",
            message=f"File access denied: {file_path}",
            details={"file_path": file_path, "reason": reason},
            error_type="FileAccessError"
        )

class DatabaseConnectionError(CustomMCPError):
    def __init__(self, database: str, error: str):
        super().__init__(
            code="MCP-500-101",
            message=f"Database connection failed: {database}",
            details={"database": database, "error": error},
            error_type="DatabaseConnectionError"
        )

class FileTool:
    async def execute(self, params: dict, request_id: str):
        file_path = params["file_path"]
        action = params["action"]
        
        # 检查文件访问权限
        if not self.has_access(file_path, action):
            raise FileAccessError(file_path, "Insufficient permissions")
        
        # 执行文件操作
        try:
            # 文件操作逻辑
            pass
        except PermissionError:
            raise FileAccessError(file_path, "OS permission denied")

// Node.js WebSocket 错误处理实现
io.on('connection', (socket) => {
    // 监听工具调用事件
    socket.on('tool_call', async (data) => {
        const correlationId = data.correlationId || uuid.v4();
        
        try {
            // 验证请求格式
            if (!data.toolName || typeof data.toolName !== 'string') {
                throw new MCPError('MCP-400-001', 'Missing or invalid toolName');
            }
            
            // 执行工具
            const result = await toolService.executeTool(data.toolName, data.params, socket.id);
            
            // 发送成功响应
            socket.emit('tool_result', {
                correlationId,
                success: true,
                result
            });
        } catch (error) {
            // 处理错误，发送错误响应
            const errorResponse = {
                correlationId,
                success: false,
                error: {
                    code: error.code || 'MCP-500-001',
                    type: error.error_type || 'InternalError',
                    message: error.message,
                    details: error.details || {}
                }
            };
            
            socket.emit('tool_result', errorResponse);
            
            // 记录错误日志
            logger.error('WebSocket tool call failed', {
                correlationId,
                socketId: socket.id,
                error: errorResponse.error
            });
        }
    });
    
    // 处理连接错误
    socket.on('error', (error) => {
        logger.error('WebSocket connection error', {
            socketId: socket.id,
            error: error.message
        });
    });
    
    // 处理断开连接
    socket.on('disconnect', (reason) => {
        logger.info('WebSocket disconnected', {
            socketId: socket.id,
            reason
        });
    });
});

import logging
from prometheus_client import Counter, Histogram

# 设置日志配置
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('mcp_server.log'),
        logging.StreamHandler()
    ]
)

logger = logging.getLogger('mcp_server')

# 定义 Prometheus 指标
ERROR_COUNTER = Counter('mcp_server_errors_total', 'Total number of MCP Server errors', ['code', 'type'])
REQUEST_DURATION = Histogram('mcp_server_request_duration_seconds', 'Request duration in seconds', ['endpoint'])

# 错误日志记录装饰器
def log_error(func):
    async def wrapper(*args, **kwargs):
        try:
            return await func(*args, **kwargs)
        except MCPError as e:
            # 记录 MCP 错误
            logger.error(f"MCP Error: {e.code} - {e.message}", {
                "error_code": e.code,
                "error_type": "ApplicationError",
                "details": e.details
            })
            
            # 更新 Prometheus 指标
            ERROR_COUNTER.labels(code=e.code, type="ApplicationError").inc()
            
            raise
        except Exception as e:
            # 记录未捕获的异常
            logger.error(f"Unexpected Error: {str(e)}", {
                "error_type": type(e).__name__,
                "error_message": str(e)
            })
            
            # 更新 Prometheus 指标
            ERROR_COUNTER.labels(code="MCP-500-001", type="InternalError").inc()
            
            raise
    
    return wrapper

# 在路由处理函数中使用
@app.post("/api/v2/tool_call")
@log_error
async def tool_call(request: ToolCallRequest):
    # 请求处理逻辑
    pass

from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

class RetryableToolExecutor(ToolExecutor):
    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=2, max=10),
        retry=retry_if_exception_type((NetworkError, DatabaseConnectionError)),
        reraise=True
    )
    async def execute_tool_with_retry(self, tool_name: str, params: dict, request_id: str):
        # 调用父类的 execute_tool 方法
        return await super().execute_tool(tool_name, params, request_id)
    
    async def execute_tool(self, tool_name: str, params: dict, request_id: str):
        try:
            return await self.execute_tool_with_retry(tool_name, params, request_id)
        except Exception as e:
            # 重试失败后，记录详细日志
            logger.error(f"Tool execution failed after 3 retries", {
                "tool_name": tool_name,
                "request_id": request_id,
                "error": str(e)
            })
            raise

# MCP Server 错误处理配置示例
logging:
  level: INFO
  format: json
  file: ./logs/mcp_server.log
  rotation: daily
  retention: 30d

error_handling:
  # 敏感数据过滤配置
  sensitive_data_filter:
    enabled: true
    patterns:
      - password
      - api_key
      - token
    replacement: "***"
  
  # 重试配置
  retry:
    enabled: true
    max_attempts: 3
    wait_initial: 1000ms
    wait_multiplier: 2
    max_wait: 10000ms
    retryable_errors:
      - MCP-500-002
      - MCP-503-001
  
  # 监控配置
  monitoring:
    prometheus:
      enabled: true
      port: 9090
    jaeger:
      enabled: true
      endpoint: http://localhost:14268/api/traces

# MCP 错误处理测试用例示例
import pytest
from mcp_server.errors import MCPError, FileAccessError
from mcp_server.executor import ToolExecutor

class TestToolExecutor:
    def test_tool_not_found(self):
        """测试工具不存在错误"""
        executor = ToolExecutor({})
        
        with pytest.raises(MCPError) as excinfo:
            await executor.execute_tool("non_existent_tool", {}, "test_request_id")
        
        assert excinfo.value.code == "MCP-404-001"
        assert "Tool not found" in excinfo.value.message
    
    def test_invalid_params(self):
        """测试无效参数错误"""
        # 假设存在一个需要特定参数的工具
        executor = ToolExecutor({"test_tool": TestTool()})
        
        with pytest.raises(MCPError) as excinfo:
            await executor.execute_tool("test_tool", {"invalid_param": "value"}, "test_request_id")
        
        assert excinfo.value.code == "MCP-400-002"
        assert "Invalid tool parameters" in excinfo.value.message
    
    def test_file_access_error(self):
        """测试自定义文件访问错误"""
        executor = ToolExecutor({"file_tool": FileTool()})
        
        with pytest.raises(FileAccessError) as excinfo:
            await executor.execute_tool("file_tool", {"path": "/protected/file"}, "test_request_id")
        
        assert excinfo.value.code == "MCP-403-101"
        assert "File access denied" in excinfo.value.message