MCP 工具开发实战：在 MCP-Server-and-PostgreSQL 零售分析项目中构建安全、高性能的数据库查询工具

本指南聚焦 mcp-for-beginners 仓库中 11-MCPServerHandsOnLabs/06-Tools 实验的核心内容：为 Zava Retail 分析型 MCP 服务器构建一套完整的工具层——涵盖统一的工具基座架构、SQL 注入防护、销售分析、Schema 自省与商业智能分析。读完本文，你将掌握如何设计带参数校验与度量埋点的 MCP 工具基类、如何用多层校验保障自定义查询安全、如何用模板化 SQL 提供稳定的分析能力，以及如何将工具接入 MCP Server 实现实验 中定义的数据库连接层。

学习目标

完成本实验后，你将能够：

• 设计带复杂参数校验的进阶 MCP 工具
• 实现具备 SQL 注入防护的安全数据库查询工具
• 创建用于动态查询的 Schema 自省（Schema Introspection）能力
• 构建面向商业智能的定制化分析工具
• 应用全面的错误处理与优雅降级策略
• 优化工具在生成环境负载下的性能

工具层整体架构

MCP 工具是 AI 助手与数据系统之间的接口。一个设计良好的工具应在维持安全性与性能的同时，为复杂操作提供结构化、经过校验的访问。本实验围绕一个零售 MCP 服务器展开：它内置一套完整的工具集，支持以自然语言查询销售数据、产品目录与业务分析，同时保持严格的安全边界与最优性能。

工具层的设计遵循"统一基类 + 分类注册 + 统一返回结构"的模式，所有工具最终都通过 MCP Server 实现实验 中的 mcp.tool() 注册暴露给 AI 客户端，并共享同一数据库连接池 db_provider。

工具分类与统一返回结构

工具分类使用枚举 ToolCategory 组织，便于发现与归档：

class ToolCategory(Enum):
    """Tool categorization for organization and discovery."""
    DATABASE_QUERY = "database_query"
    SCHEMA_INTROSPECTION = "schema_introspection"
    ANALYTICS = "analytics"
    UTILITY = "utility"
    ADMINISTRATIVE = "administrative"

所有工具的执行结果统一封装为 ToolResult 数据类，携带执行状态、数据、错误信息、元数据、执行耗时与行数，方便上层统一处理：

@dataclass
class ToolResult:
    """Standardized tool result structure."""
    success: bool
    data: Any = None
    error: Optional[str] = None
    metadata: Optional[Dict[str, Any]] = None
    execution_time_ms: Optional[float] = None
    row_count: Optional[int] = None

BaseTool 基类：校验、度量与错误处理的统一入口

BaseTool 是所有 MCP 工具的抽象基类。它要求子类实现两个抽象方法：execute（执行工具逻辑）与 get_input_schema（返回用于输入校验的 JSON Schema）。

class BaseTool(ABC):
    """Abstract base class for all MCP tools."""

    def __init__(self, name: str, description: str, category: ToolCategory):
        self.name = name
        self.description = description
        self.category = category
        self.call_count = 0
        self.total_execution_time = 0.0

    @abstractmethod
    async def execute(self, **kwargs) -> ToolResult:
        """Execute the tool with given parameters."""
        pass

    @abstractmethod
    def get_input_schema(self) -> Dict[str, Any]:
        """Get JSON schema for tool input validation."""
        pass

关键设计在于 call 包装方法——它是所有工具执行的统一入口，顺序完成三件事：

1. 输入校验：调用 _validate_input，对照 get_input_schema 中声明的 required 与 properties 检查必填参数是否齐全；缺失时抛出 ValueError（生产环境可引入 jsonschema 库做完整的类型校验）；
2. 指标埋点与日志：记录 call_count、执行耗时 execution_time_ms 与累计耗时，并通过 _sanitize_parameters 对日志参数脱敏；
3. 错误兜底：try/except 捕获所有异常，返回 success=False 的 ToolResult，保证工具失败时 AI 客户端拿到结构化错误而非未处理异常。

