MCP 如何支持多租户架构？
MCP 的多租户支持对于企业级应用至关重要，它允许在单一 MCP 服务器实例中为多个客户或组织提供隔离的服务。以下是详细的实现方法：多租户架构设计MCP 多租户应考虑以下方面：数据隔离：确保不同租户的数据完全隔离资源隔离：隔离计算资源和配额安全隔离：实现租户级别的认证和授权性能隔离：防止单个租户影响其他租户1. 租户识别和上下文from typing import Optionalfrom dataclasses import dataclass@dataclassclass TenantContext: """租户上下文""" tenant_id: str tenant_name: str user_id: str permissions: list quotas: dictclass TenantContextManager: def __init__(self): self.contexts = {} def create_context( self, tenant_id: str, tenant_name: str, user_id: str, permissions: list, quotas: dict = None ) -> TenantContext: """创建租户上下文""" context = TenantContext( tenant_id=tenant_id, tenant_name=tenant_name, user_id=user_id, permissions=permissions, quotas=quotas or self._get_default_quotas(tenant_id) ) self.contexts[tenant_id] = context return context def get_context(self, tenant_id: str) -> Optional[TenantContext]: """获取租户上下文""" return self.contexts.get(tenant_id) def set_current_context(self, tenant_id: str): """设置当前租户上下文""" context = self.get_context(tenant_id) if not context: raise ValueError(f"租户 {tenant_id} 不存在") # 使用线程本地存储或异步上下文变量 import contextvars current_tenant.set(context) def _get_default_quotas(self, tenant_id: str) -> dict: """获取默认配额""" return { "max_tools": 100, "max_resources": 1000, "max_requests_per_minute": 1000, "max_storage_mb": 1024 }# 当前租户上下文变量current_tenant = contextvars.ContextVar('current_tenant', default=None)2. 数据隔离from sqlalchemy import create_engine, Column, String, Integer, Textfrom sqlalchemy.ext.declarative import declarative_basefrom sqlalchemy.orm import sessionmaker, scoped_sessionBase = declarative_base()class TenantData(Base): """租户数据表""" __tablename__ = 'tenant_data' id = Column(Integer, primary_key=True) tenant_id = Column(String(50), nullable=False, index=True) data_key = Column(String(100), nullable=False) data_value = Column(Text) __table_args__ = ( # 确保租户隔离 Index('idx_tenant_key', 'tenant_id', 'data_key', unique=True), )class MultiTenantDatabase: def __init__(self, database_url: str): self.engine = create_engine(database_url) Base.metadata.create_all(self.engine) self.SessionLocal = scoped_session( sessionmaker(autocommit=False, autoflush=False, bind=self.engine) ) def get_session(self, tenant_id: str): """获取租户专属的数据库会话""" session = self.SessionLocal() # 添加租户过滤器 from sqlalchemy import event @event.listens_for(session, 'before_flush') def add_tenant_filter(session, context, instances): for instance in session.new: if hasattr(instance, 'tenant_id'): instance.tenant_id = tenant_id return session def query_tenant_data( self, tenant_id: str, data_key: str ) -> Optional[str]: """查询租户数据""" session = self.get_session(tenant_id) try: result = session.query(TenantData).filter( TenantData.tenant_id == tenant_id, TenantData.data_key == data_key ).first() return result.data_value if result else None finally: session.close() def save_tenant_data( self, tenant_id: str, data_key: str, data_value: str ): """保存租户数据""" session = self.get_session(tenant_id) try: existing = session.query(TenantData).filter( TenantData.tenant_id == tenant_id, TenantData.data_key == data_key ).first() if existing: existing.data_value = data_value else: new_data = TenantData( tenant_id=tenant_id, data_key=data_key, data_value=data_value ) session.add(new_data) session.commit() except Exception as e: session.rollback() raise e finally: session.close()3. 资源配额管理from collections import defaultdictimport timeclass QuotaManager: def __init__(self): self.quotas = {} self.usage = defaultdict(lambda: defaultdict(int)) self.rate_limits = {} def set_quota( self, tenant_id: str, quota_type: str, limit: int ): """设置租户配额""" if tenant_id not in self.quotas: self.quotas[tenant_id] = {} self.quotas[tenant_id][quota_type] = limit def check_quota( self, tenant_id: str, quota_type: str, amount: int = 1 ) -> bool: """检查配额是否足够""" if tenant_id not in self.quotas: return True limit = self.quotas[tenant_id].get(quota_type) if limit is None: return True current_usage = self.usage[tenant_id][quota_type] return current_usage + amount <= limit def consume_quota( self, tenant_id: str, quota_type: str, amount: int = 1 ) -> bool: """消耗配额""" if not self.check_quota(tenant_id, quota_type, amount): return False self.usage[tenant_id][quota_type] += amount return True def get_usage( self, tenant_id: str, quota_type: str ) -> int: """获取使用量""" return self.usage[tenant_id][quota_type] def reset_usage(self, tenant_id: str): """重置使用量""" if tenant_id in self.usage: self.usage[tenant_id].clear() def check_rate_limit( self, tenant_id: str, window: int = 60, max_requests: int = 100 ) -> bool: """检查速率限制""" now = time.time() if tenant_id not in self.rate_limits: self.rate_limits[tenant_id] = [] # 清理过期的请求记录 self.rate_limits[tenant_id] = [ timestamp for timestamp in self.rate_limits[tenant_id] if now - timestamp < window ] # 检查是否超过限制 if len(self.rate_limits[tenant_id]) >= max_requests: return False # 记录新请求 self.rate_limits[tenant_id].append(now) return True4. 租户级别的工具和资源from mcp.server import Serverfrom functools import wrapsclass MultiTenantServer(Server): def __init__(self, name: str, tenant_manager: TenantContextManager): super().__init__(name) self.tenant_manager = tenant_manager self.tenant_tools = defaultdict(dict) self.tenant_resources = defaultdict(dict) def tenant_tool( self, name: str, description: str, tenant_id: str = None ): """租户专属工具装饰器""" def decorator(func): # 注册工具 self.tenant_tools[tenant_id or "default"][name] = { "function": func, "description": description } @wraps(func) async def wrapper(*args, **kwargs): # 获取当前租户 context = current_tenant.get() if not context: raise PermissionError("未找到租户上下文") # 检查租户权限 if tenant_id and context.tenant_id != tenant_id: raise PermissionError("无权访问此工具") # 执行工具 return await func(*args, **kwargs) return wrapper return decorator def tenant_resource( self, uri: str, name: str, description: str, tenant_id: str = None ): """租户专属资源装饰器""" def decorator(func): # 注册资源 self.tenant_resources[tenant_id or "default"][uri] = { "function": func, "name": name, "description": description } @wraps(func) async def wrapper(*args, **kwargs): # 获取当前租户 context = current_tenant.get() if not context: raise PermissionError("未找到租户上下文") # 检查租户权限 if tenant_id and context.tenant_id != tenant_id: raise PermissionError("无权访问此资源") # 执行资源 return await func(*args, **kwargs) return wrapper return decorator async def list_tools(self, tenant_id: str = None) -> list: """列出可用工具""" context = current_tenant.get() if not context: return [] # 获取默认工具和租户专属工具 tools = [] # 添加默认工具 for name, tool_info in self.tenant_tools["default"].items(): tools.append({ "name": name, "description": tool_info["description"] }) # 添加租户专属工具 if context.tenant_id in self.tenant_tools: for name, tool_info in self.tenant_tools[context.tenant_id].items(): tools.append({ "name": name, "description": tool_info["description"] }) return tools5. 租户认证和授权import jwtfrom datetime import datetime, timedeltafrom typing import Dict, Anyclass TenantAuthenticator: def __init__(self, secret_key: str): self.secret_key = secret_key def generate_token( self, tenant_id: str, user_id: str, permissions: list, expires_in: int = 3600 ) -> str: """生成租户令牌""" payload = { "tenant_id": tenant_id, "user_id": user_id, "permissions": permissions, "exp": datetime.utcnow() + timedelta(seconds=expires_in), "iat": datetime.utcnow() } token = jwt.encode(payload, self.secret_key, algorithm="HS256") return token def verify_token(self, token: str) -> Dict[str, Any]: """验证租户令牌""" try: payload = jwt.decode(token, self.secret_key, algorithms=["HS256"]) return payload except jwt.ExpiredSignatureError: raise ValueError("令牌已过期") except jwt.InvalidTokenError: raise ValueError("无效的令牌") def check_permission( self, token: str, required_permission: str ) -> bool: """检查权限""" payload = self.verify_token(token) permissions = payload.get("permissions", []) return required_permission in permissions or "admin" in permissions6. 租户监控和报告from collections import defaultdictfrom datetime import datetime, timedeltaclass TenantMonitor: def __init__(self): self.metrics = defaultdict(lambda: defaultdict(list)) def record_metric( self, tenant_id: str, metric_name: str, value: float ): """记录指标""" timestamp = datetime.now() self.metrics[tenant_id][metric_name].append({ "value": value, "timestamp": timestamp }) # 限制历史记录大小 if len(self.metrics[tenant_id][metric_name]) > 1000: self.metrics[tenant_id][metric_name] = \ self.metrics[tenant_id][metric_name][-1000:] def get_metrics( self, tenant_id: str, metric_name: str, since: datetime = None ) -> list: """获取指标""" if tenant_id not in self.metrics: return [] if metric_name not in self.metrics[tenant_id]: return [] records = self.metrics[tenant_id][metric_name] if since: records = [ record for record in records if record["timestamp"] >= since ] return records def get_aggregated_metrics( self, tenant_id: str, metric_name: str, since: datetime = None ) -> dict: """获取聚合指标""" records = self.get_metrics(tenant_id, metric_name, since) if not records: return {} values = [record["value"] for record in records] return { "count": len(values), "sum": sum(values), "avg": sum(values) / len(values), "min": min(values), "max": max(values) } def generate_tenant_report( self, tenant_id: str, since: datetime = None ) -> dict: """生成租户报告""" if not since: since = datetime.now() - timedelta(days=7) report = { "tenant_id": tenant_id, "period": { "start": since, "end": datetime.now() }, "metrics": {} } if tenant_id in self.metrics: for metric_name in self.metrics[tenant_id]: report["metrics"][metric_name] = \ self.get_aggregated_metrics(tenant_id, metric_name, since) return report最佳实践：数据隔离：使用租户 ID 作为所有数据表的主键或索引配额管理：为每个租户设置合理的资源配额权限控制：实施细粒度的租户级权限控制性能监控：监控每个租户的资源使用情况安全审计：记录所有租户操作用于审计弹性扩展：根据租户需求动态扩展资源通过完善的多租户支持，可以在单一 MCP 服务器实例中为多个客户或组织提供隔离、安全、高效的服务。
