好的我们来换一个同样对开发者极具实用价值的主题。这次我们聚焦于“如何利用 MCP 协议将本地数据库如 MySQL、PostgreSQL安全、智能地暴露给 AI 助手进行查询与分析”。基于 MCP 协议构建数据库智能查询助手原理、安全实践与代码实现在数据驱动的开发与运维工作中开发者经常需要查询数据库以获取信息、验证假设或生成报告。传统方式需要手动连接数据库、编写 SQL过程繁琐且容易出错。本文将深入探讨如何利用MCPModel Context Protocol协议构建一个能与 AI 助手如 Cursor、Claude Desktop自然对话并安全执行数据库查询的智能工具。一、场景与价值为什么需要数据库 MCP Server想象以下场景开发调试AI 助手根据你描述的“查找最近一周订单量大于 100 的用户及其邮箱”自动生成并执行 SQL返回结果。数据探查在编写数据分析代码前让 AI 快速帮你查看某个表的 schema、样本数据或统计信息。运维监控询问 AI “当前数据库的连接数是多少” 或 “找出过去一小时慢查询最多的表”。报告生成结合 AI 的文本生成能力将查询结果自动汇总成一段分析文字或 Markdown 报告。核心价值自然语言交互用说话的方式查询数据降低 SQL 编写门槛。上下文感知AI 能结合对话历史理解你的查询意图甚至提出优化建议。安全可控通过 MCP Server 封装可以严格限制 AI 能访问的数据库、表、以及可执行的操作如只读 SELECT避免直接暴露数据库连接凭证和危险操作如 DROP TABLE。二、架构设计安全是首要原则一个健壮的数据库 MCP Server 架构应遵循“最小权限”和“操作隔离”原则[AI Client (Cursor/Claude)] | | (通过 stdio 发送 JSON-RPC 请求如 execute_query) v [MCP Server (守护进程)] | 1. 解析、验证请求 | 2. 应用安全策略SQL白名单、表权限等 | 3. 记录审计日志 v [Database Connector Pool] | v [Target Database (MySQL/PostgreSQL/SQLite)]关键安全考量连接池与凭证管理Server 启动时建立连接池AI 客户端永不接触数据库密码。凭证通过环境变量或加密配置文件管理。SQL 注入防御绝不允许 AI 直接拼接 SQL 字符串执行。应使用参数化查询或严格限制 AI 只能调用预定义的“查询模板”。权限细分为 MCP Server 创建专用的数据库用户仅授予SELECT权限在特定的 schema 或表上。查询限制在 Server 端强制添加LIMIT子句防止 AI 无意中发起全表扫描导致性能问题。审计日志记录所有由 AI 发起的查询、执行时间、结果行数便于追踪和复盘。三、实战构建一个 Python FastMCP 数据库 Server我们以PostgreSQL为例使用fastmcp、asyncpg和pydantic来构建。1. 项目初始化与依赖mkdir mcp-db-assistant cd mcp-db-assistant python -m venv venv source venv/bin/activatepip install fastmcp asyncpg pydantic python-dotenv2. 核心代码实现a. 配置与数据库连接管理 (config.py和database.py)# config.py from pydantic_settings import BaseSettings from pydantic import SecretStr class Settings(BaseSettings): db_host: str localhost db_port: int 5432 db_name: str mydb db_user: str mcp_user db_password: SecretStr # 允许查询的 schema 白名单 allowed_schemas: list[str] [public, analytics] # 默认查询行数限制 default_query_limit: int 100 class Config: env_file .env settings Settings()# database.py import asyncpg from typing import Optional from config import settings class DatabaseManager: _pool: Optional[asyncpg.Pool] None classmethod async def get_pool(cls) - asyncpg.Pool: if cls._pool is None: cls._pool await asyncpg.create_pool( hostsettings.db_host, portsettings.db_port, databasesettings.db_name, usersettings.db_user, passwordsettings.db_password.get_secret_value(), min_size1, max_size5, command_timeout60, # 查询超时设置 ) return cls._pool classmethod async def close(cls): if cls._pool: await cls._pool.close()b. 安全查询执行器 (query_executor.py)# query_executor.py import re from typing import Any, List, Dict from database import DatabaseManager from config import settings class SafeQueryExecutor: staticmethod def _validate_and_sanitize_query(query: str) - str: 1. 转换为小写便于检查。 2. 检查是否仅为 SELECT 查询防御性。 3. 确保查询中包含有效的 LIMIT 子句防止数据过载。 4. 检查 schema 白名单简单示例实际需更复杂解析。 query_lower query.strip().lower() # 只允许 SELECT 查询 if not query_lower.startswith(select): raise ValueError(只允许执行 SELECT 查询。) # 检查是否试图访问非白名单 schema简单正则匹配 for schema in settings.allowed_schemas: # 防止类似 information_schema 被意外禁止这里逻辑需根据实际情况调整 pass # 此处省略复杂的 SQL 解析建议使用 sqlparse 库进行更精准的分析 # 如果没有 LIMIT则添加默认限制 if limit not in query_lower: # 简单处理在末尾添加 LIMIT。注意如果查询已有子句如 ORDER BY这可能会破坏语法。 # 更健壮的做法是使用 SQL 解析器。 if query.rstrip().endswith(;): query query.rstrip()[:-1] query f{query} LIMIT {settings.default_query_limit}; elif limit in query_lower: # 如果已有 LIMIT检查其值是否过大 limit_match re.search(rlimit\s(\d), query_lower) if limit_match and int(limit_match.group(1)) 1000: raise ValueError(查询结果行数限制不能超过 1000 行。) return query staticmethod async def execute_query(query: str) - Dict[str, Any]: 执行经过验证和清理的查询并返回结果。 safe_query SafeQueryExecutor._validate_and_sanitize_query(query) pool await DatabaseManager.get_pool() async with pool.acquire() as conn: try: # 使用参数化查询防止注入如果查询有参数应通过args传入 rows await conn.fetch(safe_query) # 将结果转换为字典列表 result [dict(row) for row in rows] return { success: True, data: result, row_count: len(result), query_executed: safe_query } except Exception as e: return { success: False, error: str(e), query_attempted: safe_query }c. MCP Server 主程序 (main.py)# main.py from fastmcp import FastMCP from query_executor import SafeQueryExecutor import asyncio mcp FastMCP(Database Query Assistant) mcp.tool() async def query_database(sql_query: str) - dict: 在允许的 schema 上安全地执行一个 SELECT SQL 查询。 查询将自动添加行数限制如果未指定并且只读。 Args: sql_query (str): 要执行的 SELECT SQL 语句。请确保语法正确。 Returns: dict: 包含执行状态、数据如果成功或错误信息的字典。 result await SafeQueryExecutor.execute_query(sql_query) return result mcp.tool() async def get_table_schema(table_name: str) - dict: 获取指定表的 schema 信息列名、数据类型。 Args: table_name (str): 表名可包含 schema如 public.users。 Returns: dict: 表的列定义信息。 # 这是一个预定义的安全查询示例不直接拼接用户输入到 SQL 中。 # 实际应用中应对 table_name 进行严格的验证和转义或使用信息模式查询。 query f SELECT column_name, data_type, is_nullable FROM information_schema.columns WHERE table_name {table_name.split(.)[-1]} -- 简单提取表名部分 AND table_schema COALESCE(NULLIF(split_part({table_name}, ., 1), ), public) ORDER BY ordinal_position; return await SafeQueryExecutor.execute_query(query) if __name__ __main__: # 运行 MCP Server mcp.run(transportstdio)3. 配置与环境变量 (.env文件)DB_HOSTlocalhost DB_PORT5432 DB_NAMEyour_database DB_USERmcp_query_user DB_PASSWORDyour_strong_password_here ALLOWED_SCHEMASpublic,analytics DEFAULT_QUERY_LIMIT1004. 在 Cursor 中配置与使用配置 Cursor编辑 Cursor 的settings.json。{ mcpServers: { db-query-assistant: { command: python, args: [/path/to/your/mcp-db-assistant/main.py], env: { // 环境变量也可以在这里直接设置但更推荐使用 .env 文件 } } } }开始对话你“帮我查一下public.users表里最近7天活跃的用户数。”AI (Cursor)它会思考并可能调用get_table_schema先了解users表结构然后生成类似SELECT COUNT(*) FROM public.users WHERE last_active_date CURRENT_DATE - INTERVAL 7 days;的 SQL并通过query_database工具执行最后将结果以清晰的格式呈现给你。四、高级功能与优化方向自然语言转 SQL (NL2SQL)在 MCP Server 内集成一个小型 LLM如通过本地 Ollama 运行 SQLCoder将用户的自然语言描述直接转换为安全的 SQL 查询体验更无缝。查询结果后处理让 AI 对查询结果进行总结、可视化建议“生成一个柱状图来描述这个分布”或格式转换“将结果保存为 CSV”。动态工具发现根据数据库 schema 动态生成“查询表A”、“统计表B”等特定工具使 AI 的提示更精准。连接多数据源扩展 Server使其能同时连接 MySQL、Snowflake、BigQuery 等AI 可以帮你进行跨库查询。五、总结通过 MCP 协议将数据库查询能力赋予 AI 助手创造了一个强大的“数据协作者”。它不是替代专业的 BI 工具或数据工程师而是在开发、调试、即时探查场景下提供了一个无比便捷的交互界面。成功的关键在于“在便利性与安全性之间找到平衡”。通过本文的架构和代码示例你可以构建一个既强大又安全的数据库查询 MCP Server让你的数据对话变得自然、高效且可控。