摘要

想让 AI 助手直接操作数据库的人,迟早要面对两个问题：怎么连上,以及连上之后凭什么放心。这篇文章记录我在一台国产化环境（鲲鹏 + 麒麟 V10 + KingbaseES V9）上把 KES MCP Server 从零装到能用的全过程。先把结论放在前面：

1. 连得上。从克隆官方仓库到 Claude Code 完成 MCP 握手,前后大概十条命令。协议版本协商到 2025-06-18,服务端注册了 10 个工具,用只读账号发的第一条查询就拿到了真实结果。
2. 能干活。一条 30 万行表上的慢查询,AI 靠假设索引仿真给出了优化方案,全程没建任何真实索引。仿真预估优化器成本从 6212.5 降到 19.7；我随后真建了索引验证,实测成本 20.05,跟预估差了约 1.9%。执行时间从 24.540 ms（3 次采样取中位数）掉到 0.030 ms,差不多 818 倍。
3. 敢放手。受限模式下我把 INSERT、UPDATE、DELETE、DROP、DDL、多语句注入六类危险操作挨个试了一遍,全部在 Server 层被拦下。后来我故意把服务切到非受限模式再试,写入被数据库层的账号权限挡住了。三层防线各拦各的,每一层都有报错截图为证。

测试环境声明

项目	配置
云主机	华为云 ECS（4 vCPU / 7.1 GiB 内存 / 39 GB 系统盘）
CPU	HiSilicon ARMv8（鲲鹏）,aarch64
操作系统	Kylin Linux Advanced Server V10 (Tercel),内核 4.19.90-17.5.ky10.aarch64
数据库	KingbaseES V009R001C010（KES V9,单机部署,端口 54321）
MCP Server	kingbase-mcp 0.3.0（官方 Gitee 仓库）,serverInfo 上报组件版本 1.29.0
运行时	Python 3.12.13（uv 0.12.1 托管）、ksycopg2 2.9.1（aarch64 构建）
MCP 客户端	Claude Code（本机 Windows,经 SSH stdio 接入）；协议版本 2025-06-18
测试数据	mcp_demo 库：customers 20000 行、orders 300000 行、finance.salaries 500 行（未授权 schema）

一、背景：接上数据库容易,放心交出数据库难

写一个能让大模型执行 SQL 的接口,一个下午就够了。难的从来不是连接本身,而是连接之后运维同事必然会问的那几句：AI 生成的 SQL 写坏数据怎么办？它要是 DROP 了一张表呢？它会不会读到不该读的东西？这些问题答不上来,"AI 接数据库"就只能停在演示环境里,上不了生产。

MCP（Model Context Protocol,模型上下文协议）把这件事往前推了一步。AI 客户端和工具服务端之间用 JSON-RPC 通信,工具能做什么、不能做什么,由服务端声明和控制。电科金仓在 Gitee 开源的 KES MCP Server（仓库名 kingbase-mcp）就是这个思路的落地：把 KingbaseES 的结构探索、SQL 执行、执行计划分析、索引推荐、健康巡检封装成 MCP 工具,再配上访问模式控制和 SQL 校验。

官方对它的定位是"使 AI 助手具备数据库结构探索、SQL 执行、执行计划分析、索引优化和健康检查能力"。清单看着很全。但文档不会替你回答敢不敢放手,所以我的做法很直接：每项能力真跑一遍,每道护栏故意撞一次,报错原样贴出来。

二、架构全景：一次调用要过几道关

动手前先把链路看清楚。KES MCP Server 是个中间层,上游接支持 MCP 协议的 AI 客户端（Claude Code、Cursor、TRAE 这些）,下游用金仓原生 Python 驱动 ksycopg2 连 KES。

图 1 说明：本文实测环境的分层架构。数据来源：kingbase-mcp 官方仓库 README 与本文部署实测,版本号均为实测值。

三、部署实战：十条命令,两个坑

部署目标很朴素：在麒麟 V10 的 aarch64 机器上把它装到能用。动手前我预想的最大障碍是 Python 版本,kingbase-mcp 要求 3.12+,而麒麟自带的是 3.7.4。后来发现用 uv 的托管 Python 就绕过去了,系统环境一点不用动。

先看环境底账：

图 2 说明（场景 1）：操作系统、CPU 架构、内存与 KES 版本确认,select version() 返回 KingbaseES V009R001C010。

