从“自然语言问数“到“AI 自动调优“：我给金仓写了一个 MCP Server

一篇让金仓数据库"开口说话"的实战文。100 行 Python + Anthropic 官方 mcp SDK，做出一个 6 工具 / 3 道安全闸的 MCP Server，让任何 AI 客户端（Claude Desktop / Claude Code / Cursor）能用中文问金仓——不仅能查数据，还能让 AI 看执行计划、自动给出索引 DDL 建议。代码、协议往返、攻击拦截、自动调优全过程实测，环境 KingbaseES V009R001C010。

一、为什么"AI + 数据库"是国产数据库的下一个护城河

国产数据库的 SQL 兼容性、TPC-C 跑分已经卷得差不多了，下一站的差距其实在生态——尤其是"AI 原生访问数据库"的能力。

2025 年 Anthropic 推出的 MCP（Model Context Protocol） 把这件事的解法定了型：它是一个开放协议，让 AI 应用与外部工具之间用标准格式通信。你把数据库能力包装成一个 MCP Server，任何 MCP 兼容的客户端接上来（Claude Desktop、Cline、Cursor、Claude Code……），就能让大模型自己决定"该查哪张表、写什么 SQL"，然后用大白话把结果讲出来。

金仓是 PG 内核——这意味着我们不需要等官方出 MCP Server，用 psycopg2 + 官方 mcp SDK 就能自己手撸一个，100 行核心代码就能跑通。

本文的差异化

中文圈已有的金仓 MCP 实践（如 CSDN 上的若干文章）大多停在"4 工具 + 简单 SELECT"的水平。本文在此基础上多走两步：

技术栈：

二、架构：100 行代码的"翻译官"

整条链路只有三个角色：

[AI 客户端] [MCP Server] [KingbaseES]
Claude Code stdio kingbase_mcp.py SQL ai_ro 只读账号
Claude Desktop ←───── 6 个 tool ←──── t_order 100 万行
Cursor 3 道安全闸

MCP Server 向 AI 暴露 6 个工具（Tool），大模型像调用函数一样调用它们：

用 Anthropic 官方 mcp SDK 的 FastMCP，一个装饰器就把普通 Python 函数变成 MCP 工具：

from mcp.server.fastmcp import FastMCP
mcp = FastMCP("kingbase")

@mcp.tool()
def run_query(sql: str) –> str:
"""执行一条只读 SELECT 查询，返回最多 50 行结果。"""
... # 三道安全闸 + psycopg2 执行

if name == "main":
mcp.run() # stdio 传输

关键认知：工具函数的 docstring 不是注释，是给大模型看的说明书——Agent 靠它判断什么时候该调哪个工具。这是写 MCP Server 和写普通后端接口最不一样的地方：你的注释第一次有了"读者是 AI"。

三、三道安全闸：把 AI 关进笼子里

Demo 谁都能跑通，能不能上生产，全看安全。一个能连生产库、还听大模型指挥的服务，如果只做到"能查"，那不是功能，是事故。本文给 Server 设计了 三道互相独立、任意一道都能单独兜底 的闸门。

闸 1：数据库账号本身只读

MCP Server 连库用的 ai_ro 账号只被 GRANT SELECT。建账号的脚本（step4_mcp/setup_ai_ro.py）：

CREATE USER ai_ro WITH PASSWORD 'Kingbase@2026';
GRANT USAGE ON SCHEMA public TO ai_ro;
GRANT SELECT ON ALL TABLES IN SCHEMA public TO ai_ro;
ALTER DEFAULT PRIVILEGES IN SCHEMA public
GRANT SELECT ON TABLES TO ai_ro; — 未来新建的表也自动授权

实测验证（脚本里直接拿 ai_ro 连库执行 DELETE）：

→ 试图 DELETE FROM t_order（应被权限层挡掉）:
✓ 闸 1 生效：对表 t_order 权限不够

→ 试图 DROP TABLE（应被权限层挡掉）:
✓ 闸 1 生效：必须是表 t_order 的属主

这是最硬的一道——就算前面两道全被绕过，数据库自己也不会让 AI 写入。

闸 2：会话级只读 + 超时

代码里每次开连接都强制：

c.set_session(readonly=True) # 会话级只读
options=f"-c statement_timeout=5000" # 5 秒超时，防 AI 写出慢查询拖垮库

闸 3：应用层 SQL 白名单

run_query / explain_plan / suggest_index 都共用一道正则白名单：
_WRITE_KEYWORDS = re.compile(
r"\\b(insert|update|delete|drop|truncate|alter|create|grant|revoke|"
r"merge|vacuum|reindex|call|do|copy)\\b", re.IGNORECASE)
_MULTI_STMT = re.compile(r";\\s*\\S") # 分号后还有内容 → 多语句注入

