How does MCP handle version management and compatibility?

Version management and compatibility for MCP are crucial for ensuring system stability and maintainability. Here are detailed version management strategies and compatibility handling methods:

Version Management Strategy

MCP version management should consider following aspects:

1. Semantic Versioning: Use semantic version numbers (SemVer)
2. Backward Compatibility: Ensure new versions are backward compatible with old versions
3. Deprecation Strategy: Clear feature deprecation and removal process
4. Migration Guide: Provide detailed version migration guide
5. Version Negotiation: Client-server version negotiation mechanism

1. Semantic Version Control

python
from dataclasses import dataclass
from typing import Optional
import re

@dataclass
class Version:
    """Version number"""
    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':
        """Parse version string"""
        # Match semantic version format
        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"Invalid version format: {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:
        """Check version compatibility"""
        # Same major version, backward compatible minor version
        if self.major != other.major:
            return False
        
        # Compatible if current version >= other version
        if (self.minor, self.patch) >= (other.minor, other.patch):
            return True
        
        return False
    
    def __lt__(self, other: 'Version') -> bool:
        """Version comparison"""
        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
        
        # Prerelease version comparison
        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:
        """Version equality"""
        return (
            self.major == other.major and
            self.minor == other.minor and
            self.patch == other.patch and
            self.prerelease == other.prerelease
        )

# Current MCP version
MCP_VERSION = Version(1, 0, 0)

2. Version Negotiation Mechanism

python
from typing import Dict, List, Optional

class VersionNegotiator:
    """Version negotiator"""
    
    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]:
        """Negotiate best version"""
        # Find highest version supported by client
        client_versions_sorted = sorted(client_versions, reverse=True)
        
        for client_version in client_versions_sorted:
            # Check if server supports this version
            for server_version in self.supported_versions:
                if server_version.is_compatible(client_version):
                    return server_version
        
        # No compatible version found
        return None
    
    def get_server_info(self) -> Dict:
        """Get server version information"""
        return {
            "version": str(self.server_version),
            "supported_versions": [
                str(v) for v in self.supported_versions
            ]
        }

class MCPClient:
    """MCP client"""
    
    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:
        """Connect to server and negotiate version"""
        server_version = Version.parse(server_info["version"])
        server_supported_versions = [
            Version.parse(v)
            for v in server_info["supported_versions"]
        ]
        
        # Create temporary negotiator
        negotiator = VersionNegotiator(server_version)
        
        # Negotiate version
        self.negotiated_version = negotiator.negotiate_version(
            self.supported_versions
        )
        
        if not self.negotiated_version:
            print("Unable to negotiate compatible version")
            return False
        
        print(f"Negotiated version: {self.negotiated_version}")
        return True

3. Feature Deprecation Management

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

class DeprecationStatus(Enum):
    """Deprecation status"""
    STABLE = "stable"
    DEPRECATED = "deprecated"
    REMOVED = "removed"

@dataclass
class DeprecationInfo:
    """Deprecation information"""
    status: DeprecationStatus
    deprecated_in: Optional[Version] = None
    removed_in: Optional[Version] = None
    replacement: Optional[str] = None
    message: Optional[str] = None

class DeprecationManager:
    """Deprecation manager"""
    
    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
    ):
        """Mark feature as deprecated"""
        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):
        """Remove feature"""
        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:
        """Check feature status"""
        if feature_name not in self.deprecations:
            return True, None
        
        info = self.deprecations[feature_name]
        
        if info.status == DeprecationStatus.REMOVED:
            return False, f"Feature {feature_name} was removed in version {info.removed_in}"
        
        if info.status == DeprecationStatus.DEPRECATED:
            if current_version >= info.removed_in:
                return False, f"Feature {feature_name} was removed in version {info.removed_in}"
            
            warning = f"Feature {feature_name} is deprecated and will be removed in version {info.removed_in}"
            if info.replacement:
                warning += f", please use {info.replacement}"
            
            return True, warning
        
        return True, None
    
    def get_deprecation_warnings(
        self,
        current_version: Version
    ) -> List[str]:
        """Get all deprecation warnings"""
        warnings = []
        
        for feature_name, info in self.deprecations.items():
            if info.status == DeprecationStatus.DEPRECATED:
                if current_version < info.removed_in:
                    warning = f"Feature {feature_name} is deprecated"
                    if info.replacement:
                        warning += f", please use {info.replacement}"
                    warnings.append(warning)
        
        return warnings