安装就四条命令：

curl -LsSf https://astral.sh/uv/install.sh | sh
git clone --depth 1 https://gitee.com/king-db/kingbase-mcp.git
cd kingbase-mcp && uv venv --python 3.12
UV_DEFAULT_INDEX=https://pypi.tuna.tsinghua.edu.cn/simple uv pip install .

装完的状态：

图 3 说明（场景 2）：uv 0.12.1、kingbase-mcp 0.3.0、Python 3.12.13、ksycopg2 2.9.1（针对 Python3.12 Kingbase V9 的构建,编译信息见输出）与 CLI 参数一览。

过程比我预想的顺。uv 自己拉了 3.12.13 的托管解释器,依赖走清华镜像几十秒装完,aarch64 上没碰到编译失败。担心过的 ksycopg2 驱动问题也没出现：PyPI 上的版本直接提供 aarch64 + Python 3.12 的构建。这里提醒一句,KES 安装目录 Interface/Python 下自带的驱动包只有 Python 3.5 和 2.7 的版本,别走那条老路。

坑还是有两个。一个是 uv 建的虚拟环境默认不带 pip,我习惯性敲 pip show 直接报 No such file,查版本得用 python -c "import importlib.metadata ..." 这种写法。另一个是备查性质的：如果驱动运行时报找不到 libkci 动态库,要设 KSYCOPG2_LIB_PATH 指到 KES 安装目录的 Server/lib。我这台机器上 PyPI 版驱动自带依赖,没触发。

连接配置就一个环境变量 DATABASE_URI,格式 kingbase://用户:密码@主机:端口/库名。

四、连得上：一次真实的 MCP 握手

我没有一上来就挂客户端,而是先直接朝 stdio 进程发 JSON-RPC 原始报文,把 initialize、tools/list、tools/call 三步看个明白。连接用的是只读账号 mcp_reader（建号语句在第七节和附录）,不是 system 管理员。这个顺序建议保持：先用裸协议确认链路没问题,再上客户端,出问题好定位。

图 4 说明（场景 3）：initialize 返回 serverInfo（kingbase-mcp,版本 1.29.0）与协议版本 2025-06-18；tools/list 返回 10 个工具；execute_sql 首次调用返回 {'version': 'KingbaseES V009R001C010'}。

这里我踩了个挺典型的坑。第一版测试脚本把三条请求一次性写进 stdin 然后关掉,结果只收到前两个响应,execute_sql 的结果永远等不到。排查半天才明白：服务端看到 stdin 的 EOF 就开始退出,第三个请求的数据库调用还没跑完进程就没了。改成发完请求保持 stdin 打开、收齐响应再关,立刻正常。另外 stdio 模式下日志走 stderr、报文走 stdout,两边分得很干净,这点做得规范。

Claude Code 的接入配置（本机 Windows,经 SSH 把 stdio 接到远程服务器）：

claude mcp add kes -- ssh -i ~/.ssh/kes_key root@<服务器IP> \
  "DATABASE_URI='kingbase://mcp_reader:<密码>@127.0.0.1:54321/mcp_demo' \
   /root/kingbase-mcp/.venv/bin/kingbase-mcp --access-mode restricted"

顺带说个 Windows 下的小插曲：我最先尝试用 claude mcp add-json 传 JSON 配置,引号转义折腾了两轮都报 Invalid input,换成上面这种 -- 直通写法一次成功。遇到同样问题的可以直接抄。

这套拓扑我很满意的一点：数据库端口 54321 完全不对公网开放（我这台机器的安全组实测就是不通的）,AI 客户端的所有流量都走 SSH 加密通道。从 Windows 本机跑一次握手加一次查询的文字记录：

[id=1] connected: {"name": "kingbase-mcp", "version": "1.29.0"} protocol 2025-06-18
[id=2] isError=False result: [{'orders_total': 300000}]
round-trip total: 1.1 s, missing=[]

注册完新开一个 Claude Code 会话,效果是这样的（组图 3 张）：

图 5 说明（场景 13 之 1）：本机 Claude Code 的 /mcp 管理面板实拍,kes 服务器状态为 connected,注册 10 tools。

图 6 说明（场景 13 之 2）：Tools for kes 前 5 项——List Schemas、List Objects、Get Object Details、Explain Query、Analyze Workload Indexes,每一项右侧均标注 read-only。

