企业官网建设流程全解析

• 教程
• 文档
• 人工智能

导读

本篇文章基于 11-MCPServerHandsOnLabs 学习路径中的 Lab 01「Core Architecture Concepts」，系统拆解一个数据库集成的 MCP Server 应如何做架构分层、数据库建模与生产化设计。你将掌握 FastMCP 协议层、业务逻辑层、数据访问层与基础设施层的职责划分，学会多租户 Schema、Row Level Security（RLS）、pgvector 语义搜索、连接池与资源生命周期、分层错误处理及查询性能监控的完整落地方案，并能在自己的 MCP 项目中直接复用这些模式。

本 Lab 在学习路径中的定位

该 Lab 属于「MCP Server with PostgreSQL」实战学习路径的第二个环节。整个路径以虚构连锁零售企业Zava Retail Analytics为案例（详见 00-Introduction：8 家实体门店 + 1 家线上门店，覆盖门店经理、区域经理、高管等不同权限角色），目标是构建一个「AI 助手 ↔ PostgreSQL 数据库」之间的安全、可扩展桥梁：

• 面向 AI 助手暴露标准化的 MCP 工具（Schema 查询、SQL 执行、语义搜索）；
• 通过RLS保证门店级数据隔离；
• 通过pgvector提供产品语义搜索能力；
• 通过连接池、超时、重试、监控保证生产可用性。

Lab 01 聚焦的是「架构蓝图」：不急着写业务功能，而是先把四层架构、数据模型、连接管理、错误处理、性能优化这些骨架立起来。后续 Lab 02（安全与多租户）、Lab 04（数据库 Schema）、Lab 05（MCP Server 实现）会逐一落代码，本 Lab 则是这些实现的设计依据。

学习目标

完成本 Lab 后，你将能够：

• 分析带数据库集成的 MCP Server 的分层架构；
• 理解每个架构组件的角色与职责边界；
• 设计支持多租户 MCP 应用的数据库 Schema；
• 实现连接池与资源管理策略；
• 应用面向生产系统的错误处理与日志模式；
• 评估不同架构方案之间的权衡。

MCP Server 分层架构

本 Lab 给出的 MCP Server 采用分层架构（Layered Architecture）：把「协议通信」「业务规则」「数据访问」「横切关注点」四个职责分离，每一层只关心自己该管的事，从而提升可维护性与可测试性。这与后续 05-MCP-Server 中的实际工程结构（config.py、sales_analysis_postgres.py、sales_analysis.py、health_check.py）一一对应。

Layer 1：协议层（FastMCP）

职责：处理 MCP 协议通信与消息路由，负责将 AI 助手的工具调用请求转化为安全的服务调用。

# 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 模型对请求/响应做校验（Annotated[str, Field(...)]即为声明式参数约束）；
• 异步支持：非阻塞 I/O，支撑高并发；
• 错误处理：标准化错误响应。

在 05-MCP-Server 的实际实现 中，协议层具体化为三个带类型注解的工具：get_multiple_table_schemas（多表 Schema 检索）、execute_sales_query（带 RLS 的 SQL 执行）、get_current_utc_date（取 UTC 时间，供时间敏感查询使用），工具名即对 AI 模型的「能力描述」，参数注解即「使用契约」。可以推断，协议层在本项目中的作用是把「AI 的自然语言意图」收敛为「白名单内的、类型受控的数据库操作」。

Layer 2：业务逻辑层

职责：实现业务规则，并在协议层与数据层之间做协调编排。

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 服务之间的调用；
• 数据转换：把原始数据转换为业务洞察；
• 缓存策略：对高频查询做性能优化。

Layer 3：数据访问层

职责：管理数据库连接、查询执行与数据映射，是「RLS 上下文注入」的关键位置。

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 集成：行级安全上下文管理。

需要说明的是：本 Lab 架构图中的 RLS 上下文参数名为app.current_rls_user_id，而在 04-Database 的实际落地 中，Zava 零售案例最终采用app.current_store_id（门店维度隔离，见下文「数据库设计模式」），两种命名是同一套「先 set_config 注入上下文、再由 RLS 策略自动过滤」机制的体现。实际实现的PostgreSQLSchemaProvider.set_rls_context会先在连接上执行SELECT set_config(...)，再执行业务查询，RLS 策略随后自动生效。

Layer 4：基础设施层

职责：处理日志、监控、配置等横切关注点（Cross-cutting Concerns）。

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)

关键特性：

• 结构化日志（控制台 + 文件双通道）；
• 指标埋点（Counter 统计总量、Histogram 统计耗时分布）；
• 集中式配置加载。

在 05-MCP-Server 的实际实现 中，配置层进一步细化为DatabaseConfig、AzureConfig、ServerConfig三个 dataclass，全部从环境变量读取并带默认值（例如POSTGRES_HOST默认localhost、POSTGRES_PORT默认5432、连接池min_connections=2、max_connections=10、command_timeout=30），并统一映射为 asyncpg 连接参数（application_name='zava-mcp-server'、jit='off'、work_mem='4MB'、statement_timeout跟随超时配置）。这意味着基础设施层的「配置」与「日志」在工程上是可复用、可替换的模块。