async def call(self, **kwargs) -> ToolResult:
    """Wrapper for tool execution with metrics and error handling."""
    start_time = time.time()
    self.call_count += 1

    try:
        # Validate input parameters
        self._validate_input(kwargs)

        # Log tool execution
        logger.info(
            f"Executing tool: {self.name}",
            extra={
                'tool_name': self.name,
                'tool_category': self.category.value,
                'parameters': self._sanitize_parameters(kwargs)
            }
        )

        # Execute the tool
        result = await self.execute(**kwargs)

        # Record execution time
        execution_time = (time.time() - start_time) * 1000
        result.execution_time_ms = execution_time
        self.total_execution_time += execution_time

        # Log success
        logger.info(
            f"Tool execution completed: {self.name}",
            extra={
                'tool_name': self.name,
                'execution_time_ms': execution_time,
                'success': result.success,
                'row_count': result.row_count
            }
        )

        return result

    except Exception as e:
        execution_time = (time.time() - start_time) * 1000

        logger.error(
            f"Tool execution failed: {self.name}",
            extra={
                'tool_name': self.name,
                'execution_time_ms': execution_time,
                'error': str(e)
            },
            exc_info=True
        )

        return ToolResult(
            success=False,
            error=f"Tool execution failed: {str(e)}",
            execution_time_ms=execution_time
        )

日志脱敏逻辑默认屏蔽 password、token、secret、key 等敏感字段：

def _sanitize_parameters(self, kwargs: Dict[str, Any]) -> Dict[str, Any]:
    """Sanitize parameters for logging (remove sensitive data)."""
    # Remove or mask sensitive parameters
    sanitized = kwargs.copy()
    sensitive_keys = ['password', 'token', 'secret', 'key']

    for key in sanitized:
        if any(sensitive in key.lower() for sensitive in sensitive_keys):
            sanitized[key] = "***MASKED***"

    return sanitized

get_statistics 提供工具级的使用统计，可用于监控与调优：

def get_statistics(self) -> Dict[str, Any]:
    """Get tool usage statistics."""
    return {
        'name': self.name,
        'category': self.category.value,
        'call_count': self.call_count,
        'total_execution_time_ms': self.total_execution_time,
        'average_execution_time_ms': (
            self.total_execution_time / self.call_count
            if self.call_count > 0 else 0
        )
    }

DatabaseTool：数据库工具的公共能力

DatabaseTool 继承 BaseTool，将类别固定为 DATABASE_QUERY，并注入 db_provider（即 MCP Server 实现实验 中定义的 PostgreSQLSchemaProvider 全局实例）。它提供两个核心能力：

连接上下文管理——通过 asynccontextmanager 保证连接正确获取与释放：

@asynccontextmanager
async def get_connection(self):
    """Get database connection with proper context management."""
    conn = None
    try:
        conn = await self.db_provider.get_connection()
        yield conn
    finally:
        if conn:
            await self.db_provider.release_connection(conn)

统一查询执行——execute_query 是安全与性能的汇聚点：

1. 若传入 store_id，先执行 SELECT retail.set_store_context($1) 设置门店上下文，这对应 Database 实验 中定义的 retail.set_store_context 函数——该函数会校验门店存在且激活、写入 app.current_store_id 配置项并记入审计日志，从而触发 PostgreSQL 行级安全（RLS）策略，实现多租户数据隔离；
2. 使用 conn.fetch(query, *params) 参数化查询执行，参数通过占位符 $1/$2/$3 绑定，从机制上杜绝字符串拼接型注入；
3. 记录耗时、将行转换为字典并返回 ToolResult。

