MCP 的版本管理和兼容性如何处理？

MCP 的版本管理和兼容性对于确保系统的稳定性和可维护性至关重要。以下是详细的版本管理策略和兼容性处理方法：

版本管理策略

MCP 版本管理应考虑以下方面：

1. 语义化版本控制：使用语义化版本号（SemVer）
2. 向后兼容性：确保新版本向后兼容旧版本
3. 废弃策略：明确的功能废弃和移除流程
4. 迁移指南：提供详细的版本迁移指南
5. 版本协商：客户端和服务器的版本协商机制

1. 语义化版本控制

python
from dataclasses import dataclass
from typing import Optional
import re

@dataclass
class Version:
    """版本号"""
    major: int
    minor: int
    patch: int
    prerelease: Optional[str] = None
    build_metadata: Optional[str] = None
    
    def __str__(self) -> str:
        version_str = f"{self.major}.{self.minor}.{self.patch}"
        
        if self.prerelease:
            version_str += f"-{self.prerelease}"
        
        if self.build_metadata:
            version_str += f"+{self.build_metadata}"
        
        return version_str
    
    @classmethod
    def parse(cls, version_str: str) -> 'Version':
        """解析版本字符串"""
        # 匹配语义化版本格式
        pattern = r'^(\d+)\.(\d+)\.(\d+)(?:-([0-9A-Za-z-]+(?:\.[0-9A-Za-z-]+)*))?(?:\+([0-9A-Za-z-]+(?:\.[0-9A-Za-z-]+)*))?$'
        
        match = re.match(pattern, version_str)
        if not match:
            raise ValueError(f"无效的版本格式: {version_str}")
        
        major = int(match.group(1))
        minor = int(match.group(2))
        patch = int(match.group(3))
        prerelease = match.group(4)
        build_metadata = match.group(5)
        
        return cls(
            major=major,
            minor=minor,
            patch=patch,
            prerelease=prerelease,
            build_metadata=build_metadata
        )
    
    def is_compatible(self, other: 'Version') -> bool:
        """检查版本兼容性"""
        # 主版本号相同，次版本号向后兼容
        if self.major != other.major:
            return False
        
        # 如果当前版本 >= 其他版本，则兼容
        if (self.minor, self.patch) >= (other.minor, other.patch):
            return True
        
        return False
    
    def __lt__(self, other: 'Version') -> bool:
        """版本比较"""
        if self.major != other.major:
            return self.major < other.major
        
        if self.minor != other.minor:
            return self.minor < other.minor
        
        if self.patch != other.patch:
            return self.patch < other.patch
        
        # 预发布版本比较
        if self.prerelease and not other.prerelease:
            return True
        if not self.prerelease and other.prerelease:
            return False
        if self.prerelease and other.prerelease:
            return self.prerelease < other.prerelease
        
        return False
    
    def __eq__(self, other: 'Version') -> bool:
        """版本相等"""
        return (
            self.major == other.major and
            self.minor == other.minor and
            self.patch == other.patch and
            self.prerelease == other.prerelease
        )

# 当前 MCP 版本
MCP_VERSION = Version(1, 0, 0)

2. 版本协商机制

python
from typing import Dict, List, Optional

class VersionNegotiator:
    """版本协商器"""
    
    def __init__(self, server_version: Version):
        self.server_version = server_version
        self.supported_versions: List[Version] = [
            Version(1, 0, 0),
            Version(0, 9, 0),
        ]
    
    def negotiate_version(
        self,
        client_versions: List[Version]
    ) -> Optional[Version]:
        """协商最佳版本"""
        # 找到客户端支持的最高版本
        client_versions_sorted = sorted(client_versions, reverse=True)
        
        for client_version in client_versions_sorted:
            # 检查服务器是否支持此版本
            for server_version in self.supported_versions:
                if server_version.is_compatible(client_version):
                    return server_version
        
        # 没有找到兼容版本
        return None
    
    def get_server_info(self) -> Dict:
        """获取服务器版本信息"""
        return {
            "version": str(self.server_version),
            "supported_versions": [
                str(v) for v in self.supported_versions
            ]
        }

class MCPClient:
    """MCP 客户端"""
    
    def __init__(self, supported_versions: List[Version]):
        self.supported_versions = supported_versions
        self.negotiated_version: Optional[Version] = None
    
    async def connect(self, server_info: Dict) -> bool:
        """连接到服务器并协商版本"""
        server_version = Version.parse(server_info["version"])
        server_supported_versions = [
            Version.parse(v)
            for v in server_info["supported_versions"]
        ]
        
        # 创建临时协商器
        negotiator = VersionNegotiator(server_version)
        
        # 协商版本
        self.negotiated_version = negotiator.negotiate_version(
            self.supported_versions
        )
        
        if not self.negotiated_version:
            print("无法协商兼容的版本")
            return False
        
        print(f"协商版本: {self.negotiated_version}")
        return True