图 7 说明（场景 13 之 3）：后 5 项——Analyze Query Indexes、Analyze Database Health、Get Top Queries、Analyze DB Config、Execute SQL (Read-Only),同样全部标注 read-only。

留意工具面板上的一个细节：10 个工具每一项都标着 read-only。受限模式不只在服务端生效,客户端入口处就把 AI 能做什么、不能做什么亮明了。

Cursor 用户在 .cursor/mcp.json 里写等价的 command/args 就行,官方 README 有模板。

后面的场景截图里会频繁出现一个叫 mcpq 的命令。那是我自己写的 60 行小工具（全文在附录）,每次调用完整走一遍 initialize + tools/call,命令行上只留工具名和 JSON 参数。写它纯粹是为了截图清晰,本质上就是个最小化 MCP 客户端。

五、能干活（上）：结构探索与业务查询

演示库是个电商场景,2 万客户、30 万订单。我模拟 AI 助手接到任务后的标准动作：先看有什么表、表长什么样,再执行业务查询。

图 8 说明（场景 4）：list_objects 列出 public 下两张基础表；get_object_details 返回 orders 表逐列定义（id 自增主键、customer_id、status、amount、pay_channel 等）。

图 9 说明（场景 5）：count(*) 返回 300000；"已支付订单金额 TOP3 城市"聚合查询返回 shanghai 25005 单 / 12535492、hangzhou 25005 单 / 12505152、shenzhen 24990 单 / 12439720。

结构和数据的返回都是机器可解析的 JSON,AI 客户端拿到就能组织成自然语言回答；30 万行的 join 聚合在这台 4 核小机器上跑得也算利索。倒是 list_schemas 的行为让我多看了一眼：它返回的清单里压根没有那个未授权的 finance schema。也就是说,AI 从入口处就"看不见"无权限的对象,跟第七节的权限实验对上了。

六、能干活（下）：一条慢查询的 AI 调优全过程

这是全文我最想写的部分。任务本身平平无奇：select * from orders where customer_id=4242 and status='paid',orders 表 30 万行,customer_id 和 status 上都没索引。但处理过程能看出这套工具跟"能执行 SQL 的接口"差在哪。

先交代两个概念。假设索引（hypothetical index）：把一个不存在的索引"虚拟地"注册进优化器,看执行计划会怎么变,但不真建,靠 KES 的 sys_hypo 扩展实现（实测版本 1.1.2）。索引推荐：analyze_query_indexes 工具,背后是 DTA（Database Tuning Advisor）一类的算法,会给出候选索引和量化收益。

6.1 定位问题,然后一头撞在护栏上

AI 先看执行计划：全表扫描,优化器总成本 6212.46（Gather 节点,其中并行扫描节点 5212.06）。接着尝试假设索引仿真,工具回复说需要 sys_hypo 扩展,提示消息把这个扩展是什么、装它要什么权限、以后怎么卸载都讲清楚了。于是 AI 顺着提示尝试自己装扩展。结果：

图 10 说明（场景 6）：上：无索引执行计划（Seq Scan,成本 0.00…5212.06）；中：假设索引仿真提示需要 sys_hypo 扩展；下：AI 尝试 create extension 被拒——cannot execute CREATE EXTENSION in a read-only transaction。

说实话这个报错让我愣了几秒,然后越想越觉得合理。CREATE EXTENSION 明明在受限模式的语句白名单里（官方有意为之,装监控类扩展是常见需求）,语法树校验也确实放行了。但受限模式还有第二重机制：所有语句都包在只读事务里跑。写系统目录的操作到了数据库引擎层,照样被摁住。两道机制叠着用,白名单万一有疏漏,只读事务还能兜底。

6.2 DBA 出场：一条命令解锁,AI 继续干活

扩展安装属于结构变更,本来就该人工审批。我切到 DBA 角色用 ksql 装上 sys_hypo,AI 那边重试,仿真立刻通了,索引推荐也一并出来了：

图 11 说明（场景 7）：DBA 执行 create extension sys_hypo 后,假设索引 (customer_id, status) 仿真生效——计划变为 Bitmap Index Scan + Bitmap Heap Scan,总成本 19.67；analyze_query_indexes 输出推荐摘要：1 条推荐、基线成本 6212.5、新成本 19.7、预估索引体积 755.9 kB、改善倍数 315.8。

