MCP Server 如何管理资源访问

{
  "name": "resource_name",
  "type": "resource_type",
  "description": "资源描述",
  "uri": "resource_uri",
  "access_methods": ["read", "write", "execute", "delete"],
  "permission_model": "rbac",
  "properties": {
    "property1": {
      "type": "string",
      "description": "属性1描述"
    },
    "property2": {
      "type": "number",
      "description": "属性2描述"
    }
  },
  "metadata": {
    "owner": "resource_owner",
    "created_at": "2026-01-01T00:00:00Z",
    "updated_at": "2026-01-01T00:00:00Z",
    "tags": ["tag1", "tag2"]
  },
  "quota": {
    "read": 1000,
    "write": 100,
    "execute": 50,
    "delete": 10
  }
}

from typing import Dict, List, Any

class RBACAccessControl:
    def __init__(self):
        # 角色权限映射：{role: {resource: [methods]}}
        self.role_permissions: Dict[str, Dict[str, List[str]]] = {}
        # 用户角色映射：{user: [roles]}
        self.user_roles: Dict[str, List[str]] = {}
    
    def add_role(self, role: str):
        """添加角色"""
        if role not in self.role_permissions:
            self.role_permissions[role] = {}
    
    def assign_role(self, user: str, role: str):
        """为用户分配角色"""
        if user not in self.user_roles:
            self.user_roles[user] = []
        if role not in self.user_roles[user]:
            self.user_roles[user].append(role)
    
    def remove_role(self, user: str, role: str):
        """移除用户的角色"""
        if user in self.user_roles and role in self.user_roles[user]:
            self.user_roles[user].remove(role)
            if not self.user_roles[user]:
                del self.user_roles[user]
    
    def grant_permission(self, role: str, resource: str, methods: List[str]):
        """为角色授予资源访问权限"""
        if role not in self.role_permissions:
            self.add_role(role)
        if resource not in self.role_permissions[role]:
            self.role_permissions[role][resource] = []
        for method in methods:
            if method not in self.role_permissions[role][resource]:
                self.role_permissions[role][resource].append(method)
    
    def revoke_permission(self, role: str, resource: str, methods: List[str] = None):
        """撤销角色的资源访问权限"""
        if role in self.role_permissions and resource in self.role_permissions[role]:
            if methods:
                # 撤销指定方法的权限
                for method in methods:
                    if method in self.role_permissions[role][resource]:
                        self.role_permissions[role][resource].remove(method)
                # 如果该资源没有权限了，移除该资源
                if not self.role_permissions[role][resource]:
                    del self.role_permissions[role][resource]
            else:
                # 撤销所有方法的权限
                del self.role_permissions[role][resource]
            # 如果该角色没有任何权限了，移除该角色
            if not self.role_permissions[role]:
                del self.role_permissions[role]
    
    def check_access(self, user: str, resource: str, method: str) -> bool:
        """检查用户是否有资源访问权限"""
        if user not in self.user_roles:
            return False
        
        for role in self.user_roles[user]:
            if role in self.role_permissions and resource in self.role_permissions[role]:
                if method in self.role_permissions[role][resource] or "*" in self.role_permissions[role][resource]:
                    return True
        
        return False
    
    def get_user_permissions(self, user: str) -> Dict[str, List[str]]:
        """获取用户的所有资源访问权限"""
        permissions = {}
        if user in self.user_roles:
            for role in self.user_roles[user]:
                if role in self.role_permissions:
                    for resource, methods in self.role_permissions[role].items():
                        if resource not in permissions:
                            permissions[resource] = []
                        for method in methods:
                            if method not in permissions[resource]:
                                permissions[resource].append(method)
        return permissions

from typing import Dict, Any, Callable

class ABACAccessControl:
    def __init__(self):
        # 访问控制策略：{policy_id: (condition, effect)}
        self.policies: Dict[str, tuple] = {}
    
    def add_policy(self, policy_id: str, condition: Callable[[Dict[str, Any]], bool], effect: bool):
        """添加访问控制策略"""
        self.policies[policy_id] = (condition, effect)
    
    def remove_policy(self, policy_id: str):
        """移除访问控制策略"""
        if policy_id in self.policies:
            del self.policies[policy_id]
    
    def check_access(self, context: Dict[str, Any]) -> bool:
        """检查访问是否允许"""
        # 上下文包含访问者属性、资源属性、操作属性、环境属性
        # 例如：{"accessor": {"id": "user1", "role": "admin"}, "resource": {"id": "resource1", "type": "data"}, "action": "read", "environment": {"time": "2026-01-01T12:00:00Z", "ip": "192.168.1.1"}}
        
        # 默认拒绝访问
        result = False
        
        # 遍历所有策略，找到匹配的策略
        for policy_id, (condition, effect) in self.policies.items():
            if condition(context):
                result = effect
                # 最后匹配的策略生效
        
        return result
    
    def get_matching_policies(self, context: Dict[str, Any]) -> List[str]:
        """获取匹配的策略"""
        matching_policies = []
        for policy_id, (condition, effect) in self.policies.items():
            if condition(context):
                matching_policies.append(policy_id)
        return matching_policies