async def execute_query(
    self,
    query: str,
    params: tuple = None,
    store_id: str = None
) -> ToolResult:
    """Execute database query with security and performance monitoring."""
    async with self.get_connection() as conn:
        try:
            # Set store context if provided
            if store_id:
                await conn.execute("SELECT retail.set_store_context($1)", store_id)

            # Execute query
            start_time = time.time()

            if params:
                rows = await conn.fetch(query, *params)
            else:
                rows = await conn.fetch(query)

            execution_time = (time.time() - start_time) * 1000

            # Convert rows to dictionaries
            data = [dict(row) for row in rows]

            return ToolResult(
                success=True,
                data=data,
                row_count=len(data),
                execution_time_ms=execution_time
            )

        except Exception as e:
            logger.error(f"Database query failed: {str(e)}")
            return ToolResult(
                success=False,
                error=f"Query execution failed: {str(e)}"
            )

查询校验与安全

自定义 SQL 查询是注入攻击的高发入口，本实验通过 QueryValidator 提供四层防护：语句类型白名单、危险关键字检测、注入特征正则匹配、表/模式访问白名单。

import re
import sqlparse
from typing import List, Dict, Any, Set
from enum import Enum

class QueryRisk(Enum):
    """Query risk levels."""
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    CRITICAL = "critical"

class QueryValidator:
    """Validate and analyze SQL queries for security risks."""

    # Dangerous SQL keywords and patterns
    DANGEROUS_KEYWORDS = {
        'DROP', 'DELETE', 'TRUNCATE', 'ALTER', 'CREATE', 'INSERT',
        'UPDATE', 'GRANT', 'REVOKE', 'EXEC', 'EXECUTE', 'sp_',
        'xp_', 'BULK', 'OPENROWSET', 'OPENDATASOURCE'
    }

    # Allowed read-only operations
    SAFE_KEYWORDS = {
        'SELECT', 'WITH', 'UNION', 'ORDER', 'GROUP', 'HAVING',
        'WHERE', 'FROM', 'JOIN', 'AS', 'ON', 'IN', 'EXISTS',
        'CASE', 'WHEN', 'THEN', 'ELSE', 'END', 'AND', 'OR', 'NOT'
    }

    # Allowed schemas and tables
    ALLOWED_SCHEMAS = {'retail', 'information_schema', 'pg_catalog'}
    ALLOWED_TABLES = {
        'customers', 'products', 'sales_transactions',
        'sales_transaction_items', 'product_categories',
        'product_embeddings', 'stores'
    }

    def __init__(self):
        self.injection_patterns = [
            # SQL injection patterns
            r"(\b(UNION|union)\s+(ALL\s+)?(SELECT|select))",
            r"(\b(DROP|drop)\s+(TABLE|table|DATABASE|database))",
            r"(\b(DELETE|delete)\s+(FROM|from))",
            r"(\b(INSERT|insert)\s+(INTO|into))",
            r"(\b(UPDATE|update)\s+\w+\s+(SET|set))",
            r"(\b(EXEC|exec|EXECUTE|execute)\s*\()",
            r"(\b(sp_|xp_)\w+)",
            r"(--\s*$)",  # SQL comments
            r"(/\*.*?\*/)",  # Block comments
            r"(;\s*(DROP|DELETE|INSERT|UPDATE|CREATE|ALTER))",
            r"(\bOR\b\s+['\"]?\w+['\"]?\s*=\s*['\"]?\w+['\"]?)",  # OR injection
            r"(\bAND\b\s+['\"]?\w+['\"]?\s*=\s*['\"]?\w+['\"]?)",  # AND injection
        ]

        self.compiled_patterns = [re.compile(pattern, re.IGNORECASE) for pattern in self.injection_patterns]

注意 ALLOWED_TABLES 与 Database 实验 中创建的 retail schema 表一一对应（stores、customers、products、sales_transactions、sales_transaction_items、product_categories、product_embeddings），形成"代码层白名单 + 数据库层 RLS"的双重防线。

validate_query 是校验主流程，输出结构化的验证结果：

