1. 业务方一句中文DBA 就得写一段 SQL“上周谁买得最多”——这句话在业务群里出现的频率大概和“在吗”差不多。它不难但它烦。DBA 手头正调着慢查询看到这句话就得切窗口、翻表结构、写一段 GROUP BY、跑出来贴回去。问题用中文描述答案在数据库里中间那段 SQL 却要最贵的人力来填。KingbaseES 作为国产数据库里生态比较完整的一员本身并不缺查询能力缺的是“让不懂 SQL 的人直接问”的那层翻译。MCP Server 就是干这个的它把数据库能力包装成 AI Agent 能调用的工具Agent 负责把中文翻成 SQL、把结果翻回中文数据库只管执行。整条链路里MCP Server 是那个“嘴替”的嗓子。这篇要交付的东西很具体一个能连 KingbaseES 的 MCP Server 配置骨架包含连接参数和工具声明启动之后由 AI Agent 发一句自然语言查询确认它被转成 SQL、打到金仓、拿回真实结果。适合手上有一台能跑 KingbaseES 的机器、想让 AI 帮忙查数但又不敢直接放开写权限的人。下面按“先跑通、再收紧”的顺序来。2. 前置准备TaoToken 与金仓连接信息MCP Server 本身不产生智能它只是工具层。真正把中文翻成 SQL 的“大脑”是 LLM所以你需要一个能调模型的入口。我这边用的是 TaoToken它提供 OpenAI 兼容的接口接进 Agent 侧比较省事。官网在 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content API 地址是 https://taotoken.net/api 注意 API 路径不带 UTM 参数配置里填干净的那个。先去控制台把 Key 建出来入口在 https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content Key 管理页在 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 。建完先别急着写代码用模型对话页 https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 发一句“你好”确认 Key 是通的省得后面把网络问题误判成代码问题。金仓这边需要准备四样东西主机地址、端口KingbaseES 默认 54321、库名、以及一个只读账号。只读账号是后面安全设计的地基现在就要建不要图省事用超级用户。建账号的 SQL 大致是这样-- 用管理员账号执行 CREATE USER ai_ro WITH PASSWORD 换成你的强密码; GRANT CONNECT ON DATABASE test TO ai_ro; 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;Python 侧需要mcp和psycopg2两个包。KingbaseES 兼容 PG 协议所以直接用 psycopg2 连 54321 端口即可不需要额外驱动改造pip install mcp[cli] psycopg2-binary装完可以先用一段最小脚本验证连通性确认账号权限和端口都对import psycopg2 conn psycopg2.connect( host10.0.0.12, port54321, dbnametest, userai_ro, password你的密码 ) cur conn.cursor() cur.execute(select version();) print(cur.fetchone()[0]) cur.execute(select count(*) from orders;) print(orders 行数:, cur.fetchone()[0])如果这一步报permission denied说明授权没生效回到上面的 GRANT 检查如果报连接超时先确认端口和防火墙别急着改代码。3. 可复制的 MCP Server 骨架MCP Server 的核心是把普通 Python 函数用装饰器暴露成工具。工具函数的 docstring 不是写给人看的注释是写给大模型看的说明书——Agent 靠它判断什么时候该调哪个工具。这是写 MCP Server 和写普通后端接口最大的区别你的注释第一次有了“读者是 AI”。下面这份骨架可以直接存成kingbase_mcp.py。它声明了四个工具list_tables让 Agent 先看清有哪些表describe_table让它知道字段名和类型run_query是核心查询能力db_info用来验明后端身份。import re import psycopg2 from mcp.server.fastmcp import FastMCP mcp FastMCP(kingbase) DB dict(host10.0.0.12, port54321, dbnametest, userai_ro, password你的密码) WRITE_WORDS re.compile( r\b(insert|update|delete|drop|alter|truncate|create|grant|revoke)\b, re.IGNORECASE) def get_conn(): conn psycopg2.connect(**DB) conn.set_session(readonlyTrue) # 闸二会话级只读 cur conn.cursor() cur.execute(SET statement_timeout5s) # 防慢查询拖垮库 return conn mcp.tool() def list_tables() - str: 列出当前库 public schema 下的所有表及大致行数用于先了解有哪些数据。 conn get_conn(); cur conn.cursor() cur.execute( select relname, n_live_tup from pg_stat_user_tables where schemanamepublic order by n_live_tup desc ) rows cur.fetchall() conn.close() return \n.join(f{r[0]} (~{r[1]} 行) for r in rows) or 没有业务表 mcp.tool() def describe_table(table: str) - str: 查看某张表的字段结构并返回 3 行示例数据用于写正确的 SQL。 if not re.fullmatch(r[A-Za-z_][A-Za-z0-9_]*, table): return 表名不合法 conn get_conn(); cur conn.cursor() cur.execute( select column_name, data_type from information_schema.columns where table_name%s order by ordinal_position , (table,)) cols cur.fetchall() cur.execute(fselect * from {table} limit 3) sample cur.fetchall() conn.close() head \n.join(f{c[0]} {c[1]} for c in cols) return f字段:\n{head}\n示例:\n{sample} mcp.tool() def run_query(sql: str) - str: 执行一条只读 SELECT/WITH 查询并返回结果最多 50 行。 s sql.strip().rstrip(;) if ; in s: return 已拒绝只允许单条语句 if not re.match(r^(select|with)\b, s, re.IGNORECASE): return 已拒绝只允许 SELECT/WITH 查询 if WRITE_WORDS.search(s): return 已拒绝检测到写操作关键字 conn get_conn(); cur conn.cursor() try: cur.execute(s) rows cur.fetchmany(50) cols [d[0] for d in cur.description] return f{cols}\n \n.join(str(r) for r in rows) except Exception as e: return f查询出错{e} finally: conn.close() mcp.tool() def db_info() - str: 返回数据库版本与当前连接身份用于确认后端真身。 conn get_conn(); cur conn.cursor() cur.execute(select version(), current_user, current_database()) v, u, d cur.fetchone() conn.close() return f{v} | {u} | {d} if __name__ __main__: mcp.run() # 默认 stdio 传输几个参数值得单独说。set_session(readonlyTrue)是会话级只读就算应用层白名单被绕过这一层也会挡statement_timeout5s防止 Agent 写出笛卡尔积把库拖死fetchmany(50)限制返回行数避免一次拉回几十万行把上下文撑爆。这三个数字不是拍脑袋是踩过坑之后定下来的——超时给 5 秒是因为正常业务查询都在 1 秒内给太长等于没给。4. 验证一句中文金仓作答服务写好了先别急着接客户端用官方 SDK 写个最小客户端走一遍完整协议握手确认它真的是标准 MCP Server而不是自己发明的接口。import asyncio from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client async def main(): params StdioServerParameters( commandpython3, args[/root/mcp/kingbase_mcp.py]) async with stdio_client(params) as (r, w): async with ClientSession(r, w) as session: await session.initialize() tools await session.list_tools() print(工具:, [t.name for t in tools.tools]) info await session.call_tool(db_info, {}) print(身份:, info.content[0].text) res await session.call_tool(run_query, { sql: select member_name, sum(amount) s from orders group by member_name order by s desc limit 1}) print(结果:, res.content[0].text) asyncio.run(main())跑通的话你会看到三行输出工具列表里有list_tables、describe_table、run_query、db_info身份那行返回类似KingbaseES V009R003C018 | ai_ro | test说明后端确实是金仓、连接身份是只读账号结果那行返回真实的聚合数据比如某个会员的累计消费额。这三个数字都来自金仓的真实计算没有一处硬编码。接下来才是主菜让 Agent 自己决定调哪个工具。把上面客户端里的call_tool换成让模型决策的流程或者直接接进支持 MCP 的客户端问一句“谁是消费冠军一共花了多少钱”。Agent 会先调list_tables摸清有哪些表再调describe_table看清orders的字段最后生成select member_name, sum(amount) ... group by ... order by ... limit 1并执行。你看到的回答是中文但底下跑的是真实 SQL。再试一个需要 JOIN 的问题“积分最高的会员买过东西吗”Agent 会自己组织member和orders两表的 LEFT JOIN。这一步能过说明工具声明里的 docstring 写得够清楚——Agent 知道describe_table能帮它看清字段所以敢写 JOIN。5. 本篇常见错排查报permission denied for table orders这是只读账号在权限层拦截不是 bug。检查GRANT SELECT是否覆盖了目标表以及ALTER DEFAULT PRIVILEGES有没有执行。如果新表没权限多半是建表时没走默认权限。报connection refused或超时先确认 KingbaseES 监听的是 54321 而不是 5432再确认pg_hba.conf里允许你的客户端 IP 连接。金仓的配置文件路径和 PG 略有差异别照搬 PG 的教程。工具列表为空检查mcp.tool()装饰器是否加在函数上以及mcp.run()是否在__main__里执行。FastMCP 靠装饰器注册工具漏一个就少一个。Agent 生成的 SQL 报语法错多半是describe_table返回的字段信息不够Agent 猜错了列名。把示例数据从 3 行加到 5 行或者把字段类型也带上通常能解决。查询被自己的白名单误杀WRITE_WORDS正则会匹配到列名里带update的情况比如updated_at。如果业务表有这类字段把正则改成匹配独立单词边界或者改成只检查语句开头和分号后的片段。返回结果被截断fetchmany(50)是故意的。如果确实需要更多行让 Agent 加limit或聚合而不是放开这个上限——上下文窗口比数据库连接更贵。6. 接进真实工位与后续要在真实环境用起来不需要写胶水代码。任何 MCP 兼容客户端只要在配置里加一段把kingbase这个 Server 注册进去重启后工具列表里就会出现它。之后对着聊天框问“这个月各产品卖了多少”Agent 会自动走list_tables→describe_table→run_query的流程SQL 由在线模型实时生成结果从金仓真实返回。如果你打算长期跑编码类或 Agent 类任务可以看下 Coding Plan 的入口 https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 按量或包月的选择在控制台里能直接看到。接入文档在 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 里面有 OpenAI 兼容接口的完整参数说明配 Agent 侧的时候对着填就行。最后留一个我自己的习惯每次改完 MCP Server先用 SDK 客户端跑一遍db_info和一条run_query确认协议握手和只读拦截都正常再去接客户端。这一步花三十秒能省掉后面半小时的“到底是模型问题还是服务问题”的排查。安全那三道闸——账号只读、会话只读、SQL 白名单——任意一道单独都能兜底但三道一起上才敢让它连生产库。