企业官网建设流程全解析

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_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ，API 地址是 https://taotoken.net/api ，注意 API 路径不带 UTM 参数，配置里填干净的那个。

先去控制台把 Key 建出来，入口在 https://taotoken.net/console?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ，Key 管理页在 https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= 。建完先别急着写代码，用模型对话页 https://taotoken.net/models?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_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( host="10.0.0.12", port=54321, dbname="test", user="ai_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(host="10.0.0.12", port=54321, dbname="test", user="ai_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(readonly=True) # 闸二：会话级只读 cur = conn.cursor() cur.execute("SET statement_timeout='5s'") # 防慢查询拖垮库 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 schemaname='public' 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(f"select * 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(readonly=True)是会话级只读，就算应用层白名单被绕过，这一层也会挡；statement_timeout='5s'防止 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( command="python3", 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_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ，按量或包月的选择在控制台里能直接看到。接入文档在 https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ，里面有 OpenAI 兼容接口的完整参数说明，配 Agent 侧的时候对着填就行。

最后留一个我自己的习惯：每次改完 MCP Server，先用 SDK 客户端跑一遍db_info和一条run_query，确认协议握手和只读拦截都正常，再去接客户端。这一步花三十秒，能省掉后面半小时的“到底是模型问题还是服务问题”的排查。安全那三道闸——账号只读、会话只读、SQL 白名单——任意一道单独都能兜底，但三道一起上，才敢让它连生产库。