内容
# 数据库助手 MCP 服务器
一个用于数据库探索和查询的只读 MCP（模型上下文协议）服务器。将任何与 MCP 兼容的客户端（AWS Kiro、Claude Desktop 等）连接到您的 PostgreSQL 或 MySQL 数据库，并探索模式、运行查询、从自然语言生成 SQL，并将结果导出为 CSV。
## 功能
- **模式发现** — 列出数据库、模式、表、列、约束、索引和外键关系
- **安全查询执行** — 基于 AST 的 SQL 验证确保仅运行只读查询
- **自然语言转 SQL** — 用普通英语描述您想要的查询，获取一个有效的 SQL 查询
- **CSV 导出** — 将任何查询结果导出为 CSV 文件
- **查询解释** — 运行 EXPLAIN ANALYZE 以了解查询性能
- **分页** — 自动分页显示大型结果（每页 100 行）
- **审计日志** — 每个查询都记录有时间戳、执行时间和行数
## 工具
| 工具 | 描述 |
|------|-------------|
| `list_databases` | 列出服务器上的所有数据库 |
| `list_schemas` | 列出模式（按允许列表过滤） |
| `list_tables` | 模式中的表，包含类型和行数 |
| `describe_table` | 列、类型、约束和索引 |
| `describe_relationships` | 外键树（传入和传出） |
| `sample_data` | 表的快速 N 行预览 |
| `validate_sql` | 在运行前检查查询是否安全 |
| `run_sql` | 执行验证的只读查询 |
| `explain_query` | EXPLAIN ANALYZE，包含执行计划 |
| `generate_sql` | 自然语言 → SQL（LLM 支持） |
| `export_csv` | 执行查询并将结果保存为 CSV |
## 安全
该服务器强制实施严格的只读访问：
- **基于 AST 的验证** — 使用 [sqlglot](https://github.com/tobymao/sqlglot) 将 SQL 解析为抽象语法树，而不是正则表达式
- **阻止操作** — INSERT、UPDATE、DELETE、DROP、ALTER、TRUNCATE、CREATE、GRANT、REVOKE、COPY、CALL
- **单语句查询** — 拒绝多语句查询（`;` 分隔）
- **危险函数阻止** — `pg_read_file`、`lo_export`、`dblink`、`LOAD_FILE` 等
- **自动 LIMIT** — 没有 LIMIT 的查询会注入一个（默认：1000 行最大）
- **查询超时** — 在数据库级别强制执行（默认：10 秒）
- **模式允许列表** — 仅限制访问特定模式
- **生成 SQL 重新验证** — LLM 生成的查询通过相同的验证器
## 先决条件
- Python 3.10+
- [uv](https://docs.astral.sh/uv/) 包管理器
- PostgreSQL 或 MySQL 数据库
## 安装
1. **克隆并安装依赖：**
```bash
cd /path/to/RDS
uv sync
```
2. **配置环境：**
```bash
cp .env.example .env
# 编辑 .env 文件，填写您的数据库凭证
```
3. **初始化测试数据库（可选）：**
```bash
psql -U postgres -f seed.sql
```
这会创建一个包含 50 张表和约 25,000 行数据的 `ecommerce` 数据库，涵盖客户、产品、订单、分析、支持等。
## 配置
设置以下环境变量（通过 `.env` 文件或系统环境）：
| 变量 | 必需 | 默认值 | 描述 |
|----------|----------|---------|-------------|
| `DB_HOST` | 是 | — | 数据库主机名或 RDS 端点 |
| `DB_PORT` | 否 | `5432` | 数据库端口 |
| `DB_NAME` | 是 | — | 数据库名称 |
| `DB_USER` | 是 | — | 数据库用户（使用只读用户） |
| `DB_PASSWORD` | 是 | — | 数据库密码 |
| `DB_TYPE` | 否 | `postgresql` | `postgresql` 或 `mysql` |
| `ALLOWED_SCHEMAS` | 否 | （所有） | 以逗号分隔的模式允许列表 |
| `QUERY_TIMEOUT` | 否 | `10` | 最大查询执行时间（秒） |
| `MAX_ROWS` | 否 | `1000` | 每查询返回的最大行数 |
| `CSV_EXPORT_DIR` | 否 | `/tmp/db_exports` | CSV 导出目录 |
| `OPENAI_BASE_URL` | 否 | — | LLM API 端点（用于 `generate_sql`） |
| `OPENAI_API_KEY` | 否 | — | LLM API 密钥（用于 `generate_sql`） |
| `MODEL` | 否 | `bedrock.claude-sonnet-4-6` | LLM 模型 ID |
## 使用
### 与 AWS Kiro 配合使用
在您的 Kiro MCP 配置中添加：
```json
{
"mcpServers": {
"database-assistant": {
"command": "/opt/homebrew/bin/uv",
"args": ["run", "--directory", "/path/to/database_mcp_server", "mcp_server.py"],
"env": {
"DB_HOST": "your-rds-endpoint.rds.amazonaws.com",
"DB_PORT": "5432",
"DB_NAME": "ecommerce",
"DB_USER": "readonly_user",
"DB_PASSWORD": "your_password",
"DB_TYPE": "postgresql",
"ALLOWED_SCHEMAS": "store"
}
}
}
}
```
### 与 Claude Desktop 配合使用
在 `claude_desktop_config.json` 中添加：
```json
{
"mcpServers": {
"database-assistant": {
"command": "uv",
"args": ["run", "--directory", "/path/to/RDS", "mcp_server.py"]
}
}
}
```
### 使用 MCP Inspector（用于测试）
```bash
uv run mcp dev mcp_server.py
```
### 直接 stdio（用于开发）
```bash
uv run mcp_server.py
```
## 示例工作流程
### 探索数据库模式
```
用户：这个数据库有哪些表？
→ list_tables(schema="store")
用户：告诉我关于 orders 表的信息
→ describe_table(table="store.orders")
用户：什么与 orders 相关？
→ describe_relationships(table="store.orders")
```
### 使用自然语言查询
```
用户：显示 2024 年收入最高的 10 位客户
→ generate_sql(question="top 10 customers by total order revenue in 2024")
→ validate_sql(query="SELECT ...")
→ run_sql(query="SELECT ...")
```
### 导出数据
```
用户：导出上个月的所有订单为 CSV
→ generate_sql(question="all orders created in the last 30 days with customer name and total")
→ export_csv(query="SELECT ...")
→ 返回：/tmp/db_exports/export_20240715_143022_a1b2c3d4.csv
```
## 架构
```
MCP 客户端（Kiro / Claude Desktop）
│
▼ （stdio）
┌─────────────────────────────┐
│ mcp_server.py │ FastMCP 服务器，11 种工具
├─────────────────────────────┤
│ sql/validator.py │ 基于 AST 的安全验证（sqlglot）
│ sql/generator.py │ NL→SQL 通过 OpenAI 兼容的 LLM
│ db/schema.py │ 模式 introspection 服务
│ db/connection.py │ 异步连接池（asyncpg/aiomysql）
│ export/csv_export.py │ CSV 文件生成
│ config.py │ 基于环境变量的配置
└─────────────────────────────┘
│
▼
PostgreSQL / MySQL / AWS RDS
```
## 项目结构
```
.
├── mcp_server.py # MCP 服务器入口点，包含工具定义
├── config.py # 从环境变量加载配置
├── db/
│ ├── connection.py # 异步连接管理器，带连接池
│ └── schema.py # 模式 introspection 服务
├── sql/
│ ├── validator.py # SQL 验证（sqlglot AST）
│ └── generator.py # LLM 基础的 SQL 生成
├── export/
│ └── csv_export.py # CSV 导出实用程序
├── seed.sql # 示例数据库（50 张表，25k+ 行）
├── pyproject.toml # 依赖项和项目元数据
├── .env.example # 环境变量模板
└── .gitignore
```
## 创建只读数据库用户
在生产环境中，请创建专用的只读用户：
```sql
-- PostgreSQL
CREATE USER readonly_user WITH PASSWORD 'secure_password';
GRANT CONNECT ON DATABASE ecommerce TO readonly_user;
GRANT USAGE ON SCHEMA store TO readonly_user;
GRANT SELECT ON ALL TABLES IN SCHEMA store TO readonly_user;
ALTER DEFAULT PRIVILEGES IN SCHEMA store GRANT SELECT ON TABLES TO readonly_user;
```
```sql
-- MySQL
CREATE USER 'readonly_user'@'%' IDENTIFIED BY 'secure_password';
GRANT SELECT ON ecommerce.* TO 'readonly_user'@'%';
FLUSH PRIVILEGES;
```
## 依赖项
- [mcp](https://pypi.org/project/mcp/) — 模型上下文协议 SDK
- [sqlglot](https://github.com/tobymao/sqlglot) — 用于验证的 SQL 解析器
- [asyncpg](https://github.com/MagicStack/asyncpg) — 异步 PostgreSQL 驱动程序
- [aiomysql](https://github.com/aio-libs/aiomysql) — 异步 MySQL 驱动程序
- [openai](https://github.com/openai/openai-python) — LLM API 客户端（用于 SQL 生成）
- [python-dotenv](https://github.com/theskumar/python-dotenv) — 环境变量加载
- [pydantic](https://github.com/pydantic/pydantic) — 数据验证
## 许可证
MIT