推荐器给的不只是"建议在 (customer_id, status) 上建索引"一句话,还有成本从 6212.5 到 19.7 的量化预估。审批变更的人拿到的是数字,不是感觉。更关键的是,到这一步为止没有建过任何真实索引,30 万行的表没有承担一点建索引的写放大。

6.3 真实索引验证：仿真到底准不准

光有预估不行,得验收。我先在无索引状态下连跑三次 EXPLAIN ANALYZE：

图 12 说明（场景 8）：无索引状态,同一查询连续三次 EXPLAIN ANALYZE：Execution Time 分别为 24.512 / 24.614 / 24.540 ms；计划为并行全表扫描,每个 worker 过滤掉 150000 行。

然后建索引（119.729 ms 完成）,再跑三次：

图 13 说明（场景 9）：create index idx_orders_cust_status on orders(customer_id, status) 耗时 119.729 ms；建后三次 EXPLAIN ANALYZE：Execution Time 分别为 0.027 / 0.030 / 0.031 ms,计划为 Bitmap Index Scan,实测成本 4.46…20.05。

汇成一张图：

图 14 说明：左图为执行时间（对数刻度,每组 3 次采样点已标出,取中位数 24.540 ms → 0.030 ms,约 818 倍）；右图为优化器成本三态对比（基线 6212.46 → 仿真预估 19.67 → 真实索引 20.05,预估与真实相差约 1.9%）。数据来源：图 11、图 12、图 13 所示实测输出。

三点感受。第一,数据稳得出乎意料：建索引前三次执行时间极差只有 0.102 ms（相对中位数约 0.4%）,建后极差 0.004 ms,不用纠结波动。第二,仿真预估的成本 19.67 和真实索引建成后的 20.05 只差约 1.9%,这个例子里假设索引给的收益判断和真实情况是一致的。第三,也得泼点冷水：推荐器预估索引体积 755.9 kB,实际建出来 9.1 MB（见图 18）,差了 12 倍。成本预估和体积预估的准头完全不在一个水平,拿体积数字做磁盘容量规划之前,还是以真实构建为准。

七、敢放手：把每道护栏都撞一遍

能力看完了,回到开头的问题：凭什么放心？我的做法是把每道护栏都故意撞一遍,撞出来的报错就是答案。

图 15 说明：受限模式下一条 SQL 的三层处置流程,三个拒绝出口的报错文案均来自图 16、图 10、图 17 的实测输出。

7.1 第一层：语法树白名单

受限模式下,把六类危险操作挨个发给 execute_sql：

图 16 说明（场景 10）：INSERT、UPDATE、DELETE、DROP TABLE、多语句注入（select 1; drop table orders）、CREATE INDEX 六条语句全部返回 Error: Error validating query,语句原文被回显在错误信息中。

结果呢？全拦了,数据库根本没收到这些语句。我特意试了 select 1; drop table orders 这种搭车写法,前半句完全合法,后半句藏着刀。语法树解析直接把它识别为多语句,整体拒绝,不存在前半句放行、后半句偷跑的空间。CREATE INDEX 被拦这一条也呼应了第六节的流程：AI 可以推荐索引,创建必须由人来。

7.2 第二层与第三层：只读事务 + 最小权限账号

第二层在 6.1 节已经撞过了：白名单放行的 CREATE EXTENSION,被只读事务摁在引擎层。第三层是数据库自己的权限体系,前提是接入账号本身就得是最小权限的：

CREATE USER mcp_reader WITH LOGIN PASSWORD '<自定义密码>';
GRANT CONNECT ON DATABASE mcp_demo TO mcp_reader;
GRANT USAGE ON SCHEMA public TO mcp_reader;
GRANT SELECT ON ALL TABLES IN SCHEMA public TO mcp_reader;

第三层到底兜不兜得住,我做了个反向实验：故意把 MCP Server 起成非受限模式（等于把第一、二层撤了）再发写入；另外用受限模式发一条语法完全合规的跨 schema 查询。

图 17 说明（场景 11）：上：unrestricted 模式下 INSERT 穿过了 Server 层,被数据库拒绝——permission denied for table orders；下：受限模式下 select * from finance.salaries 通过语法校验,被数据库拒绝——permission denied for schema finance（报错回显里可见 /* kingbase-mcp */ 来源标记）。

