MCP Tool 开发实战：为 AI 助手构建安全、可观测、高性能的数据库工具集（MCP Server Hands-On Labs 06）
MCP Tool 开发实战：为 AI 助手构建安全、可观测、高性能的数据库工具集（MCP Server Hands-On Labs 06）
导读
本篇文章是《MCP Server Hands-On Labs》学习路径的第六个实验（06-Tools），围绕 Zava Retail Analytics MCP Server 展开。在上一实验（Lab 05: MCP Server Implementation）中，你已经用 FastMCP 搭建起了服务器骨架、数据库连接池与三个基础工具；本实验则带你深入 MCP 工具开发的完整生命周期——从工具架构设计、参数校验、SQL 注入防护、Schema 内省，到商业智能分析工具的实现与性能优化。学完本篇，你将掌握：如何设计带复杂参数校验的高级 MCP 工具、如何在工具层建立查询验证与安全边界、如何构建动态 Schema 内省与自定义分析能力，以及如何将缓存、连接池、指标统计与审计日志融入工具实现。
本实验覆盖什么
本实验深入讲解如何创建"既强大又安全"的 MCP 工具，为 AI 助手提供：
- 数据库查询能力：支持自然语言驱动的销售数据分析；
- Schema 内省能力：动态发现数据库表结构，支撑 AI 查询规划；
- 分析函数：业务智能（Business Intelligence）统计与指标计算；
- 完整错误处理与优雅降级；
- 生产级性能优化。
MCP 工具是 AI 助手与数据系统之间的接口。设计良好的工具应以结构化、经过校验的方式开放复杂操作，同时保持安全边界与性能。本实验的零售 MCP 服务器实现了一套完整的工具套件，让 AI 助手能够以自然语言查询销售数据、商品目录和业务分析指标，同时严格遵守安全边界并保持最优性能。
学习目标
完成本实验后，你将能够：
- 设计带复杂参数校验的高级 MCP 工具；
- 实现带 SQL 注入防护的安全数据库查询工具；
- 创建支持动态查询的 Schema 内省能力；
- 构建用于商业智能的自定义分析工具；
- 应用全面的错误处理与优雅降级策略；
- 优化面向生产负载的工具性能。
核心工具架构
本实验首先定义了一套可复用的工具基座，位于 mcp_server/tools/base.py。所有工具类都继承自 BaseTool，统一了分类、结果结构、执行包装、参数校验与运行指标。
工具分类与统一结果结构
# mcp_server/tools/base.py
"""
Base classes and patterns for MCP tool development.
"""
from abc import ABC, abstractmethod
from typing import Any, Dict, List, Optional, Union
from dataclasses import dataclass
from enum import Enum
import asyncio
import time
import logging
from contextlib import asynccontextmanager
logger = logging.getLogger(__name__)
class ToolCategory(Enum):
"""Tool categorization for organization and discovery."""
DATABASE_QUERY = "database_query"
SCHEMA_INTROSPECTION = "schema_introspection"
ANALYTICS = "analytics"
UTILITY = "utility"
ADMINISTRATIVE = "administrative"
@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
ToolCategory 枚举将工具划分为数据库查询、Schema 内省、分析、工具类与运维类五类，便于组织与发现。ToolResult 是一个标准化的结果结构，统一携带 success 标志、data 数据、error 错误信息、metadata 元数据、execution_time_ms 执行耗时与 row_count 行数——AI 客户端与日志系统都能以一致的契约消费工具输出。
BaseTool：抽象基类与执行包装器
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
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
)
def _validate_input(self, kwargs: Dict[str, Any]):
"""Validate input parameters against schema."""
schema = self.get_input_schema()
required_props = schema.get('required', [])
properties = schema.get('properties', {})
# Check required parameters
missing_required = [prop for prop in required_props if prop not in kwargs]
if missing_required:
raise ValueError(f"Missing required parameters: {missing_required}")
# Type validation would go here
# For production, use jsonschema library for comprehensive validation
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
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
)
}
BaseTool.call() 是工具执行的统一包装器，承担四件事：
- 参数校验：基于
get_input_schema()返回的 JSON Schema 检查必填参数，缺失即抛出ValueError（生产环境可进一步引入jsonschema库做完整类型校验）； - 审计日志：记录工具名、分类与参数，但通过
_sanitize_parameters()将password、token、secret、key等敏感键掩码为***MASKED***，防止敏感信息进入日志； - 指标采集：统计每次调用的执行耗时（毫秒），累计到
total_execution_time，供get_statistics()生成调用次数、总耗时、平均耗时的运行指标； - 异常兜底：任何异常都被捕获并转换为
ToolResult(success=False, ...)，保证工具调用永远不会以裸异常形式崩溃——这正是"优雅降级"的基础。
DatabaseTool：数据库工具基类
class DatabaseTool(BaseTool):
"""Base class for database-related tools."""
def __init__(self, name: str, description: str, db_provider):
super().__init__(name, description, ToolCategory.DATABASE_QUERY)
self.db_provider = db_provider
@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)
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)}"
)
DatabaseTool 封装了所有数据库类工具的公共逻辑：
- 连接上下文管理：
get_connection()通过asynccontextmanager获取连接并在finally中确保释放，配合连接池实现资源复用； - 租户隔离上下文：
execute_query()在传入store_id时先执行SELECT retail.set_store_context($1)，为当前连接设置门店上下文——这与数据库层的 Row Level Security 策略配合，确保用户只能访问其授权门店的数据（参考 Lab 04 数据库设计 中基于current_setting('app.current_store_id', true)的 RLS 策略）； - 参数化查询：优先走
conn.fetch(query, *params)参数绑定路径，从根本上规避字符串拼接注入； - 计时与行数统计：记录执行耗时与返回行数，填充
ToolResult。
查询验证与安全：QueryValidator
自定义 SQL 查询是注入攻击的高发面。本实验在 mcp_server/tools/query_validator.py 中实现了 QueryValidator，对所有自定义查询做多重安全校验。
# mcp_server/tools/query_validator.py
"""
SQL query validation and security for MCP tools.
"""
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]
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
def _analyze_statement(self, statement, validation_result):
"""Analyze individual SQL statement."""
# Get statement type
stmt_type = statement.get_type()
# Check if statement type is allowed
if stmt_type and stmt_type.upper() not in ['SELECT', 'WITH']:
validation_result['issues'].append(f"Disallowed statement type: {stmt_type}")
validation_result['is_safe'] = False
return
# Extract tokens and analyze
for token in statement.flatten():
if token.ttype is sqlparse.tokens.Keyword:
keyword = token.value.upper()
if keyword in self.DANGEROUS_KEYWORDS:
validation_result['issues'].append(f"Dangerous keyword detected: {keyword}")
validation_result['is_safe'] = False
elif keyword in self.SAFE_KEYWORDS:
if keyword not in validation_result['allowed_operations']:
validation_result['allowed_operations'].append(keyword)
def _check_injection_patterns(self, query: str, validation_result):
"""Check for SQL injection patterns."""
for pattern in self.compiled_patterns:
matches = pattern.findall(query)
if matches:
validation_result['issues'].append(f"Potential injection pattern detected")
validation_result['is_safe'] = False
def _validate_table_access(self, query: str, validation_result):
"""Validate that only allowed tables/schemas are accessed."""
# Extract table names (simplified approach)
# In production, use proper SQL parsing
from_match = re.findall(r'FROM\s+(\w+\.?\w*)', query, re.IGNORECASE)
join_match = re.findall(r'JOIN\s+(\w+\.?\w*)', query, re.IGNORECASE)
all_tables = from_match + join_match
for table_ref in all_tables:
if '.' in table_ref:
schema, table = table_ref.split('.', 1)
if schema.lower() not in self.ALLOWED_SCHEMAS:
validation_result['issues'].append(f"Access to unauthorized schema: {schema}")
validation_result['is_safe'] = False
if table.lower() not in self.ALLOWED_TABLES:
validation_result['warnings'].append(f"Access to table: {table}")
else:
# Assume retail schema if not specified
if table_ref.lower() not in self.ALLOWED_TABLES:
validation_result['warnings'].append(f"Access to table: {table_ref}")
def _determine_risk_level(self, validation_result):
"""Determine overall risk level."""
if not validation_result['is_safe']:
if any('injection' in issue.lower() for issue in validation_result['issues']):
validation_result['risk_level'] = QueryRisk.CRITICAL
elif any('DROP' in issue or 'DELETE' in issue for issue in validation_result['issues']):
validation_result['risk_level'] = QueryRisk.HIGH
else:
validation_result['risk_level'] = QueryRisk.MEDIUM
elif validation_result['warnings']:
validation_result['risk_level'] = QueryRisk.LOW
else:
validation_result['risk_level'] = QueryRisk.LOW
# Global validator instance
query_validator = QueryValidator()
validate_query() 的四层校验逻辑值得逐层拆解：
- 语法解析：用
sqlparse解析 SQL，解析失败直接判为不安全和HIGH风险； - 语句类型白名单：
_analyze_statement()只允许SELECT/WITH类型，其余（INSERT、UPDATE、DELETE等）一律拒绝；同时对语句 token 流逐一比对DANGEROUS_KEYWORDS与SAFE_KEYWORDS，前者立即判不安全，后者计入allowed_operations； - 注入模式正则：
_check_injection_patterns()匹配 11 组预编译的正则（UNION SELECT 联合注入、注释符--//* */、OR 1=1恒真条件、存储过程前缀sp_/xp_、多语句分隔等），命中即标记"潜在注入模式"； - 表/模式访问控制：
_validate_table_access()从FROM与JOIN提取表引用，非retail/information_schema/pg_catalog模式直接拒绝，未列入ALLOWED_TABLES的表给出警告。
风险等级由 _determine_risk_level() 汇总：存在注入模式 → CRITICAL；含 DROP/DELETE → HIGH；其余不安全情况 → MEDIUM；仅有警告 → LOW；完全干净 → LOW。文件末尾导出一个全局单例 query_validator，供各工具复用。
数据库查询工具：SalesAnalysisTool
mcp_server/tools/sales_analysis.py 中的 SalesAnalysisTool 是套件中最核心的查询工具，对外暴露名称为 execute_sales_query。它的设计亮点是"预置模板 + 自定义查询"双通道：常见分析走经过验证的 SQL 模板，特殊需求走 QueryValidator 校验后的自定义 SQL。
# mcp_server/tools/sales_analysis.py
"""
Comprehensive sales analysis tool for retail data querying.
"""
from typing import Dict, Any, List, Optional
from datetime import datetime, timedelta
from .base import DatabaseTool, ToolResult
from .query_validator import query_validator
class SalesAnalysisTool(DatabaseTool):
"""Advanced sales analysis and reporting tool."""
def __init__(self, db_provider):
super().__init__(
name="execute_sales_query",
description="Execute sophisticated sales analysis queries with natural language support",
db_provider=db_provider
)
# Pre-built query templates for common analysis
self.query_templates = {
'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
""",
'top_products': """
SELECT
p.product_name,
p.brand,
SUM(sti.quantity) as total_quantity_sold,
SUM(sti.total_price) as total_revenue,
COUNT(DISTINCT st.transaction_id) as transaction_count,
AVG(sti.unit_price) as avg_price
FROM retail.sales_transaction_items sti
JOIN retail.sales_transactions st ON sti.transaction_id = st.transaction_id
JOIN retail.products p ON sti.product_id = p.product_id
WHERE st.transaction_date >= $1 AND st.transaction_date <= $2
AND st.transaction_type = 'sale'
GROUP BY p.product_id, p.product_name, p.brand
ORDER BY total_revenue DESC
LIMIT $3
""",
'customer_analysis': """
SELECT
c.customer_id,
c.first_name || ' ' || c.last_name as customer_name,
c.loyalty_tier,
COUNT(st.transaction_id) as transaction_count,
SUM(st.total_amount) as total_spent,
AVG(st.total_amount) as avg_transaction_value,
MAX(st.transaction_date) as last_purchase_date,
DATE_PART('day', CURRENT_DATE - MAX(st.transaction_date)) as days_since_last_purchase
FROM retail.customers c
LEFT JOIN retail.sales_transactions st ON c.customer_id = st.customer_id
WHERE st.transaction_date >= $1 AND st.transaction_date <= $2
AND st.transaction_type = 'sale'
GROUP BY c.customer_id, c.first_name, c.last_name, c.loyalty_tier
HAVING COUNT(st.transaction_id) > 0
ORDER BY total_spent DESC
LIMIT $3
""",
'category_performance': """
SELECT
pc.category_name,
COUNT(DISTINCT p.product_id) as unique_products,
SUM(sti.quantity) as total_quantity_sold,
SUM(sti.total_price) as total_revenue,
AVG(sti.unit_price) as avg_price,
COUNT(DISTINCT st.transaction_id) as transaction_count
FROM retail.product_categories pc
JOIN retail.products p ON pc.category_id = p.category_id
JOIN retail.sales_transaction_items sti ON p.product_id = sti.product_id
JOIN retail.sales_transactions st ON sti.transaction_id = st.transaction_id
WHERE st.transaction_date >= $1 AND st.transaction_date <= $2
AND st.transaction_type = 'sale'
GROUP BY pc.category_id, pc.category_name
ORDER BY total_revenue DESC
""",
'sales_trends': """
WITH daily_sales AS (
SELECT
DATE(transaction_date) as sales_date,
SUM(total_amount) as daily_revenue,
COUNT(*) as daily_transactions
FROM retail.sales_transactions
WHERE transaction_date >= $1 AND transaction_date <= $2
AND transaction_type = 'sale'
GROUP BY DATE(transaction_date)
),
trend_analysis AS (
SELECT
sales_date,
daily_revenue,
daily_transactions,
LAG(daily_revenue, 1) OVER (ORDER BY sales_date) as prev_day_revenue,
LAG(daily_revenue, 7) OVER (ORDER BY sales_date) as prev_week_revenue,
AVG(daily_revenue) OVER (
ORDER BY sales_date
ROWS BETWEEN 6 PRECEDING AND CURRENT ROW
) as rolling_7day_avg
FROM daily_sales
)
SELECT
sales_date,
daily_revenue,
daily_transactions,
rolling_7day_avg,
CASE
WHEN prev_day_revenue IS NOT NULL THEN
ROUND(((daily_revenue - prev_day_revenue) / prev_day_revenue * 100)::numeric, 2)
ELSE NULL
END as day_over_day_growth_pct,
CASE
WHEN prev_week_revenue IS NOT NULL THEN
ROUND(((daily_revenue - prev_week_revenue) / prev_week_revenue * 100)::numeric, 2)
ELSE NULL
END as week_over_week_growth_pct
FROM trend_analysis
ORDER BY sales_date DESC
"""
}
五个模板覆盖了零售分析的典型场景：daily_sales（每日销量与营收）、top_products（热销商品 TOP N）、customer_analysis（客户价值与忠诚度分层）、category_performance（品类表现）、sales_trends（环比增长与 7 日滚动均值，使用 PostgreSQL 窗口函数 LAG/AVG OVER）。所有模板均使用 $1、$2、$3 位置参数占位符，配合参数化执行，杜绝注入。
执行分流与自定义查询
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)}"
)
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
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
关键执行细节：
- store_id 强校验：无论模板还是自定义查询，
store_id都是必填项，缺失立即返回失败，保证租户隔离； - 模板参数默认值：
start_date默认最近 30 天、end_date默认今天、limit默认 20；字符串日期通过datetime.fromisoformat()转成date； - 参数个数自适应：
params = (start_date, end_date, limit) if '$3' in query else (start_date, end_date)，仅当模板含$3时才传入 limit； - 自定义查询双保险：
custom类型必须提供query，且必须先过QueryValidator；校验失败时把issues与risk_level一并写进错误与metadata，让调用方（AI 助手）明确知道被拒原因。
工具输入 Schema
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 直接驱动两件事：一是作为工具元数据随 MCP tools/list 暴露给 AI 客户端，让模型知道可填参数与约束；二是作为 BaseTool._validate_input() 的校验依据。store_id 使用正则 ^[a-zA-Z0-9_-]+$ 约束字符集，limit 限定 1～1000 范围，additionalProperties: False 拒绝未知字段——这些都是防御性设计的体现。
Schema 内省工具
AI 要写出正确的 SQL，前提是知道表结构。mcp_server/tools/schema_introspection.py 提供两类内省工具：SchemaIntrospectionTool（单表/全表）与 MultiTableSchemaTool（多表批量）。
单表/全表内省
# mcp_server/tools/schema_introspection.py
"""
Database schema introspection and metadata tools.
"""
from typing import Dict, Any, List
from .base import DatabaseTool, ToolResult, ToolCategory
class SchemaIntrospectionTool(DatabaseTool):
"""Tool for exploring database schema and metadata."""
def __init__(self, db_provider):
super().__init__(
name="get_table_schema",
description="Get detailed schema information for database tables",
db_provider=db_provider
)
self.category = ToolCategory.SCHEMA_INTROSPECTION
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)}"
)
execute() 根据是否传入 table_name 分流：传入则查单表，否则返回 retail 模式下全部表。单表内省依次执行四组 information_schema / pg_catalog 查询：
async def _get_single_table_schema(
self,
table_name: str,
include_constraints: bool,
include_indexes: bool,
include_statistics: bool
) -> ToolResult:
"""Get detailed schema for a single table."""
schema_info = {
'table_name': table_name,
'columns': [],
'constraints': [],
'indexes': [],
'statistics': {}
}
async with self.get_connection() as conn:
# 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
"""
columns = await conn.fetch(columns_query, table_name)
schema_info['columns'] = [dict(col) for col in columns]
# Get constraints if requested
if include_constraints:
constraints_query = """
SELECT
constraint_name,
constraint_type,
column_name,
foreign_table_name,
foreign_column_name
FROM information_schema.table_constraints tc
LEFT JOIN information_schema.key_column_usage kcu
ON tc.constraint_name = kcu.constraint_name
LEFT JOIN information_schema.referential_constraints rc
ON tc.constraint_name = rc.constraint_name
LEFT JOIN information_schema.key_column_usage fkcu
ON rc.unique_constraint_name = fkcu.constraint_name
WHERE tc.table_schema = 'retail' AND tc.table_name = $1
"""
constraints = await conn.fetch(constraints_query, table_name)
schema_info['constraints'] = [dict(const) for const in constraints]
# Get indexes if requested
if include_indexes:
indexes_query = """
SELECT
indexname as index_name,
indexdef as index_definition,
tablespace
FROM pg_indexes
WHERE schemaname = 'retail' AND tablename = $1
"""
indexes = await conn.fetch(indexes_query, table_name)
schema_info['indexes'] = [dict(idx) for idx in indexes]
# Get table statistics if requested
if include_statistics:
stats_query = """
SELECT
n_tup_ins as inserts,
n_tup_upd as updates,
n_tup_del as deletes,
n_live_tup as live_tuples,
n_dead_tup as dead_tuples,
last_vacuum,
last_autovacuum,
last_analyze,
last_autoanalyze
FROM pg_stat_user_tables
WHERE schemaname = 'retail' AND relname = $1
"""
stats = await conn.fetchrow(stats_query, table_name)
if stats:
schema_info['statistics'] = dict(stats)
return ToolResult(
success=True,
data=schema_info,
metadata={
'table_name': table_name,
'schema': 'retail',
'introspection_type': 'single_table'
}
)
include_statistics 默认关闭（False），因为 pg_stat_user_tables 的统计查询相对昂贵；include_constraints 与 include_indexes 默认开启。全表内省 _get_all_tables_schema() 则先从 information_schema.tables 列出 retail 模式下所有表，再逐表查询列信息，并在 metadata 中返回 table_count。
多表批量内省
class MultiTableSchemaTool(DatabaseTool):
"""Tool for getting schema information for multiple tables at once."""
def __init__(self, db_provider):
super().__init__(
name="get_multiple_table_schemas",
description="Get schema information for multiple tables efficiently",
db_provider=db_provider
)
self.category = ToolCategory.SCHEMA_INTROSPECTION
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)}"
)
def get_input_schema(self) -> Dict[str, Any]:
"""Get input schema for multi-table schema tool."""
return {
"type": "object",
"properties": {
"table_names": {
"type": "array",
"items": {
"type": "string",
"pattern": "^[a-zA-Z_][a-zA-Z0-9_]*$"
},
"description": "List of table names to get schema information for",
"minItems": 1,
"maxItems": 20
}
},
"required": ["table_names"],
"additionalProperties": False
}
MultiTableSchemaTool 在单次调用中批量获取多表 Schema：复用同一个连接循环查询各表，表中每行同时携带列信息与约束/索引信息（通过 key_column_usage / table_constraints 左连接）；不存在的表不会让整个调用失败，而是记录 error 字段，最终在 metadata 中分别统计 found_tables 与 missing_tables。输入 Schema 通过 minItems: 1、maxItems: 20 与正则 ^[a-zA-Z_][a-zA-Z0-9_]*$ 约束表名列表。
分析与工具类工具
BusinessIntelligenceTool：业务智能分析
mcp_server/tools/business_intelligence.py 的 BusinessIntelligenceTool 对外暴露为 generate_business_insights，支持四种分析类型：summary（经营摘要）、customer_segmentation（客户分层）、product_performance（商品表现）、seasonal_trends（季节趋势）。
# mcp_server/tools/business_intelligence.py
"""
Advanced business intelligence and analytics tools.
"""
from typing import Dict, Any, List
from datetime import datetime, timedelta
from .base import DatabaseTool, ToolResult, ToolCategory
class BusinessIntelligenceTool(DatabaseTool):
"""Advanced analytics tool for business intelligence queries."""
def __init__(self, db_provider):
super().__init__(
name="generate_business_insights",
description="Generate comprehensive business intelligence reports and insights",
db_provider=db_provider
)
self.category = ToolCategory.ANALYTICS
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() 为例，它在一条 SQL 中用多个 CTE（date_range、sales_summary、product_summary、top_category）聚合出总交易数、独立客户数、总营收、平均客单价、活跃天数、售出商品数、畅销品类等指标，再在 Python 侧推导 revenue_trend、customer_retention 与 performance_indicators（日均交易数、人均营收、单笔商品数）：
async def _generate_business_summary(self, kwargs: Dict[str, Any]) -> ToolResult:
"""Generate comprehensive business summary."""
store_id = kwargs['store_id']
days = kwargs.get('days', 30)
summary_query = """
WITH date_range AS (
SELECT CURRENT_DATE - INTERVAL '%s days' as start_date,
CURRENT_DATE as end_date
),
sales_summary AS (
SELECT
COUNT(*) as total_transactions,
COUNT(DISTINCT customer_id) as unique_customers,
SUM(total_amount) as total_revenue,
AVG(total_amount) as avg_transaction_value,
COUNT(DISTINCT DATE(transaction_date)) as active_days
FROM retail.sales_transactions st, date_range dr
WHERE st.transaction_date >= dr.start_date
AND st.transaction_date <= dr.end_date
AND st.transaction_type = 'sale'
),
product_summary AS (
SELECT
COUNT(DISTINCT p.product_id) as products_sold,
SUM(sti.quantity) as total_items_sold
FROM retail.sales_transaction_items sti
JOIN retail.sales_transactions st ON sti.transaction_id = st.transaction_id
JOIN retail.products p ON sti.product_id = p.product_id
CROSS JOIN date_range dr
WHERE st.transaction_date >= dr.start_date
AND st.transaction_date <= dr.end_date
AND st.transaction_type = 'sale'
),
top_category AS (
SELECT
pc.category_name,
SUM(sti.total_price) as category_revenue
FROM retail.product_categories pc
JOIN retail.products p ON pc.category_id = p.category_id
JOIN retail.sales_transaction_items sti ON p.product_id = sti.product_id
JOIN retail.sales_transactions st ON sti.transaction_id = st.transaction_id
CROSS JOIN date_range dr
WHERE st.transaction_date >= dr.start_date
AND st.transaction_date <= dr.end_date
AND st.transaction_type = 'sale'
GROUP BY pc.category_name
ORDER BY category_revenue DESC
LIMIT 1
)
SELECT
ss.*,
ps.products_sold,
ps.total_items_sold,
tc.category_name as top_category,
tc.category_revenue as top_category_revenue,
CASE
WHEN ss.active_days > 0 THEN ss.total_revenue / ss.active_days
ELSE 0
END as avg_daily_revenue
FROM sales_summary ss
CROSS JOIN product_summary ps
CROSS JOIN top_category tc
""" % days
result = await self.execute_query(summary_query, None, store_id)
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
}
return result
注意其中的防御性细节：除法前用 max(..., 1) 兜底，避免 active_days、unique_customers 为零时触发除零异常。其输入 Schema 中 analysis_type 枚举为 ["summary", "customer_segmentation", "product_performance", "seasonal_trends"]，days 限定 1～365，默认 30，同样以 store_id 为必填。
UtilityTool：工具类工具
class UtilityTool(DatabaseTool):
"""Utility tool for common operations."""
def __init__(self, db_provider):
super().__init__(
name="get_current_utc_date",
description="Get current UTC date and time for reference",
db_provider=db_provider
)
self.category = ToolCategory.UTILITY
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)}"
)
def get_input_schema(self) -> Dict[str, Any]:
"""Get input schema for utility tool."""
return {
"type": "object",
"properties": {
"format": {
"type": "string",
"enum": ["iso", "epoch", "date_only"],
"description": "Format for the returned date/time",
"default": "iso"
}
},
"additionalProperties": False
}
get_current_utc_date 让 AI 助手在写时间敏感查询前获取可信的 UTC 时间基准，支持 iso（ISO 时间戳）、epoch（Unix 时间戳）、date_only（仅日期）三种格式，默认 iso。与 Lab 05 服务器实现 中的 db_provider.get_current_utc_date() 相比，这里以独立工具形式暴露，并将返回结构统一为 ToolResult。
与服务器注册层的衔接
本实验中的工具类定义在 mcp_server/tools/ 目录下，最终需挂载到 FastMCP 服务器（Lab 05）才能被 AI 客户端发现。从 05-MCP-Server 的实现 可以看到两种注册方式的衔接点：
- 声明式注册：FastMCP 通过
@mcp.tool()装饰器注册函数式工具（如get_multiple_table_schemas、execute_sales_query、get_current_utc_date），函数签名中的Annotated[..., Field(description=...)]即生成 MCP 工具 Schema； - 类式注册：本实验的
BaseTool子类以name+get_input_schema()提供同等的工具描述与参数 Schema，call()包装器补上了指标统计与日志审计。
两者互补：类式实现更适合承载复杂状态（模板库、调用计数、累计耗时），声明式实现更适合简单工具。生产服务器通常会注册一个聚合入口（如 execute_sales_query），把 AI 传来的自然语言意图映射到本实验的模板或自定义查询通道。
性能优化与可观测性要点
回顾本实验的整个工具栈，生产级性能与可观测性体现在四处：
- 连接池：
DatabaseTool.get_connection()复用 Lab 05 的asyncpg连接池（min_size/max_size由DatabaseConfig控制，默认 2/10），避免每次工具调用新建连接； - 预编译模板：常用分析走预置 SQL 模板 + 参数绑定，既快又安全；
- 指标统计：
BaseTool累积call_count与total_execution_time，get_statistics()提供平均耗时，可对接监控面板； - 审计日志：
call()在成功/失败路径都输出结构化日志（含执行耗时、行数、掩码后的参数），配合 Lab 05 的健康检查端点 与 Lab 11 监控实验室 形成完整可观测链路。
关键要点
完成本实验后，你应该已经掌握：
- ✅ 高级工具架构：实现了带全面错误处理的复杂 MCP 工具；
- ✅ 查询验证：构建了可防注入攻击的安全 SQL 校验机制；
- ✅ 数据库工具：创建了强大的销售分析与 Schema 内省能力；
- ✅ 商业智能：开发了提供综合业务洞察的分析工具；
- ✅ 性能优化：应用了缓存、连接池与查询优化策略；
- ✅ 安全集成：实现了基于角色的访问控制与审计日志。
下一步
继续进入 Lab 07: Semantic Search Integration，你将：
- 将向量搜索能力与 MCP 工具集成；
- 构建语义化商品搜索功能；
- 实现 AI 驱动的查询理解；
- 创建融合传统查询与向量查询的混合搜索。
本实验配套的数据库 Schema（retail.stores、retail.customers、retail.products、retail.sales_transactions 等表）与 RLS 多租户策略详见 Lab 04: Database Design and Schema；服务器骨架与注册方式详见 Lab 05: MCP Server Implementation；整个 12 个实验室的学习路径总览见 11-MCPServerHandsOnLabs 索引。
登录后查看全文