# 策略示例：允许管理员在工作时间访问所有资源
policy:
  id: "admin_work_time_access"
  name: "管理员工作时间访问权限"
  description: "允许管理员在工作时间访问所有资源"
  effect: "allow"
  conditions:
    - attribute: "accessor.role"
      operator: "equals"
      value: "admin"
    - attribute: "environment.time"
      operator: "between"
      value: ["09:00", "18:00"]
  resources: ["*"]
  actions: ["*"]

# 策略示例：拒绝普通用户访问敏感数据
policy:
  id: "deny_user_sensitive_data"
  name: "拒绝普通用户访问敏感数据"
  description: "拒绝普通用户访问敏感数据"
  effect: "deny"
  conditions:
    - attribute: "accessor.role"
      operator: "equals"
      value: "user"
    - attribute: "resource.type"
      operator: "equals"
      value: "sensitive_data"
  resources: ["*"]
  actions: ["*"]

from typing import Dict, Any
import numpy as np

class RiskBasedAccessControl:
    def __init__(self):
        # 风险评估模型
        self.risk_model = self._build_risk_model()
        # 风险阈值
        self.risk_threshold = 0.7
    
    def _build_risk_model(self):
        """构建风险评估模型"""
        # 这里使用简单的风险评估模型，实际应用中可以使用机器学习模型
        def risk_model(context: Dict[str, Any]) -> float:
            risk = 0.0
            
            # 访问者风险因子
            accessor = context.get("accessor", {})
            if accessor.get("role") == "guest":
                risk += 0.3
            if accessor.get("login_failed_attempts", 0) > 3:
                risk += 0.4
            
            # 资源风险因子
            resource = context.get("resource", {})
            if resource.get("type") == "sensitive_data":
                risk += 0.5
            if resource.get("classification") == "top_secret":
                risk += 0.5
            
            # 操作风险因子
            action = context.get("action", "")
            if action in ["write", "delete", "execute"]:
                risk += 0.3
            
            # 环境风险因子
            environment = context.get("environment", {})
            if environment.get("ip") in ["10.0.0.1", "192.168.1.100"]:
                risk += 0.2
            if environment.get("time") < "06:00" or environment.get("time") > "22:00":
                risk += 0.3
            
            # 归一化风险值到 [0, 1] 范围
            return min(1.0, max(0.0, risk))
        
        return risk_model
    
    def set_risk_threshold(self, threshold: float):
        """设置风险阈值"""
        self.risk_threshold = threshold
    
    def calculate_risk(self, context: Dict[str, Any]) -> float:
        """计算访问请求的风险值"""
        return self.risk_model(context)
    
    def check_access(self, context: Dict[str, Any]) -> bool:
        """检查访问是否允许"""
        risk = self.calculate_risk(context)
        return risk < self.risk_threshold
    
    def get_risk_level(self, risk: float) -> str:
        """获取风险等级"""
        if risk < 0.3:
            return "low"
        elif risk < 0.7:
            return "medium"
        else:
            return "high"

from typing import Dict, List, Any
from .resource import Resource, DataResource, ComputeResource, NetworkResource, ServiceResource, ToolResource, ConfigurationResource