这组实验让我把三层防线的分工彻底看明白了：语法树白名单管语句类型,只读事务管引擎层写操作,账号权限管对象边界。哪怕哪天运维手一抖把服务起成了非受限模式,只要账号是 SELECT-only 的,写操作照样进不去；哪怕语句是纯读的,没授权的 schema 照样读不到。放手的底气说穿了很朴素：就算 AI 生成了错误甚至危险的 SQL,它也没有改数据的通道。

八、日常运维：健康巡检与慢查询回看

最后看 AI 当运维助手的两项基本功：巡检和慢查询定位。用到 analyze_db_health（7 类检查,可单项或逗号组合调用）和 get_top_queries（基于 sys_stat_statements 扩展,我这台机器该扩展已在 shared_preload_libraries 里,直接 CREATE EXTENSION 就能用）。有个准备工作：慢查询指纹的完整文本需要监控权限,我给 mcp_reader 授了 KES 内置的 sys_read_all_stats 角色,不然查询文本显示 insufficient privilege。

图 18 说明（场景 12）：上：connection,buffer,index 组合巡检——连接 9 总数 / 0 空闲,索引缓存命中率 99.8%、表缓存命中率 99.2%（阈值 95%）,并提示新建的 idx_orders_cust_status “仅被扫描 3 次、占用 9.1 MB”；下：按平均耗时排序的 TOP3 慢查询——perf.snapshot_timer()（KES 自带巡检定时任务,25 次调用、均值 157.13 ms）、本文的建索引语句（119.46 ms）、带 /* kingbase-mcp */ 标记的聚合查询（46.81 ms）。

巡检结果里我最喜欢的一条,是它对我们刚建的那个索引的提醒：只被扫描过 3 次（就是场景 9 那三次采样）,占 9.1 MB。工具不知道这索引是三分钟前刚建的,它只是按规则报告扫描次数低的索引。这条提醒放在真实运维里就得结合上下文判断了,AI 给的是线索,结论还得人下。TOP 慢查询那半屏也值得看：AI 自己发的查询带着来源标记,跟系统任务、DBA 操作同榜排列,慢查询要归因的时候证据链是全的。

有两处异常我原样报告。一是 health_type: "all" 一次跑全部 7 项检查时,0.3.0 版本会报连接池状态错误（connection pool is closed 之类）,单项调用和逗号组合都稳定,我就按组合方式绕过了,希望后续版本修掉。二是 vacuum 检查对 KES 内部表 sys_catalog._kingbase_loginfo 报了个 “-999,999 transactions remaining” 的负值,看着像对内部表的解读边界问题,用户表的巡检结果不受影响。

九、数据汇总

性能实验（同一查询、同一数据集,控制变量,各 3 次采样取中位数）：

指标	无索引	假设索引（仿真）	真实索引
优化器总成本	6212.46	19.67	20.05
执行时间中位数	24.540 ms	—（仿真不执行）	0.030 ms
三次采样值	24.512 / 24.614 / 24.540	—	0.027 / 0.030 / 0.031
索引体积	—	预估 755.9 kB	实测 9.1 MB
建索引耗时	—	0（不创建）	119.729 ms

数据出处：图 11、图 12、图 13、图 18。

护栏实验（受限模式,除注明外）：

试探动作	拦截层	实测报错
INSERT / UPDATE / DELETE	第一层 语法树白名单	Error validating query
DROP TABLE / CREATE INDEX	第一层 语法树白名单	Error validating query
多语句注入 select 1; drop table	第一层 语法树白名单	Error validating query
CREATE EXTENSION（白名单内）	第二层 只读事务	cannot execute CREATE EXTENSION in a read-only transaction
INSERT（unrestricted 模式）	第三层 账号权限	permission denied for table orders
跨 schema 读 finance.salaries	第三层 账号权限	permission denied for schema finance

数据出处：图 10、图 16、图 17。

十、总结与展望

回头看标题里那两个词。

连得上,实测门槛不高。uv 解决 Python 版本,清华镜像解决依赖,一个环境变量解决连接,aarch64 国产化环境没多花一分钟适配。Claude Code 走 SSH stdio 接远程实例,数据库端口从头到尾没对公网开过。

敢放手,实测是有依据的。三层防线各自独立生效,六类危险语句、只读事务边界、对象级权限,每一样都留下了真实报错。假设索引让 AI 提优化建议不必在生产表上试错,本例里成本预估和真实值只差约 1.9%,这样的数字放进变更审批单是站得住的。