4. Version Migration Guide

python
from typing import Dict, List, Callable

class MigrationGuide:
    """Version migration guide"""
    
    def __init__(self):
        self.migrations: Dict[Version, List[Migration]] = {}
    
    def add_migration(
        self,
        from_version: Version,
        to_version: Version,
        migration_func: Callable,
        description: str
    ):
        """Add migration step"""
        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]:
        """Get migration path"""
        if from_version == to_version:
            return []
        
        # Simple implementation: direct migration
        if from_version in self.migrations:
            for migration in self.migrations[from_version]:
                if migration.to_version == to_version:
                    return [migration]
        
        # Complex implementation: find multi-step migration path
        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]:
        """Recursively find migration path"""
        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:
        """Execute migration"""
        migration_path = self.get_migration_path(from_version, to_version)
        
        if not migration_path:
            print(f"Unable to find migration path from {from_version} to {to_version}")
            return False
        
        print(f"Starting migration: {from_version} -> {to_version}")
        
        for migration in migration_path:
            print(f"Executing migration: {migration.description}")
            
            try:
                await migration.func(context)
                print(f"Migration completed: {migration.description}")
            except Exception as e:
                print(f"Migration failed: {migration.description}, error: {e}")
                return False
        
        print(f"Migration completed: {from_version} -> {to_version}")
        return True

@dataclass
class Migration:
    """Migration step"""
    from_version: Version
    to_version: Version
    func: Callable
    description: str

5. Version Compatibility Check

python
from typing import Dict, List, Tuple

class CompatibilityChecker:
    """Compatibility checker"""
    
    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]
    ):
        """Add compatibility rule"""
        self.compatibility_matrix[server_version] = compatible_client_versions
    
    def check_compatibility(
        self,
        client_version: Version
    ) -> Tuple[bool, Optional[str]]:
        """Check if client version is compatible"""
        # Check if server supports this version
        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
        
        # Use default compatibility rule
        if self.current_version.is_compatible(client_version):
            return True, None
        
        return False, f"Client version {client_version} is not compatible with server version {self.current_version}"
    
    def get_compatible_versions(self) -> List[Version]:
        """Get all compatible client versions"""
        compatible_versions = []
        
        if self.current_version in self.compatibility_matrix:
            compatible_versions = self.compatibility_matrix[self.current_version]
        
        return compatible_versions

6. Version Information API

python
from fastapi import FastAPI
from typing import Dict

class VersionInfoAPI:
    """Version information 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):
        """Setup routes"""
        
        @app.get("/version")
        async def get_version() -> Dict:
            """Get current version information"""
            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:
            """Check client version compatibility"""
            checker = CompatibilityChecker(self.current_version)
            
            try:
                version = Version.parse(client_version)
            except ValueError:
                return {
                    "compatible": False,
                    "message": "Invalid version format"
                }
            
            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:
            """Get migration path"""
            try:
                from_ver = Version.parse(from_version)
                to_ver = Version.parse(to_version)
            except ValueError:
                return {
                    "success": False,
                    "message": "Invalid version format"
                }
            
            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
                ]
            }

Best Practices:

1. Semantic Versioning: Strictly follow semantic versioning specification
2. Backward Compatibility: Ensure new versions are backward compatible with old versions
3. Gradual Deprecation: Notify feature deprecation in advance, give users migration time
4. Complete Documentation: Provide detailed version migration guide
5. Version Negotiation: Implement client-server version negotiation mechanism
6. Test Coverage: Write compatibility tests for each version

Through comprehensive version management and compatibility handling, you can ensure MCP system stability and maintainability.