MCP 中的资源管理是如何工作的？

MCP 的资源管理机制允许 LLM 访问和操作外部资源，如文件、数据库记录、API 端点等。以下是详细的实现方法：

资源定义

MCP 资源通过 URI（统一资源标识符）进行标识和访问：

python
{
    "uri": "file:///path/to/resource",
    "name": "资源名称",
    "description": "资源描述",
    "mimeType": "text/plain"
}

1. 资源类型

MCP 支持多种资源类型：

• 文件资源：file:///path/to/file
• HTTP 资源：http://example.com/api/resource
• 数据库资源：db://database/table/id
• 自定义资源：custom://resource-type/id

2. 资源注册

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

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

@server.resource(
    uri="file:///config/app.json",
    name="应用配置",
    description="应用程序的配置文件",
    mimeType="application/json"
)
async def get_app_config() -> str:
    """获取应用配置"""
    return """
    {
        "name": "MyApp",
        "version": "1.0.0",
        "settings": {
            "debug": false,
            "maxConnections": 100
        }
    }
    """

@server.resource(
    uri="db://users/{id}",
    name="用户信息",
    description="用户详细信息",
    mimeType="application/json"
)
async def get_user(id: str) -> str:
    """获取用户信息"""
    user = await database.get_user(id)
    return json.dumps(user)

3. 资源访问控制

python
class ResourceAccessControl:
    def __init__(self):
        self.permissions = {}
        self.acl = {}
    
    def grant_permission(self, user: str, resource_pattern: str, access: str):
        """授予资源访问权限"""
        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:
        """检查访问权限"""
        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:
        """匹配资源模式"""
        import re
        # 将通配符转换为正则表达式
        regex = pattern.replace("*", ".*").replace("?", ".")
        return re.match(regex, uri) is not None

4. 资源缓存

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]:
        """获取缓存资源"""
        if resource_uri not in self.cache:
            return None
        
        entry = self.cache[resource_uri]
        
        # 检查是否过期
        if datetime.now() > entry["expires"]:
            del self.cache[resource_uri]
            return None
        
        return entry["data"]
    
    def set(self, resource_uri: str, data: str):
        """设置缓存资源"""
        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):
        """使缓存失效"""
        if resource_uri in self.cache:
            del self.cache[resource_uri]
    
    def clear(self):
        """清空缓存"""
        self.cache.clear()

5. 资源版本控制

python
class ResourceVersionManager:
    def __init__(self):
        self.versions = {}
    
    def save_version(self, resource_uri: str, data: str, version: str):
        """保存资源版本"""
        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]:
        """获取指定版本"""
        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]:
        """列出所有版本"""
        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]:
        """获取最新版本"""
        versions = self.list_versions(resource_uri)
        if not versions:
            return None
        
        return max(versions)

6. 资源监控

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_entry = {
            "timestamp": datetime.now().isoformat(),
            "resource": resource_uri,
            "user": user,
            "action": action
        }
        
        self.access_log.append(log_entry)
        
        # 更新指标
        self.metrics["total_access"] += 1
        self.metrics["unique_resources"].add(resource_uri)
        
        # 按类型统计
        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:
        """获取监控指标"""
        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]:
        """获取访问历史"""
        filtered = [
            log for log in self.access_log
            if log["resource"] == resource_uri
        ]
        return filtered[-limit:]

7. 资源生命周期管理

python
class ResourceLifecycleManager:
    def __init__(self):
        self.resources = {}
        self.cleanup_interval = 3600  # 1小时
    
    def register_resource(self, resource_uri: str, metadata: dict):
        """注册资源"""
        self.resources[resource_uri] = {
            "metadata": metadata,
            "created_at": datetime.now(),
            "last_accessed": datetime.now(),
            "access_count": 0
        }
    
    def access_resource(self, resource_uri: str):
        """访问资源"""
        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):
        """清理旧资源"""
        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:
        """获取资源统计"""
        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
            )
        }

最佳实践：

1. URI 设计：使用清晰、层次化的 URI 结构
2. 权限控制：实施最小权限原则
3. 缓存策略：根据资源特性设置合适的 TTL
4. 监控和日志：记录所有资源访问操作
5. 版本管理：对重要资源实施版本控制
6. 定期清理：清理不再使用的资源

通过完善的资源管理机制，可以确保 MCP 系统中资源的安全、高效访问。