问题也留了几个给后续版本：health all 模式的连接池异常、索引体积预估差 12 倍、个别内部表的巡检解读。但这些不挡主链路,结构探索、只读查询、仿真调优、巡检这一套,现在就能在生产环境跑起来。

我自己接下来打算做两件事。一是把 analyze_workload_indexes（基于 sys_stat_statements 全量工作负载的索引推荐）挂进每周巡检,AI 出周报,人做变更。二是结合 KES 的 KWR 报告,让 AI 客户端把巡检数据翻译成业务方看得懂的语言。AI 不会替 DBA 按下那个执行键,但按键之前的信息准备工作,它确实能做掉大半。

附录：可复现清单

1. 演示库与最小权限账号（ksql 以 system 连接执行）：

CREATE DATABASE mcp_demo;
\c mcp_demo

CREATE TABLE customers (
  id serial PRIMARY KEY, name varchar(64) NOT NULL, city varchar(32) NOT NULL,
  vip_level int NOT NULL DEFAULT 0, created_at timestamp NOT NULL DEFAULT now());
INSERT INTO customers(name, city, vip_level, created_at)
SELECT 'cust_' || g,
       (ARRAY['beijing','shanghai','guangzhou','shenzhen','chengdu','hangzhou'])[1 + (g % 6)],
       g % 5, now() - (g % 730) * interval '1 day'
FROM generate_series(1, 20000) g;

CREATE TABLE orders (
  id serial PRIMARY KEY, customer_id int NOT NULL, status varchar(16) NOT NULL,
  amount numeric(12,2) NOT NULL, pay_channel varchar(16) NOT NULL, created_at timestamp NOT NULL);
INSERT INTO orders(customer_id, status, amount, pay_channel, created_at)
SELECT 1 + (g * 37 % 20000),
       (ARRAY['paid','pending','cancelled','refunded'])[1 + (g % 4)],
       round((random()*999 + 1)::numeric, 2),
       (ARRAY['wechat','alipay','unionpay'])[1 + (g % 3)],
       now() - (g % 365) * interval '1 day' - (g % 86400) * interval '1 second'
FROM generate_series(1, 300000) g;
ANALYZE customers; ANALYZE orders;

CREATE SCHEMA finance;
CREATE TABLE finance.salaries (emp_no serial PRIMARY KEY, name varchar(64), salary numeric(10,2));
INSERT INTO finance.salaries(name, salary)
SELECT 'emp_' || g, 8000 + (g % 30000) FROM generate_series(1, 500) g;

CREATE EXTENSION IF NOT EXISTS sys_stat_statements;

CREATE USER mcp_reader WITH LOGIN PASSWORD '<自定义密码>';
GRANT CONNECT ON DATABASE mcp_demo TO mcp_reader;
GRANT USAGE ON SCHEMA public TO mcp_reader;
GRANT SELECT ON ALL TABLES IN SCHEMA public TO mcp_reader;
GRANT SELECT ON ALL SEQUENCES IN SCHEMA public TO mcp_reader;
GRANT sys_read_all_stats TO mcp_reader;
GRANT USAGE ON SCHEMA sys_hm TO mcp_reader;
GRANT SELECT ON ALL TABLES IN SCHEMA sys_hm TO mcp_reader;

2. MCP Server 安装与启动：

curl -LsSf https://astral.sh/uv/install.sh | sh
git clone --depth 1 https://gitee.com/king-db/kingbase-mcp.git
cd kingbase-mcp && uv venv --python 3.12
UV_DEFAULT_INDEX=https://pypi.tuna.tsinghua.edu.cn/simple uv pip install .
export DATABASE_URI='kingbase://mcp_reader:<密码>@127.0.0.1:54321/mcp_demo'
./.venv/bin/kingbase-mcp --access-mode restricted

3. mcpq 调用器全文（本文截图中的 mcpq 命令即此脚本,保存为 /root/mcpq.py,并以 exec python3 /root/mcpq.py "$@" 包一层放入 PATH）：

#!/usr/bin/env python3
import json
import os
import subprocess
import sys
import time

os.environ.setdefault("DATABASE_URI", "kingbase://mcp_reader:<密码>@127.0.0.1:54321/mcp_demo")
mode = os.environ.get("MCPQ_MODE", "restricted")
tool = sys.argv[1]
args = json.loads(sys.argv[2]) if len(sys.argv) > 2 else {}