def _gate_sql(sql):
s = sql.strip().rstrip(";").strip()
if _MULTI_STMT.search(s):
return False, "已拒绝：只允许单条语句"
head = s.split(None, 1)[0].upper()
if head not in ("SELECT", "WITH"):
return False, f"已拒绝：只允许 SELECT/WITH（实际开头: {head}）"
if _WRITE_KEYWORDS.search(s):
return False, "已拒绝：检测到写操作关键字"
return True, "OK"

安全设计哲学：三道闸冗余但不多余。应用层白名单可能被更刁钻的 SQL 绕过，但会话只读会挡；会话只读万一失效，账号权限还在。安全不赌"我的正则天衣无缝"，安全赌"攻破一层还有下一层"。

四、实战 1：协议握手 + 6 工具跑通

光说不算，我用官方 mcp SDK 写了个真正的 MCP 协议客户端去连它（step4_mcp/test_client.py），走完整的 initialize → tools/list → tools/call 握手：

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

params = StdioServerParameters(command="python", args=[SERVER_PATH])
async with stdio_client(params) as (r, w):
async with ClientSession(r, w) as session:
init = await session.initialize()
print(init.protocolVersion, init.serverInfo.name) # 2025-11-25 kingbase
tools = await session.list_tools() # 拉到 6 个 tool
await session.call_tool("db_info", {}) # 真实调用

握手 + 工具发现

✓ 协议版本: 2025–11–25
✓ Server 名: kingbase
✓ Server 版本: 1.27.2

db_info 返回数据库的身份信息：版本、兼容模式、当前连接账号、连接数。
list_tables 列出当前库下所有用户表 + 每张表的估算行数。
describe_table 返回某张表的结构（字段名/类型/可空/默认值）+ 3 行样例数据。
run_query 执行一条只读 SELECT/WITH 查询，返回最多 50 行结果 + 总匹配行数。
explain_plan 对一条 SELECT 跑 EXPLAIN ANALYZE，返回执行计划原文。
suggest_index 基于 EXPLAIN ANALYZE 自动给出索引建议。

工具调用：db_info

{
"version": "KingbaseES V009R001C010",
"database_mode": "mysql",
"current_user": "ai_ro",
"database": "test",
"active_connections": 12,
"readonly_account": "ai_ro"
}

后端确实是金仓 V009R001C010，连接身份是只读账号 ai_ro——验明正身。

工具调用：list_tables + describe_table

{
"tables": [
{"schema": "public", "table_name": "t_order", "approx_rows": 1000000},
{"schema": "public", "table_name": "t_department","approx_rows": 3},
{"schema": "public", "table_name": "t_employee", "approx_rows": 4}
]
}

describe_table('t_order') 返回 8 个字段的完整定义 + 3 行样例——大模型靠这个才能写出字段名拼对的 SQL。

五、实战 2：5 种越权攻击全部拦截

我把 5 种最经典的攻击手法扔给 run_query，全军覆没：

{"rejected": "已拒绝：只允许 SELECT/WITH 查询（实际开头: DELETE）",
"sql": "DELETE FROM t_order WHERE id = 1"}

{"rejected": "已拒绝：只允许单条语句（检测到分号后还有内容）",
"sql": "SELECT 1; DROP TABLE t_order"}

{"rejected": "已拒绝：检测到写操作关键字（insert/update/delete/…）",
"sql": "SELECT (DELETE FROM t_order RETURNING 1)"}

注意：闸 3 已经全部挡住，但就算某个被精心构造的 SQL 绕过了正则，闸 2（会话只读）会兜底；就算会话只读也失效，**闸 1（账号级只读）**是金仓权限层的硬拦截，永远生效。

六、实战 3：差异化核心 —— AI 自动给索引建议

这是本文最重要的差异化。前 4 个工具（list/describe/run_query/db_info）解决"AI 能查数据"，但真正能帮到 DBA 的是让 AI 自己看懂执行计划、给出优化建议。

6.1 explain_plan：返回执行计划原文

让 AI 调用 explain_plan，传入一条 SELECT，就能拿到 EXPLAIN ANALYZE 的完整文本：

@mcp.tool()
def explain_plan(sql: str) –> str:
"""对一条 SELECT 跑 EXPLAIN ANALYZE，返回执行计划原文。"""
ok, reason = _gate_sql(sql) # 闸 3
if not ok:
return json.dumps({"rejected": reason})
with _conn() as c:
with c.cursor() as cur:
cur.execute("EXPLAIN (ANALYZE, BUFFERS, FORMAT TEXT) " + sql)
plan = "\\n".join(r[0] for r in cur.fetchall())
return json.dumps({"sql": sql, "plan": plan}, ensure_ascii=False, indent=2)