class ResourceManager:
    def __init__(self):
        # 资源注册表：{resource_uri: Resource}
        self.resources: Dict[str, Resource] = {}
    
    def register_resource(self, resource_definition: Dict[str, Any]) -> Resource:
        """注册资源"""
        resource_type = resource_definition.get("type", "")
        resource_uri = resource_definition.get("uri", "")
        
        if not resource_uri:
            raise ValueError("资源 URI 不能为空")
        
        # 根据资源类型创建资源对象
        if resource_type == "data":
            resource = DataResource(**resource_definition)
        elif resource_type == "compute":
            resource = ComputeResource(**resource_definition)
        elif resource_type == "network":
            resource = NetworkResource(**resource_definition)
        elif resource_type == "service":
            resource = ServiceResource(**resource_definition)
        elif resource_type == "tool":
            resource = ToolResource(**resource_definition)
        elif resource_type == "configuration":
            resource = ConfigurationResource(**resource_definition)
        else:
            resource = Resource(**resource_definition)
        
        # 将资源添加到注册表
        self.resources[resource_uri] = resource
        return resource
    
    def unregister_resource(self, resource_uri: str):
        """注销资源"""
        if resource_uri in self.resources:
            del self.resources[resource_uri]
    
    def get_resource(self, resource_uri: str) -> Resource:
        """获取资源"""
        return self.resources.get(resource_uri)
    
    def list_resources(self, resource_type: str = None) -> List[Resource]:
        """列出所有资源"""
        if resource_type:
            return [resource for resource in self.resources.values() if resource.type == resource_type]
        else:
            return list(self.resources.values())
    
    def update_resource(self, resource_uri: str, updates: Dict[str, Any]):
        """更新资源信息"""
        if resource_uri in self.resources:
            resource = self.resources[resource_uri]
            # 更新资源属性
            for key, value in updates.items():
                if hasattr(resource, key):
                    setattr(resource, key, value)
            # 更新资源 URI
            if "uri" in updates and updates["uri"] != resource_uri:
                # 创建新 URI 的资源
                new_uri = updates["uri"]
                self.resources[new_uri] = resource
                # 删除旧 URI 的资源
                del self.resources[resource_uri]
    
    def check_resource_access(self, accessor: Dict[str, Any], resource_uri: str, method: str) -> bool:
        """检查资源访问权限"""
        resource = self.get_resource(resource_uri)
        if not resource:
            return False
        
        # 检查资源是否支持该访问方法
        if method not in resource.access_methods:
            return False
        
        # 调用资源的 checkAccess 方法检查访问权限
        return resource.checkAccess(accessor, method)

from typing import Dict, List, Any
import datetime

class ResourceAccessLog:
    def __init__(self, accessor: Dict[str, Any], resource_uri: str, method: str, success: bool, timestamp: datetime.datetime, ip: str, user_agent: str, risk_level: str = "medium"):
        self.accessor = accessor
        self.resource_uri = resource_uri
        self.method = method
        self.success = success
        self.timestamp = timestamp
        self.ip = ip
        self.user_agent = user_agent
        self.risk_level = risk_level

class ResourceAccessMonitor:
    def __init__(self):
        # 资源访问日志
        self.access_logs: List[ResourceAccessLog] = []
        # 资源使用统计
        self.resource_usage: Dict[str, Dict[str, int]] = {}
    
    def log_access(self, accessor: Dict[str, Any], resource_uri: str, method: str, success: bool, ip: str, user_agent: str, risk_level: str = "medium"):
        """记录资源访问日志"""
        log = ResourceAccessLog(
            accessor=accessor,
            resource_uri=resource_uri,
            method=method,
            success=success,
            timestamp=datetime.datetime.now(),
            ip=ip,
            user_agent=user_agent,
            risk_level=risk_level
        )
        self.access_logs.append(log)
        
        # 更新资源使用统计
        if resource_uri not in self.resource_usage:
            self.resource_usage[resource_uri] = {}
        if method not in self.resource_usage[resource_uri]:
            self.resource_usage[resource_uri][method] = 0
        self.resource_usage[resource_uri][method] += 1
    
    def get_access_logs(self, start_time: datetime.datetime = None, end_time: datetime.datetime = None, resource_uri: str = None, accessor_id: str = None) -> List[ResourceAccessLog]:
        """获取资源访问日志"""
        logs = self.access_logs.copy()
        
        # 按时间过滤
        if start_time:
            logs = [log for log in logs if log.timestamp >= start_time]
        if end_time:
            logs = [log for log in logs if log.timestamp <= end_time]
        
        # 按资源 URI 过滤
        if resource_uri:
            logs = [log for log in logs if log.resource_uri == resource_uri]
        
        # 按访问者 ID 过滤
        if accessor_id:
            logs = [log for log in logs if log.accessor.get("id") == accessor_id]
        
        return logs
    
    def get_resource_usage(self, resource_uri: str = None, method: str = None) -> Dict[str, Any]:
        """获取资源使用统计"""
        usage = self.resource_usage.copy()
        
        # 按资源 URI 过滤
        if resource_uri:
            if resource_uri in usage:
                usage = {resource_uri: usage[resource_uri]}
            else:
                usage = {}
        
        # 按访问方法过滤
        if method:
            filtered_usage = {}
            for resource, methods in usage.items():
                if method in methods:
                    filtered_usage[resource] = {method: methods[method]}
            usage = filtered_usage
        
        return usage
    
    def get_top_resources(self, n: int = 10) -> List[Dict[str, Any]]:
        """获取使用最多的资源"""
        # 计算每个资源的总访问次数
        resource_total = {}
        for resource, methods in self.resource_usage.items():
            total = sum(methods.values())
            resource_total[resource] = total
        
        # 按总访问次数排序
        sorted_resources = sorted(resource_total.items(), key=lambda x: x[1], reverse=True)
        
        # 返回前 n 个资源
        return [{
            "resource_uri": resource,
            "total_accesses": total
        } for resource, total in sorted_resources[:n]]
    
    def detect_anomalies(self) -> List[ResourceAccessLog]:
        """检测异常访问行为"""
        # 简单的异常检测：连续失败的访问尝试
        anomalies = []
        
        # 按访问者 ID 分组日志
        accessor_logs: Dict[str, List[ResourceAccessLog]] = {}
        for log in self.access_logs:
            accessor_id = log.accessor.get("id", "unknown")
            if accessor_id not in accessor_logs:
                accessor_logs[accessor_id] = []
            accessor_logs[accessor_id].append(log)
        
        # 检测连续失败的访问尝试
        for accessor_id, logs in accessor_logs.items():
            # 按时间排序
            logs.sort(key=lambda x: x.timestamp)
            
            # 检测连续失败
            consecutive_failures = 0
            for log in logs:
                if not log.success:
                    consecutive_failures += 1
                    if consecutive_failures >= 5:
                        anomalies.append(log)
                else:
                    consecutive_failures = 0
        
        return anomalies

