How does resource management work in MCP?

MCP's resource management mechanism allows LLMs to access and manipulate external resources such as files, database records, API endpoints, etc. Here are detailed implementation methods:

Resource Definition

MCP resources are identified and accessed through URIs (Uniform Resource Identifiers):

python
{
    "uri": "file:///path/to/resource",
    "name": "Resource Name",
    "description": "Resource description",
    "mimeType": "text/plain"
}

1. Resource Types

MCP supports multiple resource types:

• File Resources: file:///path/to/file
• HTTP Resources: http://example.com/api/resource
• Database Resources: db://database/table/id
• Custom Resources: custom://resource-type/id

2. Resource Registration

python
from mcp.server import Server
from mcp.types import Resource

server = Server("my-mcp-server")

@server.resource(
    uri="file:///config/app.json",
    name="Application Config",
    description="Application configuration file",
    mimeType="application/json"
)
async def get_app_config() -> str:
    """Get application configuration"""
    return """
    {
        "name": "MyApp",
        "version": "1.0.0",
        "settings": {
            "debug": false,
            "maxConnections": 100
        }
    }
    """

@server.resource(
    uri="db://users/{id}",
    name="User Information",
    description="User detailed information",
    mimeType="application/json"
)
async def get_user(id: str) -> str:
    """Get user information"""
    user = await database.get_user(id)
    return json.dumps(user)

3. Resource Access Control

python
class ResourceAccessControl:
    def __init__(self):
        self.permissions = {}
        self.acl = {}
    
    def grant_permission(self, user: str, resource_pattern: str, access: str):
        """Grant resource access permission"""
        if user not in self.permissions:
            self.permissions[user] = []
        
        self.permissions[user].append({
            "pattern": resource_pattern,
            "access": access  # "read", "write", "delete"
        })
    
    def check_permission(self, user: str, resource_uri: str, access: str) -> bool:
        """Check access permission"""
        if user not in self.permissions:
            return False
        
        for perm in self.permissions[user]:
            if self._match_pattern(perm["pattern"], resource_uri):
                if access in perm["access"] or perm["access"] == "all":
                    return True
        
        return False
    
    def _match_pattern(self, pattern: str, uri: str) -> bool:
        """Match resource pattern"""
        import re
        # Convert wildcards to regex
        regex = pattern.replace("*", ".*").replace("?", ".")
        return re.match(regex, uri) is not None

4. Resource Caching

python
from functools import lru_cache
from datetime import datetime, timedelta
import hashlib

class ResourceCache:
    def __init__(self, ttl: int = 3600):
        self.cache = {}
        self.ttl = ttl
    
    def get(self, resource_uri: str) -> Optional[str]:
        """Get cached resource"""
        if resource_uri not in self.cache:
            return None
        
        entry = self.cache[resource_uri]
        
        # Check if expired
        if datetime.now() > entry["expires"]:
            del self.cache[resource_uri]
            return None
        
        return entry["data"]
    
    def set(self, resource_uri: str, data: str):
        """Set cached resource"""
        self.cache[resource_uri] = {
            "data": data,
            "expires": datetime.now() + timedelta(seconds=self.ttl),
            "hash": hashlib.md5(data.encode()).hexdigest()
        }
    
    def invalidate(self, resource_uri: str):
        """Invalidate cache"""
        if resource_uri in self.cache:
            del self.cache[resource_uri]
    
    def clear(self):
        """Clear cache"""
        self.cache.clear()

5. Resource Version Control

python
class ResourceVersionManager:
    def __init__(self):
        self.versions = {}
    
    def save_version(self, resource_uri: str, data: str, version: str):
        """Save resource version"""
        if resource_uri not in self.versions:
            self.versions[resource_uri] = {}
        
        self.versions[resource_uri][version] = {
            "data": data,
            "timestamp": datetime.now().isoformat()
        }
    
    def get_version(self, resource_uri: str, version: str) -> Optional[str]:
        """Get specified version"""
        if resource_uri not in self.versions:
            return None
        
        return self.versions[resource_uri].get(version, {}).get("data")
    
    def list_versions(self, resource_uri: str) -> List[str]:
        """List all versions"""
        if resource_uri not in self.versions:
            return []
        
        return list(self.versions[resource_uri].keys())
    
    def get_latest_version(self, resource_uri: str) -> Optional[str]:
        """Get latest version"""
        versions = self.list_versions(resource_uri)
        if not versions:
            return None
        
        return max(versions)

6. Resource Monitoring

python
class ResourceMonitor:
    def __init__(self):
        self.access_log = []
        self.metrics = {
            "total_access": 0,
            "unique_resources": set(),
            "access_by_type": {}
        }
    
    def log_access(self, resource_uri: str, user: str, action: str):
        """Log resource access"""
        log_entry = {
            "timestamp": datetime.now().isoformat(),
            "resource": resource_uri,
            "user": user,
            "action": action
        }
        
        self.access_log.append(log_entry)
        
        # Update metrics
        self.metrics["total_access"] += 1
        self.metrics["unique_resources"].add(resource_uri)
        
        # Statistics by type
        resource_type = resource_uri.split("://")[0]
        if resource_type not in self.metrics["access_by_type"]:
            self.metrics["access_by_type"][resource_type] = 0
        self.metrics["access_by_type"][resource_type] += 1
    
    def get_metrics(self) -> dict:
        """Get monitoring metrics"""
        return {
            "total_access": self.metrics["total_access"],
            "unique_resources": len(self.metrics["unique_resources"]),
            "access_by_type": self.metrics["access_by_type"]
        }
    
    def get_access_history(self, resource_uri: str, limit: int = 100) -> List[dict]:
        """Get access history"""
        filtered = [
            log for log in self.access_log
            if log["resource"] == resource_uri
        ]
        return filtered[-limit:]

7. Resource Lifecycle Management

python
class ResourceLifecycleManager:
    def __init__(self):
        self.resources = {}
        self.cleanup_interval = 3600  # 1 hour
    
    def register_resource(self, resource_uri: str, metadata: dict):
        """Register resource"""
        self.resources[resource_uri] = {
            "metadata": metadata,
            "created_at": datetime.now(),
            "last_accessed": datetime.now(),
            "access_count": 0
        }
    
    def access_resource(self, resource_uri: str):
        """Access resource"""
        if resource_uri in self.resources:
            self.resources[resource_uri]["last_accessed"] = datetime.now()
            self.resources[resource_uri]["access_count"] += 1
    
    def cleanup_old_resources(self, max_age_days: int = 30):
        """Clean up old resources"""
        cutoff = datetime.now() - timedelta(days=max_age_days)
        
        to_remove = []
        for uri, info in self.resources.items():
            if info["last_accessed"] < cutoff:
                to_remove.append(uri)
        
        for uri in to_remove:
            del self.resources[uri]
        
        return len(to_remove)
    
    def get_resource_stats(self) -> dict:
        """Get resource statistics"""
        return {
            "total_resources": len(self.resources),
            "total_accesses": sum(r["access_count"] for r in self.resources.values()),
            "oldest_resource": min(
                (r["created_at"] for r in self.resources.values()),
                default=None
            )
        }

Best Practices:

1. URI Design: Use clear, hierarchical URI structures
2. Access Control: Implement principle of least privilege
3. Caching Strategy: Set appropriate TTL based on resource characteristics
4. Monitoring and Logging: Log all resource access operations
5. Version Management: Implement version control for important resources
6. Regular Cleanup: Clean up unused resources

Through comprehensive resource management mechanisms, you can ensure secure and efficient access to resources in MCP systems.