对一条已经加过索引的 SQL（同库已有业务索引）：

SELECT * FROM t_order
WHERE status = 5 AND create_time BETWEEN '2025-06-01' AND '2025-09-30'
ORDER BY amount DESC LIMIT 50

返回：

Limit (cost=15273.72..15273.85 ...) (actual time=39.973..39.980 ...)
–> Sort (cost=15273.72..15370.08 ...) (actual time=39.971..39.975 ...)
Sort Key: amount DESC NULLS LAST
–> Bitmap Heap Scan on t_order (...rows=37351...)
–> Bitmap Index Scan on idx_order_status_time_amt

6.2 suggest_index：自动 DDL 推荐

更进一步的工具——读 EXPLAIN，自己给出索引建议：

@mcp.tool()
def suggest_index(sql: str) -> str:
"""基于 EXPLAIN ANALYZE 自动给出索引建议。"""
… # 内部调 _analyze_plan

诊断逻辑（启发式规则）：

场景 A：已有索引的 SQL

{
"needs_index": false,
"diagnosis": ["已使用索引扫描，无需新增索引"],
"suggested_ddl": []
}

AI 自己判断出"不需要加索引"——这种克制比无脑推荐强太多了。

场景 B：新场景，没建过索引的 SQL

这是更典型的实战场景。给 AI 一条新查询：

SELECT * FROM t_order
WHERE user_id = 12345 AND city = '北京'

执行计划（先扫 100 万行 + 扔掉 33 万行 + 1390 ms）：

Gather (cost=1000.00..19305.20 ...) (actual time=8.787..1389.518 ...)
Workers Planned: 2 Workers Launched: 2
–> Parallel Seq Scan on t_order (...rows=1 loops=3)
Filter: ((user_id = 12345) AND ((city)::text = '北京'))
Rows Removed by Filter: 333332
Execution Time: 1389.956 ms

AI 调用 suggest_index，自动返回：

{
"needs_index": true,
"target_table": "t_order",
"filter_columns": ["user_id", "city"],
"sort_column": null,
"diagnosis": [
"Seq Scan 扔掉了 333332 行（>1000），强烈建议加索引",
"建议复合索引列顺序：['user_id', 'city']"
],
"suggested_ddl": [
"CREATE INDEX idx_t_order_user_id_city ON t_order(user_id, city);"
]
}

AI 不仅识别出"该加索引"，还自动生成了正确的复合索引 DDL——列顺序遵循"等值在前、范围在后"的最优原则。DBA 拿到这个 DDL 直接 CREATE INDEX 就完事了。

以前 DBA 看完 EXPLAIN 才能写出来的索引 DDL，现在 AI 在工具调用层就帮你生成好了。

七、接 Claude Code：3 行配置上工位

跑通协议只是验证可用。要把它真实插到工位上，最便捷的方式是接 Claude Code——程序员写代码时就在用的 IDE。

7.1 安装依赖 + 启动账号

pip install "mcp[cli]" psycopg2–binary
python step4_mcp/setup_ai_ro.py # 创建 ai_ro 只读账号（一次性）

7.2 注册 MCP Server 到 Claude Code

在任意目录执行（一次性，永久生效）：

claude mcp add kingbase ––env KES_USER=ai_ro ––env KES_PASS=Kingbase@2026 –– python F:/——文章撰写——/金仓/kes–python–demo/step4_mcp/kingbase_mcp.py

或者用项目级配置（.mcp.json）：

{
"mcpServers": {
"kingbase": {
"command": "python",
"args": ["F:/——文章撰写——/金仓/kes-python-demo/step4_mcp/kingbase_mcp.py"],
"env": {
"KES_HOST": "localhost",
"KES_PORT": "54321",
"KES_DB": "test",
"KES_USER": "ai_ro",
"KES_PASS": "Kingbase@2026"
}
}
}
}

7.3 重启 Claude Code，开始问数

之后对着 Claude Code 直接用中文问：

用户：连上的金仓数据库里有哪些业务表？t_order表里有多少行订单？
AI ：连上的金仓数据库（public schema）里共有 4 张业务表：
┌──────────────┬─────────────┐
│ 表名 │ 估算行数 │
├──────────────┼─────────────┤
│ t_order │ 约 100 万 │
├──────────────┼─────────────┤
│ t_department │ —（未统计） │
├──────────────┼─────────────┤
│ t_employee │ —（未统计） │
├──────────────┼─────────────┤
│ t_null_test │ —（未统计） │
└──────────────┴─────────────┘
t_order 实际订单数：1,000,000 行（精确 COUNT，与估算一致）。
/////////////////////////////////////////////////////////////////////////////
用户：哪个城市的订单金额总和最高？前三名分别是多少？