from typing import Dict, Any

class ResourceQuota:
    def __init__(self, resource_uri: str, quota_type: str, limit: int, used: int = 0, unit: str = "count"):
        self.resource_uri = resource_uri
        self.quota_type = quota_type  # 例如："read", "write", "execute", "total"
        self.limit = limit
        self.used = used
        self.unit = unit
    
    def check_quota(self, amount: int = 1) -> bool:
        """检查配额是否足够"""
        return self.used + amount <= self.limit
    
    def consume_quota(self, amount: int = 1) -> bool:
        """消耗配额"""
        if self.check_quota(amount):
            self.used += amount
            return True
        else:
            return False
    
    def release_quota(self, amount: int = 1):
        """释放配额"""
        self.used = max(0, self.used - amount)
    
    def reset_quota(self):
        """重置配额使用量"""
        self.used = 0
    
    def update_limit(self, new_limit: int):
        """更新配额限制"""
        self.limit = new_limit
    
    def get_usage_percentage(self) -> float:
        """获取配额使用率"""
        if self.limit == 0:
            return 0.0
        return (self.used / self.limit) * 100

class ResourceQuotaManager:
    def __init__(self):
        # 资源配额映射：{resource_uri: {quota_type: ResourceQuota}}
        self.quotas: Dict[str, Dict[str, ResourceQuota]] = {}
    
    def set_quota(self, resource_uri: str, quota_type: str, limit: int, unit: str = "count"):
        """设置资源配额"""
        if resource_uri not in self.quotas:
            self.quotas[resource_uri] = {}
        
        # 如果配额已存在，更新限制，否则创建新配额
        if quota_type in self.quotas[resource_uri]:
            self.quotas[resource_uri][quota_type].update_limit(limit)
            self.quotas[resource_uri][quota_type].unit = unit
        else:
            self.quotas[resource_uri][quota_type] = ResourceQuota(resource_uri, quota_type, limit, 0, unit)
    
    def remove_quota(self, resource_uri: str, quota_type: str = None):
        """移除资源配额"""
        if resource_uri in self.quotas:
            if quota_type:
                # 移除指定类型的配额
                if quota_type in self.quotas[resource_uri]:
                    del self.quotas[resource_uri][quota_type]
                # 如果该资源没有配额了，移除该资源
                if not self.quotas[resource_uri]:
                    del self.quotas[resource_uri]
            else:
                # 移除该资源的所有配额
                del self.quotas[resource_uri]
    
    def check_quota(self, resource_uri: str, quota_type: str, amount: int = 1) -> bool:
        """检查资源配额是否足够"""
        if resource_uri in self.quotas and quota_type in self.quotas[resource_uri]:
            return self.quotas[resource_uri][quota_type].check_quota(amount)
        else:
            # 没有设置配额，默认允许访问
            return True
    
    def consume_quota(self, resource_uri: str, quota_type: str, amount: int = 1) -> bool:
        """消耗资源配额"""
        if resource_uri in self.quotas and quota_type in self.quotas[resource_uri]:
            return self.quotas[resource_uri][quota_type].consume_quota(amount)
        else:
            # 没有设置配额，默认允许访问
            return True
    
    def release_quota(self, resource_uri: str, quota_type: str, amount: int = 1):
        """释放资源配额"""
        if resource_uri in self.quotas and quota_type in self.quotas[resource_uri]:
            self.quotas[resource_uri][quota_type].release_quota(amount)
    
    def reset_quota(self, resource_uri: str, quota_type: str = None):
        """重置资源配额使用量"""
        if resource_uri in self.quotas:
            if quota_type:
                # 重置指定类型的配额
                if quota_type in self.quotas[resource_uri]:
                    self.quotas[resource_uri][quota_type].reset_quota()
            else:
                # 重置该资源的所有配额
                for quota in self.quotas[resource_uri].values():
                    quota.reset_quota()
    
    def get_quota_status(self, resource_uri: str, quota_type: str = None) -> Dict[str, Any]:
        """获取资源配额状态"""
        status = {}
        
        if resource_uri in self.quotas:
            if quota_type:
                # 获取指定类型的配额状态
                if quota_type in self.quotas[resource_uri]:
                    quota = self.quotas[resource_uri][quota_type]
                    status[quota_type] = {
                        "limit": quota.limit,
                        "used": quota.used,
                        "unit": quota.unit,
                        "usage_percentage": quota.get_usage_percentage()
                    }
            else:
                # 获取该资源的所有配额状态
                for qtype, quota in self.quotas[resource_uri].items():
                    status[qtype] = {
                        "limit": quota.limit,
                        "used": quota.used,
                        "unit": quota.unit,
                        "usage_percentage": quota.get_usage_percentage()
                    }
        
        return status
    
    def get_overquota_resources(self) -> List[Dict[str, Any]]:
        """获取配额超标的资源"""
        overquota_resources = []
        
        for resource_uri, quotas in self.quotas.items():
            for quota_type, quota in quotas.items():
                if quota.get_usage_percentage() >= 100:
                    overquota_resources.append({
                        "resource_uri": resource_uri,
                        "quota_type": quota_type,
                        "limit": quota.limit,
                        "used": quota.used,
                        "usage_percentage": quota.get_usage_percentage()
                    })
        
        return overquota_resources

