01-Architecture_核心架构概念
核心架构概念
本实验内容
本实验深入探讨了 MCP 服务器架构模式、数据库设计原则，以及支持强大、可扩展的数据库集成 AI 应用的技术实现策略。
概述
构建一个生产级 MCP 服务器并集成数据库需要慎重的架构决策。本实验分解了关键组件、设计模式和技术考量，使我们的 Zava Retail 分析解决方案具有稳健性、安全性和可扩展性。
您将了解每一层如何交互、为何选择特定技术，以及如何将这些模式应用到您自己的 MCP 实现中。
学习目标
完成本实验后，您将能够：
- 分析 MCP 服务器的分层架构及其数据库集成
- 理解每个架构组件的角色和职责
- 设计支持多租户 MCP 应用的数据库模式
- 实现连接池和资源管理策略
- 应用生产系统的错误处理和日志记录模式
- 评估不同架构方法之间的权衡
️ MCP 服务器架构层
我们的 MCP 服务器采用了分层架构，以分离关注点并促进可维护性：
第一层：协议层 (FastMCP)
职责：处理 MCP 协议通信和消息路由
# FastMCP server setup from fastmcp import FastMCP mcp = FastMCP("Zava Retail Analytics") # Tool registration with type safety @mcp.tool() async def execute_sales_query( ctx: Context, postgresql_query: Annotated[str, Field(description="Well-formed PostgreSQL query")] ) -> str: """Execute PostgreSQL queries with Row Level Security.""" return await query_executor.execute(postgresql_query, ctx)
主要特点：
- 协议合规性：完全支持 MCP 规范
- 类型安全：使用 Pydantic 模型进行请求/响应验证
- 异步支持：非阻塞 I/O 提供高并发能力
- 错误处理：标准化错误响应
第二层：业务逻辑层
职责：实现业务规则并协调协议层与数据层之间的交互
class SalesAnalyticsService: """Business logic for retail analytics operations.""" async def get_store_performance( self, store_id: str, time_period: str ) -> Dict[str, Any]: """Calculate store performance metrics.""" # Validate business rules if not self._validate_store_access(store_id): raise UnauthorizedError("Access denied for store") # Coordinate data retrieval sales_data = await self.db_provider.get_sales_data(store_id, time_period) metrics = self._calculate_metrics(sales_data) return { "store_id": store_id, "period": time_period, "metrics": metrics, "insights": self._generate_insights(metrics) }
主要特点：
- 业务规则执行：验证存储访问和数据完整性
- 服务协调：在数据库和 AI 服务之间进行编排
- 数据转换：将原始数据转化为业务洞察
- 缓存策略：优化频繁查询的性能
第三层：数据访问层
职责：管理数据库连接、查询执行和数据映射
class PostgreSQLProvider: """Data access layer for PostgreSQL operations.""" def __init__(self, connection_config: Dict[str, Any]): self.connection_pool: Optional[Pool] = None self.config = connection_config async def execute_query( self, query: str, rls_user_id: str ) -> List[Dict[str, Any]]: """Execute query with RLS context.""" async with self.connection_pool.acquire() as conn: # Set RLS context await conn.execute( "SELECT set_config('app.current_rls_user_id', $1, false)", rls_user_id ) # Execute query with timeout try: rows = await asyncio.wait_for( conn.fetch(query), timeout=30.0 ) return [dict(row) for row in rows] except asyncio.TimeoutError: raise QueryTimeoutError("Query execution exceeded timeout")
主要特点：
- 连接池：高效的资源管理
- 事务管理：支持 ACID 合规性和回滚处理
- 查询优化：性能监控和优化
- RLS 集成：行级安全上下文管理
第四层：基础设施层
职责：处理日志记录、监控和配置等跨领域问题
class InfrastructureManager: """Infrastructure concerns management.""" def __init__(self): self.logger = self._setup_logging() self.metrics = self._setup_metrics() self.config = self._load_configuration() def _setup_logging(self) -> Logger: """Configure structured logging.""" logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.StreamHandler(), logging.FileHandler('mcp_server.log') ] ) return logging.getLogger(__name__) async def track_query_execution( self, query_type: str, duration: float, success: bool ): """Track query performance metrics.""" self.metrics.counter('query_total').labels( type=query_type, status='success' if success else 'error' ).inc() self.metrics.histogram('query_duration').labels( type=query_type ).observe(duration)
️ 数据库设计模式
我们的 PostgreSQL 模式实现了多租户 MCP 应用的几个关键设计模式：
1. 多租户模式设计
-- Core retail entities with store-based partitioning CREATE TABLE retail.stores ( store_id UUID PRIMARY KEY DEFAULT gen_random_uuid(), name VARCHAR(100) NOT NULL, location VARCHAR(200) NOT NULL, manager_id UUID NOT NULL, created_at TIMESTAMP DEFAULT NOW() ); CREATE TABLE retail.customers ( customer_id UUID PRIMARY KEY DEFAULT gen_random_uuid(), store_id UUID REFERENCES retail.stores(store_id), first_name VARCHAR(50) NOT NULL, last_name VARCHAR(50) NOT NULL, email VARCHAR(100) UNIQUE, created_at TIMESTAMP DEFAULT NOW() ); CREATE TABLE retail.orders ( order_id UUID PRIMARY KEY DEFAULT gen_random_uuid(), customer_id UUID REFERENCES retail.customers(customer_id), store_id UUID REFERENCES retail.stores(store_id), order_date TIMESTAMP DEFAULT NOW(), total_amount DECIMAL(10,2) NOT NULL, status VARCHAR(20) DEFAULT 'pending' );
设计原则：
- 外键一致性：确保跨表数据完整性
- 存储 ID 传播：每个事务表都包含 store_id
- UUID 主键：分布式系统的全局唯一标识符
- 时间戳跟踪：记录所有数据更改的审计轨迹
2. 行级安全 (RLS) 实现
-- Enable RLS on multi-tenant tables ALTER TABLE retail.customers ENABLE ROW LEVEL SECURITY; ALTER TABLE retail.orders ENABLE ROW LEVEL SECURITY; ALTER TABLE retail.order_items ENABLE ROW LEVEL SECURITY; -- Store manager can only see their store's data CREATE POLICY store_manager_customers ON retail.customers FOR ALL TO store_managers USING (store_id = get_current_user_store()); CREATE POLICY store_manager_orders ON retail.orders FOR ALL TO store_managers USING (store_id = get_current_user_store()); -- Regional managers see multiple stores CREATE POLICY regional_manager_orders ON retail.orders FOR ALL TO regional_managers USING (store_id = ANY(get_user_store_list())); -- Support function for RLS context CREATE OR REPLACE FUNCTION get_current_user_store() RETURNS UUID AS $$ BEGIN RETURN current_setting('app.current_rls_user_id')::UUID; EXCEPTION WHEN OTHERS THEN RETURN '00000000-0000-0000-0000-000000000000'::UUID; END; $$ LANGUAGE plpgsql SECURITY DEFINER;
RLS 优势：
- 自动过滤：数据库强制数据隔离
- 应用简化：无需复杂的 WHERE 子句
- 默认安全性：避免意外访问错误数据
- 审计合规：明确的数据访问边界
3. 向量搜索模式
-- Product embeddings for semantic search CREATE TABLE retail.product_description_embeddings ( product_id UUID PRIMARY KEY REFERENCES retail.products(product_id), description_embedding vector(1536), last_updated TIMESTAMP DEFAULT NOW() ); -- Optimize vector similarity search CREATE INDEX idx_product_embeddings_vector ON retail.product_description_embeddings USING ivfflat (description_embedding vector_cosine_ops); -- Semantic search function CREATE OR REPLACE FUNCTION search_products_by_description( query_embedding vector(1536), similarity_threshold FLOAT DEFAULT 0.7, max_results INTEGER DEFAULT 20 ) RETURNS TABLE( product_id UUID, name VARCHAR, description TEXT, similarity_score FLOAT ) AS $$ BEGIN RETURN QUERY SELECT p.product_id, p.name, p.description, (1 - (pde.description_embedding <=> query_embedding)) AS similarity_score FROM retail.products p JOIN retail.product_description_embeddings pde ON p.product_id = pde.product_id WHERE (pde.description_embedding <=> query_embedding) <= (1 - similarity_threshold) ORDER BY similarity_score DESC LIMIT max_results; END; $$ LANGUAGE plpgsql;
连接管理模式
高效的数据库连接管理对 MCP 服务器性能至关重要：
连接池配置
class ConnectionPoolManager: """Manages PostgreSQL connection pools.""" async def create_pool(self) -> Pool: """Create optimized connection pool.""" return await asyncpg.create_pool( host=self.config.db_host, port=self.config.db_port, database=self.config.db_name, user=self.config.db_user, password=self.config.db_password, # Pool configuration min_size=2, # Minimum connections max_size=10, # Maximum connections max_inactive_connection_lifetime=300, # 5 minutes # Query configuration command_timeout=30, # Query timeout server_settings={ "application_name": "zava-mcp-server", "jit": "off", # Disable JIT for stability "work_mem": "4MB", # Limit work memory "statement_timeout": "30s" } ) async def execute_with_retry( self, query: str, params: Tuple = None, max_retries: int = 3 ) -> List[Dict[str, Any]]: """Execute query with automatic retry logic.""" for attempt in range(max_retries): try: async with self.pool.acquire() as conn: if params: rows = await conn.fetch(query, *params) else: rows = await conn.fetch(query) return [dict(row) for row in rows] except (ConnectionError, InterfaceError) as e: if attempt == max_retries - 1: raise # Exponential backoff await asyncio.sleep(2 ** attempt) logger.warning(f"Database connection failed, retrying ({attempt + 1}/{max_retries})")
资源生命周期管理
class MCPServerManager: """Manages MCP server lifecycle and resources.""" async def startup(self): """Initialize server resources.""" # Create database connection pool self.db_pool = await self.pool_manager.create_pool() # Initialize AI services self.ai_client = await self.create_ai_client() # Setup monitoring self.metrics_collector = MetricsCollector() logger.info("MCP server startup complete") async def shutdown(self): """Cleanup server resources.""" try: # Close database connections if self.db_pool: await self.db_pool.close() # Cleanup AI client if self.ai_client: await self.ai_client.close() # Flush metrics await self.metrics_collector.flush() logger.info("MCP server shutdown complete") except Exception as e: logger.error(f"Error during shutdown: {e}") async def health_check(self) -> Dict[str, str]: """Verify server health status.""" status = {} # Check database connection try: async with self.db_pool.acquire() as conn: await conn.fetchval("SELECT 1") status["database"] = "healthy" except Exception as e: status["database"] = f"unhealthy: {e}" # Check AI service try: await self.ai_client.health_check() status["ai_service"] = "healthy" except Exception as e: status["ai_service"] = f"unhealthy: {e}" return status
️ 错误处理和弹性模式
强大的错误处理确保 MCP 服务器的可靠运行：
分层错误类型
class MCPError(Exception): """Base MCP server error.""" def __init__(self, message: str, error_code: str = "MCP_ERROR"): self.message = message self.error_code = error_code super().__init__(message) class DatabaseError(MCPError): """Database operation errors.""" def __init__(self, message: str, query: str = None): super().__init__(message, "DATABASE_ERROR") self.query = query class AuthorizationError(MCPError): """Access control errors.""" def __init__(self, message: str, user_id: str = None): super().__init__(message, "AUTHORIZATION_ERROR") self.user_id = user_id class QueryTimeoutError(DatabaseError): """Query execution timeout.""" def __init__(self, query: str): super().__init__(f"Query timeout: {query[:100]}...", query) self.error_code = "QUERY_TIMEOUT" class ValidationError(MCPError): """Input validation errors.""" def __init__(self, field: str, value: Any, constraint: str): message = f"Validation failed for {field}: {constraint}" super().__init__(message, "VALIDATION_ERROR") self.field = field self.value = value
错误处理中间件
@contextmanager async def error_handling_context(operation_name: str, user_id: str = None): """Centralized error handling for operations.""" start_time = time.time() try: yield # Success metrics duration = time.time() - start_time metrics.operation_success.labels(operation=operation_name).inc() metrics.operation_duration.labels(operation=operation_name).observe(duration) except ValidationError as e: logger.warning(f"Validation error in {operation_name}: {e.message}", extra={ "operation": operation_name, "user_id": user_id, "error_type": "validation", "field": e.field }) metrics.operation_error.labels(operation=operation_name, type="validation").inc() raise except AuthorizationError as e: logger.warning(f"Authorization error in {operation_name}: {e.message}", extra={ "operation": operation_name, "user_id": user_id, "error_type": "authorization" }) metrics.operation_error.labels(operation=operation_name, type="authorization").inc() raise except DatabaseError as e: logger.error(f"Database error in {operation_name}: {e.message}", extra={ "operation": operation_name, "user_id": user_id, "error_type": "database", "query": e.query[:100] if e.query else None }) metrics.operation_error.labels(operation=operation_name, type="database").inc() raise except Exception as e: logger.error(f"Unexpected error in {operation_name}: {str(e)}", extra={ "operation": operation_name, "user_id": user_id, "error_type": "unexpected" }, exc_info=True) metrics.operation_error.labels(operation=operation_name, type="unexpected").inc() raise MCPError(f"Internal server error in {operation_name}")
性能优化策略
查询性能监控
class QueryPerformanceMonitor: """Monitor and optimize query performance.""" def __init__(self): self.slow_query_threshold = 1.0 # seconds self.query_stats = defaultdict(list) @contextmanager async def monitor_query(self, query: str, operation_type: str = "unknown"): """Monitor query execution time and performance.""" start_time = time.time() query_hash = hashlib.md5(query.encode()).hexdigest()[:8] try: yield duration = time.time() - start_time # Record performance metrics self.query_stats[operation_type].append(duration) # Log slow queries if duration > self.slow_query_threshold: logger.warning(f"Slow query detected", extra={ "query_hash": query_hash, "duration": duration, "operation_type": operation_type, "query": query[:200] }) # Update metrics metrics.query_duration.labels(type=operation_type).observe(duration) except Exception as e: duration = time.time() - start_time logger.error(f"Query failed", extra={ "query_hash": query_hash, "duration": duration, "operation_type": operation_type, "error": str(e) }) raise def get_performance_summary(self) -> Dict[str, Any]: """Generate performance summary report.""" summary = {} for operation_type, durations in self.query_stats.items(): if durations: summary[operation_type] = { "count": len(durations), "avg_duration": sum(durations) / len(durations), "max_duration": max(durations), "min_duration": min(durations), "slow_queries": len([d for d in durations if d > self.slow_query_threshold]) } return summary
缓存策略
class QueryCache: """Intelligent query result caching.""" def __init__(self, redis_url: str = None): self.cache = {} # In-memory fallback self.redis_client = redis.Redis.from_url(redis_url) if redis_url else None self.cache_ttl = 300 # 5 minutes default async def get_cached_result( self, cache_key: str, query_func: Callable, ttl: int = None ) -> Any: """Get result from cache or execute query.""" ttl = ttl or self.cache_ttl # Try cache first cached_result = await self._get_from_cache(cache_key) if cached_result is not None: metrics.cache_hit.labels(type="query").inc() return cached_result # Execute query metrics.cache_miss.labels(type="query").inc() result = await query_func() # Cache result await self._set_in_cache(cache_key, result, ttl) return result def _generate_cache_key(self, query: str, user_context: str) -> str: """Generate consistent cache key.""" key_data = f"{query}:{user_context}" return hashlib.sha256(key_data.encode()).hexdigest()
关键要点
完成本实验后，您应该了解：
✅ 分层架构：如何在 MCP 服务器设计中分离关注点
✅ 数据库模式：多租户模式设计和 RLS 实现
✅ 连接管理：高效的连接池和资源生命周期管理
✅ 错误处理：分层错误类型和弹性模式
✅ 性能优化：监控、缓存和查询优化
✅ 生产准备：基础设施问题和操作模式
下一步
继续学习 实验 02：安全性和多租户，深入了解：
- 行级安全实现细节
- 身份验证和授权模式
- 多租户数据隔离策略
- 安全审计和合规性考量
额外资源
架构模式
- Python 中的清晰架构 - Python 应用的架构模式
- 数据库设计模式 - 关系数据库设计原则
- 微服务模式 - 服务架构模式
PostgreSQL 高级主题
- PostgreSQL 性能调优 - 数据库优化指南
- 连接池最佳实践 - 连接管理
- 查询规划和优化 - 查询性能
Python 异步模式
- AsyncIO 最佳实践 - 异步编程模式
- FastAPI 架构 - 现代 Python Web 架构
- Pydantic 模型 - 数据验证和序列化
下一步：准备探索安全模式？继续学习 实验 02：安全性和多租户
免责声明：
本文档使用AI翻译服务 Co-op Translator 进行翻译。尽管我们努力确保翻译的准确性，但请注意，自动翻译可能包含错误或不准确之处。原始语言的文档应被视为权威来源。对于关键信息，建议使用专业人工翻译。我们对因使用此翻译而产生的任何误解或误读不承担责任。
评论区
(0)
U