3. 功能废弃管理

python
from enum import Enum
from typing import Dict, List, Optional
from datetime import datetime

class DeprecationStatus(Enum):
    """废弃状态"""
    STABLE = "stable"
    DEPRECATED = "deprecated"
    REMOVED = "removed"

@dataclass
class DeprecationInfo:
    """废弃信息"""
    status: DeprecationStatus
    deprecated_in: Optional[Version] = None
    removed_in: Optional[Version] = None
    replacement: Optional[str] = None
    message: Optional[str] = None

class DeprecationManager:
    """废弃管理器"""
    
    def __init__(self):
        self.deprecations: Dict[str, DeprecationInfo] = {}
    
    def deprecate_feature(
        self,
        feature_name: str,
        deprecated_in: Version,
        removed_in: Version,
        replacement: str = None,
        message: str = None
    ):
        """标记功能为废弃"""
        self.deprecations[feature_name] = DeprecationInfo(
            status=DeprecationStatus.DEPRECATED,
            deprecated_in=deprecated_in,
            removed_in=removed_in,
            replacement=replacement,
            message=message
        )
    
    def remove_feature(self, feature_name: str, removed_in: Version):
        """移除功能"""
        if feature_name in self.deprecations:
            self.deprecations[feature_name].status = DeprecationStatus.REMOVED
            self.deprecations[feature_name].removed_in = removed_in
    
    def check_feature(
        self,
        feature_name: str,
        current_version: Version
    ) -> tuple:
        """检查功能状态"""
        if feature_name not in self.deprecations:
            return True, None
        
        info = self.deprecations[feature_name]
        
        if info.status == DeprecationStatus.REMOVED:
            return False, f"功能 {feature_name} 已在版本 {info.removed_in} 中移除"
        
        if info.status == DeprecationStatus.DEPRECATED:
            if current_version >= info.removed_in:
                return False, f"功能 {feature_name} 已在版本 {info.removed_in} 中移除"
            
            warning = f"功能 {feature_name} 已废弃，将在版本 {info.removed_in} 中移除"
            if info.replacement:
                warning += f"，请使用 {info.replacement}"
            
            return True, warning
        
        return True, None
    
    def get_deprecation_warnings(
        self,
        current_version: Version
    ) -> List[str]:
        """获取所有废弃警告"""
        warnings = []
        
        for feature_name, info in self.deprecations.items():
            if info.status == DeprecationStatus.DEPRECATED:
                if current_version < info.removed_in:
                    warning = f"功能 {feature_name} 已废弃"
                    if info.replacement:
                        warning += f"，请使用 {info.replacement}"
                    warnings.append(warning)
        
        return warnings

4. 版本迁移指南

python
from typing import Dict, List, Callable

class MigrationGuide:
    """版本迁移指南"""
    
    def __init__(self):
        self.migrations: Dict[Version, List[Migration]] = {}
    
    def add_migration(
        self,
        from_version: Version,
        to_version: Version,
        migration_func: Callable,
        description: str
    ):
        """添加迁移步骤"""
        if from_version not in self.migrations:
            self.migrations[from_version] = []
        
        migration = Migration(
            from_version=from_version,
            to_version=to_version,
            func=migration_func,
            description=description
        )
        
        self.migrations[from_version].append(migration)
    
    def get_migration_path(
        self,
        from_version: Version,
        to_version: Version
    ) -> List[Migration]:
        """获取迁移路径"""
        if from_version == to_version:
            return []
        
        # 简单实现：直接迁移
        if from_version in self.migrations:
            for migration in self.migrations[from_version]:
                if migration.to_version == to_version:
                    return [migration]
        
        # 复杂实现：查找多步迁移路径
        return self._find_migration_path(from_version, to_version)
    
    def _find_migration_path(
        self,
        from_version: Version,
        to_version: Version,
        visited: set = None
    ) -> List[Migration]:
        """递归查找迁移路径"""
        if visited is None:
            visited = set()
        
        if from_version in visited:
            return []
        
        visited.add(from_version)
        
        if from_version == to_version:
            return []
        
        if from_version not in self.migrations:
            return []
        
        for migration in self.migrations[from_version]:
            path = self._find_migration_path(
                migration.to_version,
                to_version,
                visited.copy()
            )
            
            if path is not None:
                return [migration] + path
        
        return []
    
    async def execute_migration(
        self,
        from_version: Version,
        to_version: Version,
        context: Dict
    ) -> bool:
        """执行迁移"""
        migration_path = self.get_migration_path(from_version, to_version)
        
        if not migration_path:
            print(f"无法找到从 {from_version} 到 {to_version} 的迁移路径")
            return False
        
        print(f"开始迁移: {from_version} -> {to_version}")
        
        for migration in migration_path:
            print(f"执行迁移: {migration.description}")
            
            try:
                await migration.func(context)
                print(f"迁移完成: {migration.description}")
            except Exception as e:
                print(f"迁移失败: {migration.description}, 错误: {e}")
                return False
        
        print(f"迁移完成: {from_version} -> {to_version}")
        return True