import os
import subprocess
import sys
from typing import Any, Dict

class ProcessIsolation:
    def __init__(self):
        pass
    
    def execute_in_isolation(self, code: str, params: Dict[str, Any]) -> Any:
        """在隔离进程中执行代码"""
        # 创建临时 Python 脚本文件
        import tempfile
        with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False) as f:
            f.write(f"""
import json
import sys

# 执行代码
params = {params_json}
result = None

try:
    # 执行用户提供的代码
    {code}
    # 将结果转换为 JSON 格式输出
    print(json.dumps({{"success": True, "result": result}}))
except Exception as e:
    # 捕获异常并输出错误信息
    print(json.dumps({{"success": False, "error": str(e)}}))
"""".format(
                params_json=json.dumps(params),
                code=code
            ))
            script_path = f.name
        
        try:
            # 在新进程中执行脚本
            result = subprocess.run(
                [sys.executable, script_path],
                capture_output=True,
                text=True,
                timeout=30  # 设置超时时间，防止无限循环
            )
            
            # 解析执行结果
            output = result.stdout.strip()
            if output:
                return json.loads(output)
            else:
                return {"success": False, "error": "No output from process"}
        finally:
            # 删除临时脚本文件
            os.unlink(script_path)

import docker
import json
from typing import Any, Dict