AI ：订单金额总和最高的城市：深圳（约 10.02 亿元）。

金额前三名：

┌──────┬──────┬──────────────────┬─────────┐
│ 排名 │ 城市 │ 订单总金额 │ 订单数 │
├──────┼──────┼──────────────────┼─────────┤
│ 1 │ 深圳 │ 1,001,825,669.18 │ 200,122 │
├──────┼──────┼──────────────────┼─────────┤
│ 2 │ 上海 │ 1,000,376,058.49 │ 199,823 │
├──────┼──────┼──────────────────┼─────────┤
│ 3 │ 北京 │ 999,781,765.16 │ 199,968 │
└──────┴──────┴──────────────────┴─────────┘

补充：共 5 个城市（另有杭州、广州），总金额彼此咬得很紧，都在 10 亿元上下，差距仅约千分之三。

这就是 MCP 的价值：Server 写一次，所有 AI 客户端通用。把数据库的"嘴"装上，谁来都能问。

八、可回溯：AI 触达数据库的每次调用都留痕

生产环境还有个绕不开的问题：出了事，能不能查清是谁、什么时候、让 AI 对数据库做了什么。我给每个工具加了一行审计日志：

logging.basicConfig(filename="mcp_calls.log", ...)
log = logging.getLogger("kes_mcp")

@mcp.tool()
def run_query(sql: str) –> str:
log.info("run_query sql=%s", sql)
...

跑完上面的测试后，mcp_calls.log 完整记录了每一次调用——三条正常业务查询，以及五条被拒的攻击尝试：

2026–08–08 ... | INFO | run_query sql=SELECT status, count(*) ...
2026–08–08 ... | WARNING | REJECTED sql=DELETE FROM t_order ...
2026–08–08 ... | WARNING | REJECTED sql=DROP TABLE t_order ...
2026–08–08 ... | WARNING | REJECTED sql=SELECT 1; DROP TABLE t_order
2026–08–08 ... | INFO | explain_plan sql=SELECT * FROM t_order ...
2026–08–08 ... | INFO | suggest_index sql=SELECT * FROM t_order WHERE user_id ...

被拦截的攻击也照样留痕——这正是安全审计最想要的：不仅记成功，更要记下"有人试图越权"。这只是 40 行雏形，生产上可以接统一日志平台、加调用方身份、做异常告警，但**“每一次 AI 触达数据库都可回溯”**这个原则，从第一版就得立住。

九、心得 + 国产化意义

MCP 是开放协议，谁都能用——金仓、达梦、OceanBase、PG 都一样接。真正决定能不能上生产的，是国产数据库自己的权限粒度和审计能力。本文三道闸里最硬的就是金仓的 GRANT SELECT——这是金仓作为国产数据库在权限层就顶住了 DELETE/DROP 的能力证明。AI 时代的数据库选型，"能不能精细授权 + 能不能完整审计"比"支持几种 SQL 方言"重要得多。

中文圈已有的 MCP 实践大多停在"自然语言翻译成 SQL 查数据"——这只是把 AI 当翻译官用。本文加的 explain_plan 和 suggest_index 把 AI 升级成"DBA 顾问"——AI 不仅查数据，还看执行计划、给优化建议。这才是 AI Agent 真正能帮 DBA 减负的方向。下一站可以继续扩展：compare_plans（前后执行计划对比）、profile_workload（拉 KWR 报告分析 TOP SQL）……MCP 的工具粒度让这件事可以无限叠加。

工具函数的 docstring 不是给程序员看的，是给大模型看的说明书——大模型靠它判断什么时候该调哪个工具。这意味着：

• 描述要写"动词 + 名词 + 限制条件"，比如"执行一条只读 SELECT，最多返回 50 行"——大模型看到"只读"就知道不能用来改数据。
• 参数说明要具体，比如"不要带末尾分号"——大模型才会主动去掉分号。
• 错误返回也是上下文，比如 {"rejected": "…原因…"}——大模型看到拒绝原因会自己改写 SQL 重试，或者主动告知用户。

这是 AI 时代编程范式的一次小但根本的转变。写给 AI 看的代码，要像写给一个聪明但不了解你系统的同事看一样。

附：完整代码结构

kes-python-demo/
└── step4_mcp/
├── setup_ai_ro.py # 创建 ai_ro 只读账号（闸 1）
├── kingbase_mcp.py # MCP Server（6 工具 + 3 道安全闸）
├── test_client.py # 协议握手 + 6 工具 + 5 攻击测试
├── demo_suggest.py # suggest_index 自动索引推荐 demo
└── mcp_calls.log # 审计日志

让 AI 能查金仓只是起点，让 AI 安全可控地查金仓、给出 DBA 级的优化建议，才是 MCP Server 真正的工程价值。