def validate_query(self, query: str) -> Dict[str, Any]:
    """Comprehensive query validation."""
    validation_result = {
        'is_safe': True,
        'risk_level': QueryRisk.LOW,
        'issues': [],
        'warnings': [],
        'allowed_operations': [],
        'metadata': {}
    }

    try:
        # Parse the query
        parsed = sqlparse.parse(query)

        if not parsed:
            validation_result['is_safe'] = False
            validation_result['issues'].append("Unable to parse query")
            validation_result['risk_level'] = QueryRisk.HIGH
            return validation_result

        # Analyze each statement
        for statement in parsed:
            self._analyze_statement(statement, validation_result)

        # Check for injection patterns
        self._check_injection_patterns(query, validation_result)

        # Validate table/schema access
        self._validate_table_access(query, validation_result)

        # Determine final risk level
        self._determine_risk_level(validation_result)

    except Exception as e:
        validation_result['is_safe'] = False
        validation_result['issues'].append(f"Query analysis failed: {str(e)}")
        validation_result['risk_level'] = QueryRisk.CRITICAL

    return validation_result

各辅助方法职责如下：

• _analyze_statement：通过 sqlparse 解析语句类型，仅允许 SELECT 与 WITH；遍历 token，命中 DANGEROUS_KEYWORDS 判定不安全，命中 SAFE_KEYWORDS 则记录到 allowed_operations；
• _check_injection_patterns：对 12 条注入特征正则逐一匹配（UNION SELECT、DROP TABLE、注释符 --、块注释、分号拼接多语句、OR/AND 等值注入等）；
• _validate_table_access：从 FROM 与 JOIN 子句中提取表引用。带模式名的引用校验 schema 是否在 ALLOWED_SCHEMAS 中，未授权 schema 记为 issue；不带模式名的引用按 retail 推断，未在白名单的表记入 warnings；
• _determine_risk_level：注入特征命中 → CRITICAL；DROP/DELETE 等危险操作 → HIGH；其余不安全 → MEDIUM；仅含警告或完全安全 → LOW。

模块底部创建全局单例 query_validator = QueryValidator()，供各工具复用。

数据库查询工具

销售分析工具（SalesAnalysisTool）

SalesAnalysisTool 对外注册名为 execute_sales_query，定位是"支持自然语言的复杂销售分析"。它内置 5 条预构建查询模板，覆盖零售分析高频场景：

模板键	分析内容	查询要点
daily_sales	每日销售汇总	按 DATE(transaction_date) 分组统计交易数、总收入、客单价、独立客户数
top_products	热销商品排行	三表 JOIN（sales_transaction_items、sales_transactions、products），按收入降序并 LIMIT $3
customer_analysis	客户价值分析	关联客户与交易，输出消费频次、总消费额、最近购买距今天数，HAVING 过滤无交易客户
category_performance	品类表现	四表 JOIN 统计品类下的商品数、销量、收入、均价与交易数
sales_trends	销售趋势	CTE + 窗口函数：LAG 计算日环比/周环比增长率，7 日滚动平均

模板 SQL 全部使用 $1/$2/$3 参数占位符，例如 daily_sales：

'daily_sales': """
    SELECT
        DATE(transaction_date) as sales_date,
        COUNT(*) as transaction_count,
        SUM(total_amount) as total_revenue,
        AVG(total_amount) as avg_transaction_value,
        COUNT(DISTINCT customer_id) as unique_customers
    FROM retail.sales_transactions
    WHERE transaction_date >= $1 AND transaction_date <= $2
      AND transaction_type = 'sale'
    GROUP BY DATE(transaction_date)
    ORDER BY sales_date DESC
""",

执行逻辑分派 query_type：

async def execute(self, **kwargs) -> ToolResult:
    """Execute sales analysis query."""
    query_type = kwargs.get('query_type', 'custom')
    store_id = kwargs.get('store_id')

    if not store_id:
        return ToolResult(
            success=False,
            error="store_id is required for sales analysis"
        )

    try:
        if query_type in self.query_templates:
            return await self._execute_template_query(query_type, kwargs)
        elif query_type == 'custom':
            return await self._execute_custom_query(kwargs)
        else:
            return ToolResult(
                success=False,
                error=f"Unknown query type: {query_type}"
            )
    except Exception as e:
        return ToolResult(
            success=False,
            error=f"Sales analysis failed: {str(e)}"
        )