class ContainerIsolation:
    def __init__(self, image: str = "python:3.11-slim"):
        self.client = docker.from_env()
        self.image = image
    
    def execute_in_container(self, code: str, params: Dict[str, Any]) -> Any:
        """在隔离容器中执行代码"""
        # 创建临时 Python 脚本内容
        script_content = f"""
import json
import sys

# 执行代码
params = {params_json}
result = None

try:
    # 执行用户提供的代码
    {code}
    # 将结果转换为 JSON 格式输出
    print(json.dumps({{"success": True, "result": result}}))
except Exception as e:
    # 捕获异常并输出错误信息
    print(json.dumps({{"success": False, "error": str(e)}}))
"""".format(
            params_json=json.dumps(params),
            code=code
        )
        
        # 创建临时文件
        import tempfile
        with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False) as f:
            f.write(script_content)
            script_path = f.name
        
        try:
            # 运行容器并执行脚本
            container = self.client.containers.run(
                self.image,
                command=["python", "-c", script_content],
                detach=False,
                remove=True,
                mem_limit="128m",  # 设置内存限制
                cpu_period=100000,
                cpu_quota=50000,  # 设置 CPU 限制为 0.5 核
                network_mode="none"  # 禁用网络，增强安全性
            )
            
            # 解析执行结果
            output = container.decode('utf-8').strip()
            if output:
                return json.loads(output)
            else:
                return {"success": False, "error": "No output from container"}
        except docker.errors.DockerException as e:
            return {"success": False, "error": str(e)}
        finally:
            # 删除临时脚本文件
            import os
            os.unlink(script_path)

import socket
import subprocess
from typing import List, Dict, Any

class NetworkIsolation:
    def __init__(self, allowed_hosts: List[str] = None, allowed_ports: List[int] = None):
        self.allowed_hosts = allowed_hosts or []
        self.allowed_ports = allowed_ports or []
    
    def set_allowed_hosts(self, hosts: List[str]):
        """设置允许访问的主机"""
        self.allowed_hosts = hosts
    
    def set_allowed_ports(self, ports: List[int]):
        """设置允许访问的端口"""
        self.allowed_ports = ports
    
    def check_network_access(self, host: str, port: int) -> bool:
        """检查网络访问是否允许"""
        # 检查主机是否允许
        if self.allowed_hosts and host not in self.allowed_hosts:
            return False
        
        # 检查端口是否允许
        if self.allowed_ports and port not in self.allowed_ports:
            return False
        
        return True
    
    def block_network_access(self):
        """阻止所有网络访问"""
        # 使用 iptables 阻止所有网络访问
        try:
            subprocess.run(["iptables", "-A", "OUTPUT", "-j", "DROP"], check=True)
        except subprocess.CalledProcessError:
            pass
    
    def allow_network_access(self, host: str, port: int):
        """允许特定主机和端口的网络访问"""
        try:
            # 允许访问指定主机和端口
            subprocess.run(["iptables", "-I", "OUTPUT", "-d", host, "-p", "tcp", "--dport", str(port), "-j", "ACCEPT"], check=True)
        except subprocess.CalledProcessError:
            pass

GET /api/v1/resources
Authorization: Bearer <api-key>
Content-Type: application/json

200 OK
Content-Type: application/json

[
  {
    "name": "user_data",
    "type": "data",
    "description": "用户数据资源",
    "uri": "data://user_data",
    "access_methods": ["read", "write", "delete"],
    "permission_model": "rbac",
    "properties": {
      "data_type": "json",
      "storage_location": "database",
      "encryption": true
    },
    "metadata": {
      "owner": "admin",
      "created_at": "2026-01-01T00:00:00Z",
      "updated_at": "2026-01-01T00:00:00Z",
      "tags": ["user", "data"]
    },
    "quota": {
      "read": 1000,
      "write": 100,
      "delete": 10
    }
  }
]

POST /api/v1/resources/{resource_uri}/quotas
Authorization: Bearer <api-key>
Content-Type: application/json

{
  "quota_type": "read",
  "limit": 1000,
  "unit": "count"
}

200 OK
Content-Type: application/json

{
  "message": "Quota set successfully",
  "resource_uri": "data://user_data",
  "quota": {
    "quota_type": "read",
    "limit": 1000,
    "unit": "count",
    "used": 0
  }
}

GET /api/v1/resources/{resource_uri}/logs
Authorization: Bearer <api-key>
Content-Type: application/json

{
  "start_time": "2026-01-01T00:00:00Z",
  "end_time": "2026-01-02T00:00:00Z",
  "accessor_id": "user1"
}

200 OK
Content-Type: application/json

[
  {
    "accessor": {
      "id": "user1",
      "role": "user"
    },
    "resource_uri": "data://user_data",
    "method": "read",
    "success": true,
    "timestamp": "2026-01-01T12:00:00Z",
    "ip": "192.168.1.1",
    "user_agent": "MCP Client/2.0",
    "risk_level": "low"
  }
]