MCP Server 核心架构实战：FastMCP 分层架构、PostgreSQL 多租户设计与生产级韧性模式解析

教程文档人工智能【免费下载链接】mcp-for-beginnersThis open-source curriculum introduces the fundamentals of Model Context Protocol (MCP) through real-world, cross-language examples in .NET, Java, TypeScript, JavaScript, Rust and Python. Designed for developers, it focuses on practical techniques for building modular, scalable, and secure AI workflows from session setup to service orchestration.项目地址https://gitcode.com/GitHub_Trending/mc/mcp-for-beginners点击查看免费下载导读本篇文章基于 11-MCPServerHandsOnLabs 学习路径中的 Lab 01「Core Architecture Concepts」系统拆解一个数据库集成的 MCP Server 应如何做架构分层、数据库建模与生产化设计。你将掌握 FastMCP 协议层、业务逻辑层、数据访问层与基础设施层的职责划分学会多租户 Schema、Row Level SecurityRLS、pgvector 语义搜索、连接池与资源生命周期、分层错误处理及查询性能监控的完整落地方案并能在自己的 MCP 项目中直接复用这些模式。本 Lab 在学习路径中的定位该 Lab 属于「MCP Server with PostgreSQL」实战学习路径的第二个环节。整个路径以虚构连锁零售企业Zava Retail Analytics为案例详见 00-Introduction8 家实体门店 1 家线上门店覆盖门店经理、区域经理、高管等不同权限角色目标是构建一个「AI 助手 ↔ PostgreSQL 数据库」之间的安全、可扩展桥梁面向 AI 助手暴露标准化的 MCP 工具Schema 查询、SQL 执行、语义搜索通过RLS保证门店级数据隔离通过pgvector提供产品语义搜索能力通过连接池、超时、重试、监控保证生产可用性。Lab 01 聚焦的是「架构蓝图」不急着写业务功能而是先把四层架构、数据模型、连接管理、错误处理、性能优化这些骨架立起来。后续 Lab 02安全与多租户、Lab 04数据库 Schema、Lab 05MCP 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(descriptionWell-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), timeout30.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( levellogging.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( typequery_type, statussuccess if success else error ).inc() self.metrics.histogram(query_duration).labels( typequery_type ).observe(duration)关键特性结构化日志控制台 文件双通道指标埋点Counter 统计总量、Histogram 统计耗时分布集中式配置加载。在 05-MCP-Server 的实际实现 中配置层进一步细化为DatabaseConfig、AzureConfig、ServerConfig三个 dataclass全部从环境变量读取并带默认值例如POSTGRES_HOST默认localhost、POSTGRES_PORT默认5432、连接池min_connections2、max_connections10、command_timeout30并统一映射为 asyncpg 连接参数application_namezava-mcp-server、jitoff、work_mem4MB、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覆盖典型报表查询路径。模式 2Row 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 stores 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 机制被完整工程化独立应用角色mcp_user仅授予retailSchema 的USAGE与表级SELECT/INSERT/UPDATE/DELETE并用ALTER DEFAULT PRIVILEGES覆盖未来新建表通过retail.set_store_context(store_id)SECURITY DEFINER校验门店存在且is_active再set_config(app.current_store_id, ...)写入上下文同时写入审计日志每张多租户表一个策略如customers_store_isolation、products_store_isolation、sales_transactions_store_isolation统一USING (store_id current_setting(app.current_store_id, true))明细表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索引而实际案例落地时改用HNSWUSING 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( hostself.config.db_host, portself.config.db_port, databaseself.config.db_name, userself.config.db_user, passwordself.config.db_password, # Pool configuration min_size2, # Minimum connections max_size10, # Maximum connections max_inactive_connection_lifetime300, # 5 minutes # Query configuration command_timeout30, # 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(fDatabase connection failed, retrying ({attempt 1}/{max_retries}))参数含义与建议取值范围参数示例值说明min_size2池内常驻最小连接数过低会导致突发流量时建连抖动max_size10池上限需结合max_connections与并发模型调整max_inactive_connection_lifetime300秒空闲连接回收时间默认 5 分钟防止连接被服务端静默断开command_timeout30秒单条命令超时防止慢查询拖垮服务server_settings.jitoff关闭 JIT换取短查询的执行稳定性server_settings.work_mem4MB限制单次排序/哈希内存避免 OOMserver_settings.statement_timeout30s数据库侧兜底超时与应用侧command_timeout双保险重试次数max_retries3仅对连接类错误重试指数退避2^attempt秒在 05-MCP-Server 的实现 中连接池由PostgreSQLSchemaProvider.create_pool()创建、close_pool()销毁每次查询通过get_connection()上下文管理器从池中获取连接并在执行前调用set_rls_context()注入租户上下文查询外层套asyncio.wait_for(..., timeoutconfig.database.command_timeout)超时抛出带明确文案的异常。结果集默认截断为max_rows20行并格式化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(fError 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] funhealthy: {e} # Check AI service try: await self.ai_client.health_check() status[ai_service] healthy except Exception as e: status[ai_service] funhealthy: {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明细任一组件异常整体返回 503GET /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__(fQuery 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 fValidation failed for {field}: {constraint} super().__init__(message, VALIDATION_ERROR) self.field field self.value value设计要点所有错误继承统一的MCPError基类携带error_code便于 AI 客户端与运维系统识别DatabaseError附带原始query截断 100 字符AuthorizationError附带user_idValidationError附带字段与约束——错误对象里带足上下文但绝不把敏感值写进对外消息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(operationoperation_name).inc() metrics.operation_duration.labels(operationoperation_name).observe(duration) except ValidationError as e: logger.warning(fValidation error in {operation_name}: {e.message}, extra{ operation: operation_name, user_id: user_id, error_type: validation, field: e.field }) metrics.operation_error.labels(operationoperation_name, typevalidation).inc() raise except AuthorizationError as e: logger.warning(fAuthorization error in {operation_name}: {e.message}, extra{ operation: operation_name, user_id: user_id, error_type: authorization }) metrics.operation_error.labels(operationoperation_name, typeauthorization).inc() raise except DatabaseError as e: logger.error(fDatabase 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(operationoperation_name, typedatabase).inc() raise except Exception as e: logger.error(fUnexpected error in {operation_name}: {str(e)}, extra{ operation: operation_name, user_id: user_id, error_type: unexpected }, exc_infoTrue) metrics.operation_error.labels(operationoperation_name, typeunexpected).inc() raise MCPError(fInternal server error in {operation_name})要点一个上下文管理器统一处理「记日志、打指标、按类型分类、决定是否包装后再抛」。其中Validation / Authorization属于可预期问题用warning级别不暴露栈DatabaseError用error级别附带截断后的查询文本便于排查未知异常打印exc_info完整栈用于诊断但对上层只暴露脱敏后的MCPError防止内部细节泄漏给 AI 助手。这种「一处在入口收口、按类型分诊」的模式与 05-MCP-Server 实现 中工具函数「try/except 后返回用户可读字符串」的落地方式互补服务层抛类型化异常工具层捕获后转为fError ...: {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(fSlow query detected, extra{ query_hash: query_hash, duration: duration, operation_type: operation_type, query: query[:200] }) # Update metrics metrics.query_duration.labels(typeoperation_type).observe(duration) except Exception as e: duration time.time() - start_time logger.error(fQuery 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_hashMD5 前 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(typequery).inc() return cached_result # Execute query metrics.cache_miss.labels(typequery).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 分钟适合「报表类、聚合类」变化不频繁的查询内存缓存仅作兜底生产建议切换到 Redisredis_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 方式。架构层面的分层与数据库模式在本仓库中均不受协议版本影响可放心复用。赞分享教程文档人工智能【免费下载链接】mcp-for-beginnersThis open-source curriculum introduces the fundamentals of Model Context Protocol (MCP) through real-world, cross-language examples in .NET, Java, TypeScript, JavaScript, Rust and Python. Designed for developers, it focuses on practical techniques for building modular, scalable, and secure AI workflows from session setup to service orchestration.项目地址https://gitcode.com/GitHub_Trending/mc/mcp-for-beginners点击查看免费下载相关推荐Apache Pulsar 核心概念全景解析消息模型、分层架构与多租户设计Apache Pulsar 核心概念全景解析消息模型、分层架构与多租户设计 本文以 Apache Pulsar 官方概念总览concepts overvie消息队列后端流处理漫画翻译革命manga-image-translator如何让日漫阅读效率提升10倍漫画翻译革命manga image translator如何让日漫阅读效率提升10倍 还在为看不懂日语漫画而烦恼吗还在手动截图、翻译、PS的繁琐流程中挣扎吗人工智能AI 应用计算机视觉图像处理OCR最完整SaaS设计指南多租户架构模式与实战解析最完整SaaS设计指南多租户架构模式与实战解析 你是否正在为SaaS系统设计而头疼用户数据隔离、资源共享、成本控制这些问题是否让你难以平衡本文将从实战角度文档技术博客上一篇企业级AI对话前端部署指南5步构建安全高效的SillyTavern系统下一篇文档生成RAG_Techniques中的自动化文档系统创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考