内容
数据库 MCP 服务器
一个 **只读** 的模型上下文协议（MCP）服务器，通过一组小而安全的工具，将 Oracle 数据库暴露给与 MCP 兼容的 AI 客户端（例如 Claude Desktop、VS Code MCP 扩展）。
设计为 **快速、安全、信用高效**：连接池、缓存的模式元数据、制表符分隔的输出，以及对行数/执行时间的严格限制。
---
## ✨ 功能
- 🔐 **环境变量/`.env` 中的凭证** — 不将机密提交到源代码。
- ♻️ **连接池** 按环境 — 重用 Oracle 会话。
- 🧹 **上下文管理的游标** — 错误时无泄漏。
- ⚡ **模式元数据缓存** (`lru_cache`) — 在重复的 `list_tables` / `describe_table` 调用中减少 MCP 代币/信用使用。
- 🛡️ **只读防护** — 仅允许 `SELECT` / `WITH`；阻止写入/DDL/PLSQL 关键字。
- ⏱️ **服务器端查询超时** 和 **行限制**。
- 🧾 **强化标识符验证** （修复空字符串绕过）。
- 📉 **紧凑的制表符分隔输出** 以最小化代币/信用消耗。
- 🧰 **诊断工具** (`cache_stats`) 以检查缓存命中/未命中。
---
## 📦 需求
- Python **3.10+**
- Oracle Instant Client（可选 — 启用 *thick* 模式；否则在 *thin* 模式下运行）
- Python 包：
- `oracledb`
- `mcp` （FastMCP）
- `python-dotenv` *(可选，用于 `.env` 支持)*
安装：
```bash
pip install oracledb mcp python-dotenv
```
---
## ⚙️ 配置
### 1. Oracle 客户端（可选 — thick 模式）
编辑 `server_v2.py` 顶部的常量：
```python
ORACLE_CLIENT_LIB = r"C:\path\to\instantclient_23_7" # 或设置为 None 以使用 thin 模式
```
### 2. 环境
源代码中预设了两种环境：`DEV` 和 `Another`。
> ⚠️ **安全注意：** 当前的 `server_v2.py` 仍然在 `ENVIRONMENTS` 字典中包含硬编码的凭证。在提交到任何地方之前，请将它们移动到环境变量/`.env` 中，并通过 `os.getenv(...)` 读取。
`.env` 示例：
```env
MCP_DEFAULT_ENV=DEV
MCP_LOG_LEVEL=INFO
# 调优
MCP_MAX_VAL_LEN=80
MCP_MAX_ROWS_CAP=500
MCP_QUERY_TIMEOUT_MS=30000
MCP_POOL_MIN=1
MCP_POOL_MAX=4
# （推荐）将数据库凭证移动到此处，而不是硬编码：
# DEV_USER=...
# DEV_PASSWORD=...
# DEV_DSN=...
```
### 3. 环境变量参考
| 变量 | 默认值 | 用途 |
|---|---|---|
| `MCP_DEFAULT_ENV` | `DEV` | 启动时选择的環境（`DEV` 或 `Another`）。 |
| `MCP_LOG_LEVEL` | `INFO` | Python 日志级别。 |
| `MCP_MAX_VAL_LEN` | `80` | 输出中每单元格的最大字符数（截断为 `…`）。 |
| `MCP_MAX_ROWS_CAP` | `500` | 任何查询的服务器端硬行限制。 |
| `MCP_QUERY_TIMEOUT_MS` | `30000` | 查询超时（毫秒）。 |
| `MCP_POOL_MIN` | `1` | 连接池中的最小连接数。 |
| `MCP_POOL_MAX` | `4` | 连接池中的最大连接数。 |
---
## ▶️ 运行
服务器通过 **stdio** （MCP 标准）进行通信。
```bash
python server_v2.py
```
您应该会看到启动日志，例如 `Oracle 客户端初始化（thick 模式）：...`。然后进程等待 stdin/stdout 上的 MCP 消息。
---
## 🔌 连接到 MCP 客户端
`mcp.json` 条目示例（Claude Desktop / VS Code MCP）：
```json
{
"mcpServers": {
"oracle-db": {
"command": "python",
"args": ["C:\\path\\to\\server_v2.py"],
"env": {
"MCP_DEFAULT_ENV": "DEV"
}
}
}
}
```
---
## 🧰 暴露的工具
所有工具都返回 **紧凑的制表符分隔文本** 以保持令牌使用量低。
| 工具 | 描述 |
|---|---|
| `get_current_environment()` | 显示当前活动的数据库环境。 |
| `switch_environment(env)` | 在 `DEV` 和 `Another` 之间切换。 |
| `list_tables(search="", owner="")` | 列出表格，可选名称筛选/所有者。 **已缓存。** |
| `describe_table(table_name, owner="")` | 显示表格的列、主键、外键。 **已缓存。** |
| `get_distinct_values(table_name, column_name, top_n=20, owner="")` | 列的 distinct 值 + 计数。 |
| `run_query(sql, max_rows=20)` | 执行只读 `SELECT` / `WITH` |
| `cache_stats()` | 诊断：缓存命中/未命中计数器。 |
### `run_query` 的安全规则
- 必须以 `SELECT` 或 `WITH` 开头。
- 阻止：`INSERT, UPDATE, DELETE, DROP, ALTER, CREATE, TRUNCATE, MERGE, GRANT, REVOKE, EXEC, EXECUTE, CALL, BEGIN, DECLARE, COMMIT, ROLLBACK, SAVEPOINT`。
- `max_rows` 被限制为 `MCP_MAX_ROWS_CAP`。
- 强制执行 `MCP_QUERY_TIMEOUT_MS`。
---
## 📖 使用示例
以下是与服务器交互的实用示例。提示是您在 MCP 客户端（例如 Claude Desktop、VS Code）中输入的内容；*工具调用* 行显示客户端实际发送的内容。
### 1. 检查活动环境
**提示：**
> 我当前连接到哪个 Oracle 环境？
**工具调用：**
```json
{ "tool": "get_current_environment", "arguments": {} }
```
---
### 2. 切换环境
**提示：**
> 切换到 DEV。
**工具调用：**
```json
{ "tool": "switch_environment", "arguments": { "env": "DEV" } }
```
---
### 3. 发现表格
**提示：**
> 列出所有名称中包含"CUSTOMER"的表格。
**工具调用：**
```json
{ "tool": "list_tables", "arguments": { "search": "CUSTOMER" } }
```
> 💡 重复调用相同参数的调用将从缓存中提供服务 — 调用 `cache_stats` 以验证。
---
### 4. 描述表格
**提示：**
> 描述CUSTOMER表格。
**工具调用：**
```json
{ "tool": "describe_table", "arguments": { "table_name": "CUSTOMER" } }
```
---
### 5. 探索列值
**提示：**
> CUSTOMER 表中 STATUS 列的前 5 个值是什么？它们出现的频率是多少？
**工具调用：**
```json
{
"tool": "get_distinct_values",
"arguments": {
"table_name": "CUSTOMER",
"column_name": "STATUS",
"top_n": 5
}
}
```
---
### 6. 运行只读查询
**提示：**
> 2026 年到目前为止，每月有多少客户注册？给我前 10 个工具调用：**
```json
{
"run_queryarguments": {
"sql": "SELECT TO_CHAR(CREATED_AT,'YYYY-MM') AS MONTH, COUNT(*) AS SIGNUPS FROM CUSTOMER WHERE CREATED_AT >= DATE '2026-01-01' GROUP BY TO_CHAR(CREATED_AT,'YYYY-MM') ORDER",
"": 10
}
}
```
---
### 7. 使用 CTE 查询（允许使用 `WITH`）
**提示：**
> 显示 5 个订单最多的客户。
**工具调用：**
```json
{
"tool":_query",
"arguments": {
"sql": "WITH order_counts AS (SELECT CUSTOMER_ID, COUNT(*) AS N FROM ORDERS GROUP BY CUSTOMER_ID) SELECT c.CUSTOMER_ID, c.FIRST_NAME, c.LAST_NAME, oc.N AS ORDER_COUNT FROM order_counts oc JOIN CUSTOMER c ON c.CUSTOMER_ID = oc.CUSTOMER_ID ORDER BY oc.N DESC FETCH FIRST 5 ROWS ONLY",
"max_rows": 5
}
}
```
---
### 8. 尝试写入查询时会发生什么
**提示：**
> 删除不活动的客户。
调用：**
```json
{
"tool":",
"arguments": { "sql": "DELETEER WHERE STATUS = 'INACTIVE'" }
}
```
**示例输出：**
```
错误：只允许使用只读 SELECT / WITH 语句。
检测到阻止关键字：DELETE
```
---
### 9. 检查缓存有效性
**提示：**
> 元数据缓存表现如何？
**工具调用：**
```json
{ "tool": "cache_stats", "arguments": {} }
```
**示例输出：**
```
list_tables → hits=14 misses=3 size=3/128
describe_table → hits=42 misses=8 size=8/256
```
> 高 `hits` 与 `misses` 的比例 = 在 Oracle 往返和 MCP 代币/信用方面节省大量成本。
---
### 10. 完整的对话示例
MCP 客户端内部的现实多轮流程：
> **您：** 我在 DEV 上 — 哪些表格的名称包含"ORDER"？
> **助手** *(调用 `list_tables(search="ORDER")`)* → 返回 4 个表格。
>
> **您：** 描述 `ORDERS`。
> **助手** *(调用 `describe_table(table_name="ORDERS")`)* → 返回列 + 主键/外键。
>
> **您：** `ORDER_STATUS` 的可能值是什么？
> **助手** *(调用 `get_distinct_values(table_name="ORDERS", column_name="ORDER_STATUS")`*。
>
> **您：** 显示本月到目前为止的最后 20 个已发货订单。
> **助手** *(调用 `run_query`，带有 `SELECT ... WHERE ORDER_STATUS='SHIPPED' AND ORDER_DATE >= TRUNC(SYSDATE,'MM') ORDER BY ORDER_DATE DESC FETCH FIRST 20 ROWS ONLY`)*。
由于 `list_tables` 和 `describe_table` 已缓存，因此后续关于同一模式的问题在后续回合中几乎无需成本。
---
## 💡 信用/代币优化
该服务器已调优以最小化 AI 端的信用使用量：
- **缓存元数据** — 重复的 `list_tables` / `describe_table` 调用不会重新命中 Oracle *并且* 快速返回给模型。
- **制表符分隔、截断单元格** — 稠密输出 = 比 JSON 更少的代币。
- **硬行限制** — 防止意外的巨大结果集。
- **紧凑的环境/描述输出** — 无冗长的描述。
在会话期间调用 `cache_stats` 以检查有效性。
要重置缓存（例如，在模式更改后），请重启服务器或调用内部 `_clear_metadata_cache()` （默认情况下不作为工具公开）。
---
## 🧪 本地测试
在没有 MCP 客户端的情况下进行快速健全检查：
```bash
python -c "import oracledb; print(oracledb.version)"
python server_v2.py # 然后通过 stdin 发送 MCP JSON-RPC 消息
```
对于端到端测试，在 MCP 客户端的 `mcp.json` 中注册服务器，并从客户端 UI 调用工具。
---
## 🗂️ 项目布局
```
.
├── server_v2.py # MCP 服务器（此项目）
└── README.md
```
