rds_mcp 工具
一个基于 FastMCP 的 PostgreSQL 数据库操作工具，为 AI 助手提供安全的数据库访问能力。
功能特性
- 🤖 AI 友好 - 专为 AI 助手设计的数据库工具接口
- 🔐 安全执行 - 内置 SQL 安全检查，防止危险操作
- 📊 多操作支持 - 支持查询、插入、删除、更新等多种数据库操作
- 🛡️ 权限控制 - 限制危险关键字和操作类型，保护数据库安全
- 🔌 MCP 协议 - 基于 Model Context Protocol，可与 Claude、GPTs 等 AI 系统无缝集成
技术架构
AI 助手 (Claude/GPTS等)
↓ MCP 协议
PostgreSQL MCP 工具
↓ 安全检查
PostgreSQL 数据库安装运行
前置要求
- Python 3.8+
- PostgreSQL 数据库
- 支持 MCP 协议的 AI 助手（如 Claude Desktop）
启动服务
# 进入项目目录
cd examples/snippets/clients
# 使用 SSE 传输启动服务（适用于 Claude Desktop）
uv run server fastmcp_quickstart stdio环境配置
在项目根目录创建 .env 文件，配置数据库连接信息：
RDS_HOST=your_database_host
RDS_PORT=5432
RDS_USER=your_username
RDS_PASSWORD=your_password
RDS_DATABASE=your_database_nameAI 集成示例
MCP服务配置文件（需部署服务器）
{
"mcpServers":{
"rds_interact": {
"url":"http://服务器公网ip:port/sse",
"type":"sse/streamableHttp"
}
}
}AI 使用场景
AI 助手可以通过自然语言操作数据库：
用户: "帮我查询所有年龄大于18岁的用户"
AI: 使用 execute_sql 工具执行: SELECT * FROM users WHERE age > 18
用户: "在products表中添加一个新商品"
AI: 使用 execute_sql 工具执行: INSERT INTO products (name, price) VALUES ('新商品', 99.9)工具功能
execute_sql 工具
AI 助手通过此工具执行安全的数据库操作。
支持的操作类型：
select- 数据查询（只读）insert- 插入新数据delete- 删除数据update_table- 更新现有数据create_table- 创建新表alter_table- 修改表结构truncate_table- 清空表数据
安全特性：
- 自动拦截危险操作（DROP、GRANT 等）
- 查询操作限制为 SELECT 语句
- 完整的错误处理和日志记录
安全机制
禁止的关键字
['DROP', 'CREATE DATABASE', 'GRANT', 'REVOKE', 'ALTER USER', 'FLUSH', 'SHUTDOWN']操作限制
- SELECT 操作必须以
SELECT开头 - 所有操作都经过关键字过滤
- 数据库连接使用只读权限用户（推荐）
部署选项
1. SSE 传输（推荐用于 AI 集成）
mcp.run(transport='sse')2. HTTP 传输（Web 应用）
def handler(event, context):
return mcp.handle_http(event, context)3. Stdio 传输（命令行工具）
uv run server fastmcp_quickstart stdio响应格式
成功响应
{
"result": [...],
"row_count": 5,
"status": "success"
}错误响应
{
"error": "SQL 语法错误",
"status": "failed"
}最佳实践
对于 AI 开发者
- 清晰的提示词：在系统提示中说明可用的数据库操作
- 错误处理：教导 AI 如何解析和处理数据库错误
- 查询优化：鼓励 AI 使用限制条件避免大数据量查询
对于数据库管理员
- 权限分离：为 AI 工具创建专用数据库用户
- 备份策略：定期备份重要数据
- 监控日志：审查 AI 执行的 SQL 语句
故障排除
常见问题
- 连接失败：检查 .env 文件配置和网络连接
- 权限错误：验证数据库用户权限
- SQL 被拦截：检查是否包含禁止的关键字
日志调试
启用详细日志来跟踪 AI 的数据库操作：
import logging
logging.basicConfig(level=logging.DEBUG)扩展开发
您可以基于此模板扩展更多数据库工具：
@mcp.tool()
def get_table_schema(table_name: str) -> dict:
"""获取表结构信息 - 为 AI 提供数据库元数据"""
# 实现代码...