模板查询 _execute_template_query 提供合理的默认参数：start_date 默认近 30 天、end_date 默认今天、limit 默认 20；自动识别字符串日期并解析为 date；按模板是否含 $3 决定参数元组是 (start_date, end_date, limit) 还是 (start_date, end_date)；成功后在 metadata 中回填查询类型、日期范围、store_id 与分析类型：

async def _execute_template_query(self, query_type: str, kwargs: Dict[str, Any]) -> ToolResult:
    """Execute pre-built template query."""
    query = self.query_templates[query_type]
    store_id = kwargs['store_id']

    # Default parameters for template queries
    start_date = kwargs.get('start_date', (datetime.now() - timedelta(days=30)).date())
    end_date = kwargs.get('end_date', datetime.now().date())
    limit = kwargs.get('limit', 20)

    # Convert string dates if needed
    if isinstance(start_date, str):
        start_date = datetime.fromisoformat(start_date).date()
    if isinstance(end_date, str):
        end_date = datetime.fromisoformat(end_date).date()

    # Execute query with parameters
    params = (start_date, end_date, limit) if '$3' in query else (start_date, end_date)

    result = await self.execute_query(query, params, store_id)

    if result.success:
        result.metadata = {
            'query_type': query_type,
            'date_range': f"{start_date} to {end_date}",
            'store_id': store_id,
            'analysis_type': 'template'
        }

    return result

自定义查询 _execute_custom_query 是"安全自定义 SQL"的关键路径：查询必须先通过 query_validator.validate_query，不安全时拒绝执行并返回校验结果与风险等级；通过后以无参数形式交给 execute_query 执行，并把校验警告带回 metadata：

async def _execute_custom_query(self, kwargs: Dict[str, Any]) -> ToolResult:
    """Execute custom SQL query with validation."""
    custom_query = kwargs.get('query')
    store_id = kwargs['store_id']

    if not custom_query:
        return ToolResult(
            success=False,
            error="Custom query is required when query_type is 'custom'"
        )

    # Validate the query for security
    validation = query_validator.validate_query(custom_query)

    if not validation['is_safe']:
        return ToolResult(
            success=False,
            error=f"Query validation failed: {', '.join(validation['issues'])}",
            metadata={
                'validation_result': validation,
                'risk_level': validation['risk_level'].value
            }
        )

    # Execute validated query
    result = await self.execute_query(custom_query, None, store_id)

    if result.success:
        result.metadata = {
            'query_type': 'custom',
            'store_id': store_id,
            'validation_warnings': validation.get('warnings', []),
            'analysis_type': 'custom'
        }

    return result

输入 Schema 是工具与 AI 客户端之间的契约。get_input_schema 定义了完整的参数元信息，其中 query_type 通过 enum 限制为 5 个模板键加 custom；store_id 用正则 ^[a-zA-Z0-9_-]+$ 约束；limit 限定 1~1000、默认 20；并设置 required: ["store_id"] 与 additionalProperties: False（禁止未知参数，进一步收紧攻击面）：

def get_input_schema(self) -> Dict[str, Any]:
    """Get input schema for the sales analysis tool."""
    return {
        "type": "object",
        "properties": {
            "query_type": {
                "type": "string",
                "enum": list(self.query_templates.keys()) + ["custom"],
                "description": "Type of sales analysis to perform",
                "default": "daily_sales"
            },
            "store_id": {
                "type": "string",
                "description": "Store ID for data isolation",
                "pattern": "^[a-zA-Z0-9_-]+$"
            },
            "start_date": {
                "type": "string",
                "format": "date",
                "description": "Start date for analysis (YYYY-MM-DD)"
            },
            "end_date": {
                "type": "string",
                "format": "date",
                "description": "End date for analysis (YYYY-MM-DD)"
            },
            "limit": {
                "type": "integer",
                "minimum": 1,
                "maximum": 1000,
                "description": "Maximum number of results to return",
                "default": 20
            },
            "query": {
                "type": "string",
                "description": "Custom SQL query (required when query_type is 'custom')"
            }
        },
        "required": ["store_id"],
        "additionalProperties": False
    }