数据库设计模式

Zava 案例的 PostgreSQL Schema 采用Shared Database + Shared Schema多租户模型：所有租户共享同一套表结构，靠 RLS 策略做数据隔离。本 Lab 给出三类核心模式：多租户 Schema、RLS 实现、向量搜索 Schema。

模式 1：多租户 Schema 设计

-- 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' );

设计原则：

• 外键一致性：跨表数据完整性；
• store_id 贯穿：每个事务表都携带store_id，这是后续 RLS 过滤的锚点；
• UUID 主键：分布式系统下的全局唯一标识；
• 时间戳追踪：所有数据变更的审计轨迹。

在 04-Database 的实际 Schema 中，该模式被进一步落地并细化：

• retail.stores以VARCHAR(50)业务键（seattle、redmond、bellevue、online）做主键，而不是 UUID，便于上下文注入与可读性；
• 交易实体在案例中命名为sales_transactions/sales_transaction_items（本 Lab 架构图中为orders/order_items，属同一角色的简化示意）；
• products表用UNIQUE (store_id, sku)保证「同店 SKU 唯一」，并叠加 GIN 索引支持tags数组与metadataJSONB 的灵活检索；
• 每张表配套一组针对性索引（如idx_products_price、idx_customers_loyalty_tier、部分索引WHERE is_active = TRUE），覆盖典型报表查询路径。

模式 2：Row Level Security 实现

-- 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；
• 应用简化：业务代码无需维护复杂的过滤条件；
• 默认安全：即使写错查询也不可能误读到无权数据；
• 审计合规：数据访问边界清晰可见。

在 02-Security 与 04-Database 中，这套 RLS 机制被完整工程化：

1. 独立应用角色mcp_user，仅授予retailSchema 的USAGE与表级SELECT/INSERT/UPDATE/DELETE，并用ALTER DEFAULT PRIVILEGES覆盖未来新建表；
2. 通过retail.set_store_context(store_id)（SECURITY DEFINER）校验门店存在且is_active，再set_config('app.current_store_id', ...)写入上下文，同时写入审计日志；
3. 每张多租户表一个策略，如customers_store_isolation、products_store_isolation、sales_transactions_store_isolation，统一USING (store_id = current_setting('app.current_store_id', true))；
4. 明细表sales_transaction_items本身没有store_id，策略通过子查询关联父表sales_transactions实现「经联接继承租户」的隔离。

这套组合意味着：AI 生成的任何 SQL，只要连接上下文正确，不可能越界访问其他门店数据——这正是「安全默认（Security by Default）」在真实工程中的形态。

模式 3：向量搜索 Schema

-- 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;

要点：vector(1536)对应 OpenAItext-embedding-3-small的向量维度（在 04-Database 的实际实现 中明确标注embedding_model VARCHAR(100) NOT NULL DEFAULT 'text-embedding-3-small'，并用UNIQUE (product_id, embedding_model)保证每个产品每个模型只有一条嵌入）；余弦距离用<=>算子表达，1 - 距离即相似度；阈值默认0.7、最多返回20条。

本 Lab 架构图选用IVFFlat索引，而实际案例落地时改用HNSW（USING hnsw (embedding vector_cosine_ops)，详见 04-Database 的向量索引）。从源码结构看，二者都是 pgvector 提供的近似最近邻（ANN）索引，区别在于 HNSW 构建质量更高、查询更稳，IVFFlat 更省内存——选型时应结合数据量与召回率要求权衡。实际案例的搜索函数retail.search_products_by_similarity还叠加了is_active = TRUE与store_id = current_setting('app.current_store_id', true)过滤，确保语义搜索同样受 RLS 约束。

连接管理与资源生命周期

对 MCP Server 而言，每次工具调用都可能触发数据库查询，连接管理直接决定吞吐与稳定性。

连接池配置

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})")

参数含义与建议取值范围：

在 05-MCP-Server 的实现 中，连接池由PostgreSQLSchemaProvider.create_pool()创建、close_pool()销毁；每次查询通过get_connection()上下文管理器从池中获取连接，并在执行前调用set_rls_context()注入租户上下文；查询外层套asyncio.wait_for(..., timeout=config.database.command_timeout)，超时抛出带明确文案的异常。结果集默认截断为max_rows=20行并格式化（NULL、datetime、JSON 字段均有专门处理），避免 AI 上下文被超大结果集撑爆——这是数据库类 MCP 工具一个容易被忽略但极其重要的设计点。

资源生命周期管理

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