@dataclass
class Migration:
    """迁移步骤"""
    from_version: Version
    to_version: Version
    func: Callable
    description: str

5. 版本兼容性检查

python
from typing import Dict, List, Tuple

class CompatibilityChecker:
    """兼容性检查器"""
    
    def __init__(self, current_version: Version):
        self.current_version = current_version
        self.compatibility_matrix: Dict[Version, List[Version]] = {}
    
    def add_compatibility(
        self,
        server_version: Version,
        compatible_client_versions: List[Version]
    ):
        """添加兼容性规则"""
        self.compatibility_matrix[server_version] = compatible_client_versions
    
    def check_compatibility(
        self,
        client_version: Version
    ) -> Tuple[bool, Optional[str]]:
        """检查客户端版本是否兼容"""
        # 检查服务器是否支持此版本
        if self.current_version in self.compatibility_matrix:
            compatible_versions = self.compatibility_matrix[self.current_version]
            
            for compatible_version in compatible_versions:
                if client_version.is_compatible(compatible_version):
                    return True, None
        
        # 使用默认兼容性规则
        if self.current_version.is_compatible(client_version):
            return True, None
        
        return False, f"客户端版本 {client_version} 与服务器版本 {self.current_version} 不兼容"
    
    def get_compatible_versions(self) -> List[Version]:
        """获取所有兼容的客户端版本"""
        compatible_versions = []
        
        if self.current_version in self.compatibility_matrix:
            compatible_versions = self.compatibility_matrix[self.current_version]
        
        return compatible_versions

6. 版本信息 API

python
from fastapi import FastAPI
from typing import Dict

class VersionInfoAPI:
    """版本信息 API"""
    
    def __init__(
        self,
        current_version: Version,
        supported_versions: List[Version],
        deprecation_manager: DeprecationManager,
        migration_guide: MigrationGuide
    ):
        self.current_version = current_version
        self.supported_versions = supported_versions
        self.deprecation_manager = deprecation_manager
        self.migration_guide = migration_guide
    
    def setup_routes(self, app: FastAPI):
        """设置路由"""
        
        @app.get("/version")
        async def get_version() -> Dict:
            """获取当前版本信息"""
            return {
                "version": str(self.current_version),
                "supported_versions": [str(v) for v in self.supported_versions],
                "deprecation_warnings": self.deprecation_manager.get_deprecation_warnings(
                    self.current_version
                )
            }
        
        @app.get("/version/compatibility/{client_version}")
        async def check_compatibility(client_version: str) -> Dict:
            """检查客户端版本兼容性"""
            checker = CompatibilityChecker(self.current_version)
            
            try:
                version = Version.parse(client_version)
            except ValueError:
                return {
                    "compatible": False,
                    "message": "无效的版本格式"
                }
            
            compatible, message = checker.check_compatibility(version)
            
            return {
                "compatible": compatible,
                "message": message,
                "server_version": str(self.current_version)
            }
        
        @app.get("/version/migration/{from_version}/{to_version}")
        async def get_migration_path(
            from_version: str,
            to_version: str
        ) -> Dict:
            """获取迁移路径"""
            try:
                from_ver = Version.parse(from_version)
                to_ver = Version.parse(to_version)
            except ValueError:
                return {
                    "success": False,
                    "message": "无效的版本格式"
                }
            
            path = self.migration_guide.get_migration_path(from_ver, to_ver)
            
            return {
                "success": True,
                "migration_path": [
                    {
                        "from": str(m.from_version),
                        "to": str(m.to_version),
                        "description": m.description
                    }
                    for m in path
                ]
            }

最佳实践：

1. 语义化版本：严格遵循语义化版本控制规范
2. 向后兼容：确保新版本向后兼容旧版本
3. 渐进废弃：提前通知功能废弃，给用户迁移时间
4. 文档完善：提供详细的版本迁移指南
5. 版本协商：实现客户端和服务器的版本协商机制
6. 测试覆盖：为每个版本编写兼容性测试

通过完善的版本管理和兼容性处理，可以确保 MCP 系统的稳定性和可维护性。