摘要上一篇我写的是 Docker 部署《从零到一基于 Docker 部署 KingbaseES V9R1C10》结尾埋了个钩子说想试试 KES MCP Server让 AI 客户端直接连金仓查库。这篇就把那个钩子收掉。先说实话我对 MCP 的认知也就刚入门很多细节是边跑边查的。宿主机装 uv 和 Python 3.12从 gitee 把 kingbase-mcp 仓库 clone 下来装依赖。金仓那边建只读账号 ai_mcp装 sys_hypo 和 sys_stat_statements 两个扩展。服务以 streamable-http 模式跑起来。我写了个几十行的客户端9 类工具挨个调了一遍。执行计划拿到了Top 5 慢查询拿到了7 项健康检查也跑出来了索引推荐也有。AI 跟金仓之间就差这层 MCP而这层 MCP我自己动手搭出来了。一、为什么换个玩法上次我交付的是一个能跑的容器。可真要用起来还是老流程。Navicat 里看表结构复制到别处看执行计划再开个窗口问 AI。来回倒腾烦得很。你可能也遇到过三个工具来回切人的精力全耗在搬运上。KES MCP Serverkingbase-mcp 0.3.0把金仓的能力封成了 9 类 MCP 工具list_schemas、list_objects、get_object_details、execute_sql、explain_query、analyze_db_health、analyze_db_config、get_top_queries、analyze_workload_indexes、analyze_query_indexes。Cursor、TRAE、Claude Code 这类 IDE 直接就能调。开发者在 IDE 里问一句AI 自己调工具工具去碰金仓。窗口不用切了。画了张图放上面。架构三层上层是 AI 客户端Cursor/TRAE/Claude Code中层是 KES MCP ServerPython 3.12 mcp SDK 1.29streamable-http/SSE/stdio 三种传输下层是金仓用 ai_mcp 账号接住所有调用。restricted 模式只放行 SELECT/EXPLAIN/SHOW/VACUUM 这些白名单语句。AI 想绕过去没门。二、九件事场景做什么关键命令图一AI 开发环境就绪uv Python 3.12 venv CLI 可用uv --versionuv python list--help图 2二金仓侧就绪AI 账号 扩展 最小权限授权CREATE USERGRANTCREATE EXTENSION图 3三启动 KES MCP Serverstreamable-http:8000 监听set -a; source .env; nohup kingbase-mcp ... 图 4四MCP 客户端接入initialize 工具注册表mcp_client.py tools图 5五自然语言查库表结构 数据统计get_object_detailsexecute_sql图 6六AI 驱动 SQL 调优执行计划 假设索引模拟explain_query 假设索引图 7七慢查询洞察 数据库健康巡检get_top_queriesanalyze_db_health图 8八参数画像 工作负载索引推荐analyze_db_configanalyze_query_indexes图 9装环境我先把 uv 装上Python 3.12 就位仓库 clone 完venv 建好依赖装上。uv --version显示 0.12.1。uv python list里 3.12.13 是已装状态。跑.venv/bin/kingbase-mcp --help参数选项都出来了--access-mode {unrestricted,restricted}、--transport {stdio,sse,streamable-http}。这两组开关后面都要用。装依赖那步uv pip install .我没截图。uv 的进度条一直刷截图没法看。60 个包主要的几个kingbase-mcp0.3.0、mcp1.29.0、ksycopg22.9.1、pglast7.11、starlette1.3.1、uvicorn0.52.1。ksycopg2 是金仓的官方驱动PyPI 有 Linux x86_64 的 wheel不用编译装起来省事。金仓侧准备建账号、装扩展、授权我一步步来。DROP USER IF EXISTS 先清同名。这行被跳过了上一轮建的还在PG 不让删正在被用的角色后头细说。CREATE USER ai_mcp WITH PASSWORD Kingbase123。ALTER ROLE ai_mcp SET search_path demo, public这行后面有大用场景六的坑就靠它解。GRANT USAGE ON SCHEMA demo TO ai_mcp、GRANT SELECT ON ALL TABLES IN SCHEMA demo TO ai_mcp权限卡死在 demo 只读。CREATE EXTENSION IF NOT EXISTS sys_hypo 和 sys_stat_statements已有就 NOTICEs 跳过。最后用 ai_mcp 连了一次SELECT COUNT(*) FROM demo.t_order 返回 10000上篇的表还在。截图里有一行ERROR: current logged-in user cannot be dropped。一开始我以为是脚本写错了排查半天。后来反应过来MCP Server 进程还握着 ai_mcp 的连接PG 不许删一个正在被用的角色。想改角色得先把服务停了先停 MCP、改账号、再启服务。这个顺序绕不开。把服务拉起来凭据我没直接 export写进了 /opt/kes-mcp/kingbase-mcp/.env权限 600。启动时 set -a; source .env; set a 拉进环境再 nohup .venv/bin/kingbase-mcp ... 后台跑。ps 里看不到密码history 里也没有踏实。ss -tlnp | grep 8000看到LISTEN 0 2048 0.0.0.0:8000端口就绪。curl POST /mcp 返回 400。第一眼还愣了一下查了下才发现 streamable-http 要 Accept 协商头不带就 400。算预期。tail 日志看到INFO 127.0.0.1:51032 - POST /mcp HTTP/1.1 400 Bad Request服务在响应。截图开头有行pkill -f kingbase-mcp 2/dev/null; sleep 1; echo OK清残留进程用的防端口被占。跑第二次能直接用。生产上这步得换成 systemd。工具清单我写了个客户端 mcp_client.py60 行左右放 /tmp。用 mcp SDK 1.29 的 streamablehttp_client 连 http://127.0.0.1:8000/mcp子命令tools/details/query/explain/hypo/slow/health/config/qindexes。真实场景里 Cursor/TRAE 的 AI 也是走这套协议这里只是把 AI 那层换成终端输出看得见摸得着。后面几个场景全是它跑出来的。tools 一列10 项list_schemas、list_objects、get_object_details、explain_query、analyze_workload_indexes、analyze_query_indexes、analyze_db_health、get_top_queries、analyze_db_config、execute_sql。README 写 9 项我实测 10 项多了个 analyze_db_config。每一项都带 description 和参数 schema。AI 拿到就知道怎么选、怎么传参。查表结构 统计我先调 get_object_details问它 demo.t_order 这张表长什么样。它把 5 个列、2 个索引全给我列出来了连字段类型都标得清清楚楚。表结构拿到手我心里就有数了后面写 SQL 不用再猜字段名。再问一句两种状态各有多少单、总金额多少。execute_sql 直接跑分组统计返回 N 状态 7500 单、18967315.21 元S 状态 2500 单、6417424.10 元。我特意看了眼金额Decimal 类型一分不差。这玩意儿要是用 float早晚丢钱。假设索引explain_query 跑一条 statusS 的查询返回 Bitmap Heap ScanCost 51.66..167.91。表上本来就有 status 的索引走 Bitmap 不意外。我加了个假设索引再跑Cost 掉到 47.41..163.66。收益不大表上已有 status 索引。可这套玩法搬到没索引的查询上是通的。先模拟后决定不用真建真删。这中间栽了个坑。sys_hypo 的索引名里不能带 schema 点号传 demo.t_order 进去直接报syntax error at or near .。折腾了一会儿改成传 t_order 就好。场景二那句 ALTER ROLE 已经把 search_path 指到 demo 了。这坑客户端默认参数里已经绕开。慢查询 健康get_top_queries 出了 Top 5 慢查询。排第一的是一条 198ms 的 CTE几百行的 bloat 诊断 SQL跑起来确实重。列表里还混着 MCP 自己触发的查询带/* kingbase-mcp */前缀一眼能认出来。有行显示 insufficient privilege权限不够SQL 被藏了看不全。analyze_db_health 一口气跑 7 项检查。无效索引没有重复索引没有索引膨胀没有。倒是揪出一个未用索引idx_order_status 只被扫了 2 次。连接健康8 total 0 idle。vacuum 那边有个表接近 wraparound10M 内要处理。Buffer 命中率 index 90.6%、table 96.3%表命中贴着 95% 阈值2 GiB 机器就这样。以前这套得一条条敲现在一条调用全出。这工具也有坑。all 模式 7 项并发连接池偶发connection pool is closed。我改成单项分几次调health_typeindex,vacuum,constraint稳了。参数 索引推荐analyze_db_config 查 shared_buffers返回 128 MB提示偏低建议提到 2 GB。2 GiB 机器 25% 是 512 MB128 MB 确实低。演示机凑合生产得按建议提。analyze_query_indexes 拿两条查询去分析推荐给 t_order 建 amount 单列索引。amount4000 那条查询代价从 210.0 降到 158.91.3x。statusS 那条已经命中索引没变化。1 万行的表全表扫本来就快1.3 倍不稀奇。这套东西得拿到百万、亿行的表上才有看头。三、数据汇总维度实测结果出处环境搭建uv 0.12.1 Python 3.12.13 venv 60 包就绪CLI 可用图 2凭据安全DATABASE_URI 写入 .env600 权限命令行不出密码图 4账号权限ai_mcp 最小权限demo schema 只读扩展已装图 3MCP 工具实测注册 10 项含 README 未列出的 analyze_db_config图 5自然语言查库get_object_details execute_sql 毫秒级返回图 6假设索引收益status 假设索引代价 167.91 → 163.66≈8% 提升图 7慢查询MCP 内部 CTE 198ms 居首sys_stat_statements 已采集 18 条图 8健康巡检7 项中 1 项提示idx_order_status 利用率低其余健康图 8参数画像shared_buffers128MB 25% 内存建议2 GiB 演示机可接受图 9索引推荐amount 单列索引预测 1.3x 提升210.0 → 158.9 cost图 9四、写在最后开头那个钩子到这里收住了。上一篇我交付一个能跑的容器这一篇交付AI 看得见它。两篇都在 2 vCPU / 1.9 GiB 的小机器上跑完uv 隔离环境streamable-http 多团队共用restricted 不给写权限。这几个细节凑一块AI 直连国产数据库才算有了着落。说真的跑完这趟我心里踏实了不少。国产库这些年性能、兼容性一直在追工具链还是差口气。KES MCP Server 把开发者怎么跟数据库打交道这一层补上了。你手里要是有台能跑 Docker 的小机器照着附录 A 走一遍一个下午就能复现。试试看不亏。附录 AMCP 客户端核心脚本用于 IDE 集成可按需精简#!/usr/bin/env python3 import asyncio import json import sys from mcp import ClientSession from mcp.client.streamable_http import streamablehttp_client URL http://127.0.0.1:8000/mcp TOOLS { details: (get_object_details, {schema_name: demo, object_name: t_order, object_type: table}), query: (execute_sql, {sql: SELECT status, COUNT(*) AS cnt, ROUND(SUM(amount),2) AS total FROM demo.t_order GROUP BY status;}), explain: (explain_query, {sql: SELECT * FROM demo.t_order WHERE statusS;, analyze: False}), hypo: (explain_query, {sql: SELECT * FROM demo.t_order WHERE statusS;, analyze: False, hypothetical_indexes: [{table: t_order, columns: [status]}]}), slow: (get_top_queries, {sort_by: mean_time, limit: 5}), health: (analyze_db_health, {health_type: all}), config: (analyze_db_config, {parameter: shared_buffers}), qindexes:(analyze_query_indexes, {queries: [SELECT * FROM t_order WHERE statusS, SELECT * FROM t_order WHERE amount4000], max_index_size_mb: 100, method: dta}), } def fmt(content): if content is None: return if isinstance(content, str): return content out [] for item in content: t getattr(item, text, None) if isinstance(t, str) and t: out.append(t) else: s getattr(item, resource, None) if s is not None: out.append(json.dumps(s, ensure_asciiFalse, indent2)) else: out.append(str(item)) return \n.join(out) async def main(): mode sys.argv[1] if len(sys.argv) 1 else tools async with streamablehttp_client(URL) as (read, write, _): async with ClientSession(read, write) as session: await session.initialize() if mode tools: tools await session.list_tools() print(MCP server connected: KingbaseES MCP Server (streamable-http)) print(Total tools: %d\n % len(tools.tools)) for t in tools.tools: print(- %s % t.name) print( desc: %s % (t.description or ).split(\n)[0]) print( args: %s % , .join(t.inputSchema.get(properties, {}).keys())) return name, args TOOLS[mode] print( call_tool: %s % name) print( arguments: %s % json.dumps(args, ensure_asciiFalse) if args else arguments: (none)) res await session.call_tool(name, args or {}) print( result:) print(fmt(res.content) if res.content else (no content)) if res.isError: print( isError: True) if __name__ __main__: asyncio.run(main())