要点：启动时按「连接池 → AI 客户端 → 监控采集器」的依赖顺序初始化；关闭时逆序且逐个 try/except，保证单项清理失败不阻断其余资源释放；健康检查拆分为数据库连通性（SELECT 1+ 池状态）与 AI 服务可用性两个独立维度。

工程落地时（见 05-MCP-Server 的健康检查实现），FastMCP 应用通过asynccontextmanager的lifespan钩子承载启动/关闭逻辑：启动时create_pool()并跑一次数据库健康检查，不健康则直接拒绝启动；关闭时close_pool()。同时暴露四类 HTTP 端点：

• GET /health：基础存活响应（含服务名与 UTC 时间戳）；
• GET /health/detailed：数据库连通性与连接池min_size / max_size / current_size / idle_size明细，任一组件异常整体返回 503；
• GET /health/ready：就绪探针（数据库可用才返回 200），供 Kubernetes 就绪检查使用；
• GET /health/live：存活探针，始终 200。

错误处理与韧性模式

健壮的错误处理是 MCP Server 可靠运行的前提。本 Lab 给出「分层错误类型 + 集中式处理上下文」的组合模式。

分层错误类型

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

设计要点：

• 所有错误继承统一的MCPError基类，携带error_code，便于 AI 客户端与运维系统识别；
• DatabaseError附带原始query（截断 100 字符），AuthorizationError附带user_id，ValidationError附带字段与约束——错误对象里带足上下文，但绝不把敏感值写进对外消息；
• QueryTimeoutError继承DatabaseError并覆写error_code，形成可细分的错误树。

集中式错误处理上下文

@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}")

要点：一个上下文管理器统一处理「记日志、打指标、按类型分类、决定是否包装后再抛」。其中：

• Validation / Authorization属于可预期问题，用warning级别，不暴露栈；
• DatabaseError用error级别，附带截断后的查询文本便于排查；
• 未知异常打印exc_info完整栈用于诊断，但对上层只暴露脱敏后的MCPError，防止内部细节泄漏给 AI 助手。

这种「一处在入口收口、按类型分诊」的模式，与 05-MCP-Server 实现 中工具函数「try/except 后返回用户可读字符串」的落地方式互补：服务层抛类型化异常，工具层捕获后转为f"Error ...: {e!s}"形式的友好提示返回给模型。

性能优化策略

查询性能监控

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

要点：以 1 秒为慢查询阈值，对每条查询记录query_hash（MD5 前 8 位）、耗时、操作类型，慢查询与失败查询分别走 warning / error 日志通道，并可汇总出 count、avg、max、min、慢查询数等统计报告。与之配套的数据库侧手段在 04-Database 的优化章节 中有完整脚本：retail.slow_queries视图（基于pg_stat_statements过滤平均耗时 >100ms 的查询）、retail.table_stats与retail.index_usage监控视图，以及retail.perform_maintenance()例行维护函数（ANALYZE、VACUUM、REINDEX CONCURRENTLY）。推荐将log_min_duration_statement = 1000写入postgresql.conf，与应用层慢查询日志形成对照。

缓存策略

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()

要点：

• 缓存键必须同时包含SQL 与租户上下文（query:user_context做 SHA-256），否则会把 A 门店的数据错发给 B 门店；
• 默认 TTL 300 秒（5 分钟），适合「报表类、聚合类」变化不频繁的查询；
• 内存缓存仅作兜底，生产建议切换到 Redis（redis_url存在时优先使用）以支持多实例共享；
• 命中率通过cache_hit/cache_miss指标观察，为调优提供依据。

关键要点

完成本 Lab 后，你应当理解：

• 分层架构：MCP Server 设计中如何分离关注点——协议层管通信、业务层管规则、数据层管访问、基础设施层管横切；
• 数据库模式：多租户 Schema 设计（store_id贯穿）与 RLS 实现（set_config注入上下文 + 策略自动过滤）；
• 连接管理：连接池参数（min_size/max_size/command_timeout/server_settings）与「启动建池、关闭释放、健康检查」的资源生命周期；
• 错误处理：分层错误类型与集中式处理上下文，做到「分类记录、脱敏上抛」；
• 性能优化：慢查询监控、性能汇总、基于「SQL + 租户上下文」键的查询缓存；
• 生产就绪：日志、指标、健康端点、维护任务等基础设施关注点。

下一步

继续 Lab 02: Security and Multi-Tenancy，深入掌握：

• Row Level Security 的实现细节（角色权限、上下文管理函数、审计日志）；
• 认证与授权模式（Azure 身份服务与防御纵深架构）；
• 多租户数据隔离策略；
• 安全审计与合规考量。

说明：本学习路径中的示例基于 MCP2025-11-25依赖（HTTP/SSE 与初始化方式，见 00-Introduction 的版本说明）；新建实现建议参考2026-07-28规范的无状态请求与 Streamable HTTP 方式。架构层面的分层与数据库模式在本仓库中均不受协议版本影响，可放心复用。

• 教程
• 文档
• 人工智能