p = subprocess.Popen(["/root/kingbase-mcp/.venv/bin/kingbase-mcp", "--access-mode", mode],
                     stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL,
                     universal_newlines=True, bufsize=1, env=os.environ)

def send(o):
    p.stdin.write(json.dumps(o) + "\n")
    p.stdin.flush()

send({"jsonrpc": "2.0", "id": 1, "method": "initialize",
      "params": {"protocolVersion": "2025-06-18", "capabilities": {},
                 "clientInfo": {"name": "mcpq", "version": "1.0"}}})
p.stdout.readline()
send({"jsonrpc": "2.0", "method": "notifications/initialized"})
send({"jsonrpc": "2.0", "id": 2, "method": "tools/call",
      "params": {"name": tool, "arguments": args}})

deadline = time.time() + 150
while time.time() < deadline:
    line = p.stdout.readline()
    if not line:
        break
    try:
        m = json.loads(line)
    except Exception:
        continue
    if m.get("id") == 2:
        res = m.get("result", {})
        if res.get("isError"):
            print("[isError=True]")
        for c in res.get("content", []):
            if isinstance(c, dict) and c.get("type") == "text":
                t = c["text"]
                obj = None
                try:
                    obj = json.loads(t)
                except Exception:
                    try:
                        import ast
                        obj = ast.literal_eval(t)
                    except Exception:
                        obj = None
                if obj is None:
                    t = "".join(ch for ch in t if ord(ch) < 128 or ch in "\n")
                    print("\n".join(ln.rstrip() for ln in t.splitlines() if ln.strip()))
                elif isinstance(obj, list) and obj and all(isinstance(r, dict) for r in obj):
                    for r in obj:
                        print(json.dumps(r, ensure_ascii=False, default=str))
                else:
                    print(json.dumps(obj, ensure_ascii=False, indent=1, default=str))
        break

p.stdin.close()
p.kill()

4. 正文各场景的调用命令（与截图逐字一致；<慢查询> 代指 select * from orders where customer_id=4242 and status=$$paid$$）：

mcpq list_objects '{"schema_name":"public"}'
mcpq get_object_details '{"schema_name":"public","object_name":"orders"}' | head -33
mcpq execute_sql '{"sql":"select count(*) from orders"}'
mcpq execute_sql '{"sql":"select c.city, count(*) as orders_n, round(sum(o.amount))::bigint as total from orders o join customers c on c.id=o.customer_id where o.status=$$paid$$ group by c.city order by total desc limit 3"}'
mcpq explain_query '{"sql":"<慢查询>"}'
mcpq explain_query '{"sql":"<慢查询>","hypothetical_indexes":[{"table":"orders","columns":["customer_id","status"]}]}'
mcpq execute_sql '{"sql":"create extension if not exists sys_hypo"}'
ksql -p 54321 -d mcp_demo -U system -c 'create extension sys_hypo;'
mcpq analyze_query_indexes '{"queries":["<慢查询>"]}' | head -24
mcpq execute_sql '{"sql":"insert into orders(customer_id,status,amount,pay_channel,created_at) values (1,$$paid$$,9.9,$$wechat$$,now())"}'
mcpq execute_sql '{"sql":"update orders set amount=0 where id=1"}'
mcpq execute_sql '{"sql":"delete from orders where id=1"}'
mcpq execute_sql '{"sql":"drop table orders"}'
mcpq execute_sql '{"sql":"select 1; drop table orders"}'
mcpq execute_sql '{"sql":"create index idx_try on orders(customer_id)"}'
MCPQ_MODE=unrestricted mcpq execute_sql '{"sql":"insert into orders(customer_id,status,amount,pay_channel,created_at) values (1,$$paid$$,9.9,$$wechat$$,now())"}'
mcpq execute_sql '{"sql":"select * from finance.salaries limit 3"}'
mcpq analyze_db_health '{"health_type":"connection,buffer,index"}'
mcpq get_top_queries '{"sort_by":"mean_time","limit":3}'

5. 参考资料：

• KES MCP Server 官方仓库：https://gitee.com/king-db/kingbase-mcp
• 金仓官方文档站：https://docs.kingbase.com.cn
• MCP 协议规范：https://modelcontextprotocol.io