Schema 自省工具（SchemaIntrospectionTool）

SchemaIntrospectionTool 对外注册名为 get_table_schema，属于 SCHEMA_INTROSPECTION 类别。它让 AI 模型在编写查询前先了解表结构——这与 MCP Server 实现实验 中 execute_sales_query 工具文档"先取表结构再写查询"的最佳实践相呼应。执行时通过 table_name 的有无区分单表与全表模式：

async def execute(self, **kwargs) -> ToolResult:
    """Execute schema introspection."""
    table_name = kwargs.get('table_name')
    include_constraints = kwargs.get('include_constraints', True)
    include_indexes = kwargs.get('include_indexes', True)
    include_statistics = kwargs.get('include_statistics', False)

    try:
        if table_name:
            return await self._get_single_table_schema(
                table_name, include_constraints, include_indexes, include_statistics
            )
        else:
            return await self._get_all_tables_schema(include_constraints, include_indexes)
    except Exception as e:
        return ToolResult(
            success=False,
            error=f"Schema introspection failed: {str(e)}"
        )

单表模式 _get_single_table_schema 通过 information_schema.columns 获取列定义（列名、类型、可空性、默认值、长度、精度、序号、UDT），可选查询约束（table_constraints + key_column_usage + referential_constraints 多表关联）、索引（pg_indexes）与表统计（pg_stat_user_tables 的插入/更新/删除计数、存活/死元组、最近 vacuum/analyze 时间），全部限定 table_schema = 'retail'：

# Get column information
columns_query = """
    SELECT
        column_name,
        data_type,
        is_nullable,
        column_default,
        character_maximum_length,
        numeric_precision,
        numeric_scale,
        ordinal_position,
        udt_name
    FROM information_schema.columns
    WHERE table_schema = 'retail' AND table_name = $1
    ORDER BY ordinal_position
"""

全表模式 _get_all_tables_schema 先列出 retail schema 下所有表，再逐表取列信息，返回 schema_name + tables 数组，并在 metadata 中记录 table_count。

输入 Schema 支持可选参数：table_name 用 ^[a-zA-Z_][a-zA-Z0-9_]*$ 约束（缺省返回全部表），三个布尔开关控制约束/索引/统计信息的包含与否。

多表 Schema 工具（MultiTableSchemaTool）

MultiTableSchemaTool 对外注册名为 get_multiple_table_schemas，在一次调用中批量获取多张表的结构，显著减少 AI 模型规划查询时的往返次数。执行逻辑在单连接内循环查询各表：

async def execute(self, **kwargs) -> ToolResult:
    """Execute multi-table schema introspection."""
    table_names = kwargs.get('table_names', [])

    if not table_names:
        return ToolResult(
            success=False,
            error="At least one table name is required"
        )

    try:
        schemas = {}

        async with self.get_connection() as conn:
            for table_name in table_names:
                # Get table schema
                schema_query = """
                    SELECT
                        c.column_name,
                        c.data_type,
                        c.is_nullable,
                        c.column_default,
                        c.character_maximum_length,
                        tc.constraint_type,
                        kcu.constraint_name
                    FROM information_schema.columns c
                    LEFT JOIN information_schema.key_column_usage kcu
                        ON c.table_name = kcu.table_name
                        AND c.column_name = kcu.column_name
                        AND c.table_schema = kcu.table_schema
                    LEFT JOIN information_schema.table_constraints tc
                        ON kcu.constraint_name = tc.constraint_name
                        AND kcu.table_schema = tc.table_schema
                    WHERE c.table_schema = 'retail' AND c.table_name = $1
                    ORDER BY c.ordinal_position
                """

                columns = await conn.fetch(schema_query, table_name)

                if columns:
                    schemas[table_name] = {
                        'table_name': table_name,
                        'columns': [dict(col) for col in columns]
                    }
                else:
                    schemas[table_name] = {
                        'table_name': table_name,
                        'error': 'Table not found or not accessible'
                    }

        return ToolResult(
            success=True,
            data=schemas,
            metadata={
                'requested_tables': table_names,
                'found_tables': [name for name, info in schemas.items() if 'error' not in info],
                'missing_tables': [name for name, info in schemas.items() if 'error' in info]
            }
        )
    except Exception as e:
        return ToolResult(
            success=False,
            error=f"Multi-table schema introspection failed: {str(e)}"
        )

不存在的表不会导致整体失败，而是以 error 字段标记并在 metadata 中通过 missing_tables 汇总，体现优雅降级。其输入 Schema 要求 table_names 数组（minItems: 1、maxItems: 20，每项正则约束），并以 required: ["table_names"] 强制必填。

分析与辅助工具

商业智能工具（BusinessIntelligenceTool）

BusinessIntelligenceTool 对外注册名为 generate_business_insights，属于 ANALYTICS 类别，用于生成综合性的业务报告与洞察。它通过 analysis_type 分派四种分析（summary 汇总、customer_segmentation 客户分群、product_performance 产品表现、seasonal_trends 季节性趋势），store_id 为必填：

async def execute(self, **kwargs) -> ToolResult:
    """Execute business intelligence analysis."""
    analysis_type = kwargs.get('analysis_type', 'summary')
    store_id = kwargs.get('store_id')

    if not store_id:
        return ToolResult(
            success=False,
            error="store_id is required for business intelligence analysis"
        )

    try:
        if analysis_type == 'summary':
            return await self._generate_business_summary(kwargs)
        elif analysis_type == 'customer_segmentation':
            return await self._analyze_customer_segmentation(kwargs)
        elif analysis_type == 'product_performance':
            return await self._analyze_product_performance(kwargs)
        elif analysis_type == 'seasonal_trends':
            return await self._analyze_seasonal_trends(kwargs)
        else:
            return ToolResult(
                success=False,
                error=f"Unknown analysis type: {analysis_type}"
            )
    except Exception as e:
        return ToolResult(
            success=False,
            error=f"Business intelligence analysis failed: {str(e)}"
        )

_generate_business_summary 用一条多 CTE 查询聚合销售汇总、产品汇总与最热品类，并在此基础上计算派生指标（日均交易数、客单价、单笔件数等），追加到结果中：

if result.success and result.data:
    summary = result.data[0]

    # Add derived insights
    insights = {
        'revenue_trend': 'stable',  # Would calculate based on historical data
        'customer_retention': f"{summary.get('unique_customers', 0)} active customers",
        'performance_indicators': {
            'transactions_per_day': round(summary.get('total_transactions', 0) / max(summary.get('active_days', 1), 1), 2),
            'revenue_per_customer': round(summary.get('total_revenue', 0) / max(summary.get('unique_customers', 1), 1), 2),
            'items_per_transaction': round(summary.get('total_items_sold', 0) / max(summary.get('total_transactions', 1), 1), 2)
        }
    }

    summary['insights'] = insights

    result.data = [summary]
    result.metadata = {
        'analysis_type': 'business_summary',
        'period_days': days,
        'store_id': store_id
    }

其输入 Schema 中 analysis_type 用 enum 限定四种取值（默认 summary），days 限定 1~365（默认 30）。

辅助工具（UtilityTool）

UtilityTool 对外注册名为 get_current_utc_date，属于 UTILITY 类别，为 AI 提供统一的时间基准（对"最近 30 天"等时间敏感查询至关重要）。它根据 format 参数（iso/epoch/date_only，默认 iso）选择不同的 PostgreSQL 表达式：

async def execute(self, **kwargs) -> ToolResult:
    """Execute utility operation."""
    format_type = kwargs.get('format', 'iso')

    try:
        async with self.get_connection() as conn:
            if format_type == 'iso':
                query = "SELECT CURRENT_TIMESTAMP AT TIME ZONE 'UTC' as current_utc_datetime"
            elif format_type == 'epoch':
                query = "SELECT EXTRACT(EPOCH FROM CURRENT_TIMESTAMP AT TIME ZONE 'UTC') as current_utc_epoch"
            elif format_type == 'date_only':
                query = "SELECT CURRENT_DATE as current_date"
            else:
                return ToolResult(
                    success=False,
                    error=f"Unknown format type: {format_type}"
                )

            result = await conn.fetchrow(query)

            return ToolResult(
                success=True,
                data=dict(result),
                metadata={
                    'format_type': format_type,
                    'timezone': 'UTC'
                }
            )
    except Exception as e:
        return ToolResult(
            success=False,
            error=f"Utility operation failed: {str(e)}"
        )

与服务器实现和数据库层的衔接

本实验的工具设计并非孤岛，而是与整条实验链深度咬合：

• 连接与 RLS：DatabaseTool.execute_query 中 retail.set_store_context($1) 直接调用 Database 实验 定义的函数，该函数校验门店有效性、写入 app.current_store_id 并记录审计日志；配合 customers/products/sales_transactions 等表上的 RLS 策略，实现"同一 SQL 面对不同门店返回不同数据"的多租户隔离。
• 工具注册：MCP Server 实现实验 中 sales_analysis.py 通过 @mcp.tool() 注册的 get_multiple_table_schemas、execute_sales_query、get_current_utc_date，与本实验的 MultiTableSchemaTool、SalesAnalysisTool、UtilityTool 一一对应；FastMCP 会基于工具的 Pydantic 字段描述自动生成 MCP tools/list 所需的 JSON Schema。
• 查询超时与结果截断：服务器层的 execute_query（见 sales_analysis_postgres.py 相关实现）通过 asyncio.wait_for 设置 command_timeout（默认 30 秒）并默认只返回前 20 行，防止长查询与超大结果集拖垮 AI 会话。

关键要点

完成本实验后，你应当掌握：

• ✅ 进阶工具架构：实现了带全面错误处理、参数校验与统计埋点的复杂 MCP 工具（统一继承 BaseTool/DatabaseTool，返回 ToolResult）
• ✅ 查询校验：构建了防止注入攻击的安全 SQL 校验器（语句类型白名单 + 危险关键字 + 注入正则 + 表/模式白名单四层防线）
• ✅ 数据库工具：创建了强大的销售分析与 Schema 自省能力（模板化查询 + 单表/多表/全表自省）
• ✅ 商业智能：开发了面向全面业务洞察的分析工具（汇总、客户分群、产品表现、季节性趋势）
• ✅ 性能优化：应用了连接管理、查询超时、结果截断与执行耗时度量
• ✅ 安全集成：实现了基于角色的访问控制（store_id 强制校验 + RLS 上下文）与动作审计（日志脱敏 + set_store_context 审计写入）

下一步

继续学习 实验 07：语义搜索集成，你将在此基础上：

• 将向量搜索能力与 MCP 工具集成
• 构建语义化商品搜索功能
• 实现 AI 驱动的查询理解
• 创建融合传统查询与向量查询的混合搜索

相关前置阅读：实验 05：MCP 服务器实现（工具所依赖的连接池、RLS 上下文与 FastMCP 注册层）、实验 04：数据库设计与 Schema（retail schema、RLS 策略与 set_store_context 函数）。

说明：本实验文档由 AI 翻译服务生成（见仓库 translations 版本），本文基于英文原版 11-MCPServerHandsOnLabs/06-Tools/README.md 编写；涉及外部链接的资源章节请以原文档为准。