于 KES MCP 的终端数据库 Agent 实践

最近我一直在改 kes-cli，目标不是再做一个传统命令行工具，而是把它做成一个终端聊天助手。原因也很简单，数据库排查这件事本来就不是一条命令能解决的。

平时遇到数据库问题，大概都是这个流程：先确认能不能连上库，再看有哪些 schema 和表，然后看表结构、查几条数据。如果发现 SQL 慢，还要继续看执行计划、索引、慢 SQL、锁等待。最后如果要复盘，还得把查过的东西整理成一份报告。

这些事情单独看都不复杂，但分散在数据库客户端、命令行、文档、聊天工具之间，就会变得很烦。

所以这次我想做的效果很明确：启动 kes-cli 后，直接在终端里聊天。比如我输入“看下我都有哪些表”“查一下 orders 前 5 条”“这个 SQL 为什么慢”“帮我看看数据库健康吗”，它能自己判断我要做什么，再通过 KES MCP Server 去拿真实结果，最后把结果整理出来。

image

这里面最关键的一点是，它不是让大模型自己猜数据库结构，也不是让大模型随便执行 SQL。我的想法是：模型负责理解问题和整理回答，MCP 负责连接数据库和采集证据，本地代码负责路由、安全限制和终端体验。这样分工清楚一点，工具用起来也放心一点。

如果按一条完整链路来看，这个终端工具本身就是 AI 客户端和数据库开发入口：先把模型和 KES MCP Server 配好，再确认数据库能连、工具能加载，然后用自然语言完成结构查询、只读查询、SQL 分析、运维诊断，最后把排查证据导出来。服务端编程这类变更风险更高的内容，也可以让 Agent 生成草稿和审查风险，但不会默认执行。这样它覆盖的不是某一个小功能，而是从“能连接数据库”到“能安全、稳定地完成数据库任务”的过程。

我之前也试过把所有能力拆成命令，结果越做越像工具箱。工具箱当然能用，但用户得先知道自己要拿哪个工具。数据库排查往往不是这样，很多时候用户只知道“这个查询慢”“这个表字段不确定”“现在库是不是正常”，并不知道下一步该查表结构、执行计划还是锁等待。所以我更希望 kes-cli 能先帮我判断问题类型，再把合适的能力调起来。

准备工作#
模型配置#
终端聊天工具肯定需要模型，所以第一步是把模型配置做顺。之前如果要配环境变量，用户还得自己找文档，现在我直接做了一个 /configure-model，第一次启动没有模型配置时，也会先引导配置。

这里支持三个方向：百炼、DeepSeek 和 GPT 兼容接口。我自己平时用百炼比较多，所以默认也放在百炼上。配置保存到 %APPDATA%\kes-cli\config.toml，后面启动会自动读取。

image

代码大概是这样：

def run_model_config_wizard(*, console: Console) -> AppSettings:
console.print(“当前可选：bailian、deepseek、gpt。配置会保存到 kes-cli config.toml，后续启动自动读取。”)
provider = Prompt.ask(
“请选择模型”,
console=console,
choices=[“bailian”, “deepseek”, “gpt”],
default=“bailian”,
)
if provider == “bailian”:
model = Prompt.ask(“模型名”, console=console, default=DEFAULT_BAILIAN_MODEL)
base_url = Prompt.ask(“Base URL”, console=console, default=DEFAULT_BAILIAN_BASE_URL)
这部分没做得太复杂，只要能解决“第一次怎么配模型”和“后面怎么自动读取”就够了。毕竟重点不是模型平台，而是怎么让这个终端工具通过 MCP 去完成数据库任务。

这里还有一个小细节。配置模型的时候不能让终端卡住，也不能配置完以后用户不知道保存到哪里。所以配置结束后会打印配置文件路径，后面如果要换模型，也可以重新进入 /configure-model。这个功能不算复杂，但对工具可用性很重要。否则每次启动都要用户检查环境变量，体验会很差。

我自己测试时主要看三点：第一，首次启动能不能正常进入配置；第二，百炼、DeepSeek、GPT 三个选项是否都能保存；第三，保存后再次启动是不是直接读取配置。只有这个稳定了，后面测试 MCP 和数据库能力才有意义。

MCP 配置#
接下来就是 MCP 配置。这里我只把 KES MCP Server 当作数据库工具层来接入，不单独展开 KES 本身。它对 kes-cli 的意义在于：我不用自己重新写一堆数据库采集逻辑，而是通过 MCP 工具拿到 schema、表结构、查询结果、执行计划、健康检查这些证据。

配置文件示例如下：

[mcp.servers.kingbase]
transport = “stdio”
command = “uv”
args = [
“–directory”,
“D:\AI-project\kingbase-mcp”,
“run”,
“kingbase-mcp”,
“–access-mode”,
“restricted”
]
access_mode = “restricted”

[mcp.servers.kingbase.env]
DATABASE_URI = “kingbase://user:password@127.0.0.1:54321/kes_cli_demo”
这里我默认用 stdio，本地调试比较省事，不用额外开端口。restricted 也很重要，因为这个工具不是为了让 AI 随便改库，而是先把只读查询和诊断场景跑顺。

配置完成后，我会先在终端里问一句：

先帮我检查一下 KES MCP Server 是否连接正常，工具都加载了吗
image

如果这里能看到工具加载成功，后面的表结构、查询、执行计划才有意义。否则模型再会说，也只是空聊。

所以我把 MCP 状态检查放在很靠前的位置。它不只是看“连没连上”，还要看工具有没有加载出来。比如 DATABASE_URI 写错了，MCP Server 可能能启动，但工具调用时会失败；如果 uv 启动目录不对，工具根本加载不出来；如果数据库权限不足，部分诊断结果也可能采集不到。这些都需要在一开始暴露出来，不能等到用户查表的时候才给一个模糊错误。

我现在把配置验证拆成几项来看：MCP Server 是否启动、transport 是否正常、工具数量是否符合预期、restricted 模式是否生效、数据库连接串是否能真正访问目标库。只要其中一步失败，就把失败作为结构化证据返回。

这里我也没有把 MCP 配置做成聊天技能。配置属于环境准备，真正进入聊天以后，用户应该更多关注数据库问题本身。配置写到 config.toml，状态检查用 /mcp 或自然语言触发，这样边界更清楚。

主要实现#
终端聊天入口#
入口我保持得很简单，就是：

uv run kes
启动以后进入 Textual 终端界面，用户可以直接输入自然语言，也可以输入 / 打开技能菜单。这里我特意不再支持一堆产品子命令，因为那样又会回到传统 CLI 的思路。

image

我希望用户看到的是一个聊天框，而不是一堆命令说明。比如我可以直接问：

看下 orders 有哪些字段
查一下 orders 前 5 条
帮我看看数据库健康吗
工具内部再去判断该调用哪个能力。这个入口做好以后，用户输入一句话，终端里显示识别过程、工具调用和最终回答。

我还保留了 slash 菜单，但它只是辅助，不是主入口。用户如果知道自己要做数据库任务，可以输入 /database；如果不知道，直接问自然语言也可以。

终端里还有一个重要点是交互节奏。数据库工具调用不像普通问答，可能要等 MCP 启动、执行 SQL、再让模型总结。如果界面什么都不显示，用户很容易以为卡死。所以聊天入口不仅要能输入问题，还要能显示“正在识别问题类型”“正在调用工具”“正在整理结果”这些过程。

Skills 和路由#
Skills 怎么设计#
这里我踩过一个坑。最开始我把功能拆得很细，表结构是一个命令，查询是一个命令，健康检查是一个命令，慢 SQL 又是一个命令。看起来功能很多，但用起来还是像命令行工具。

后来我把内置能力收敛成几个场景入口：

/database 结构、查询、执行计划、索引建议、健康检查、Top SQL
/ops 慢 SQL、锁等待、死锁、日志、认证失败、健康巡检
/docs 手册问答、手册检索、索引状态和索引重建
/mcp 查看 MCP Server、tools、resources、prompts
/report 导出当前会话诊断报告
/skills 查看 Agent 能力清单
/configure-model 配置后端模型
image

对应代码也比较清楚，skill 是场景入口，不是把 MCP tool 重新包装一遍：

SYSTEM_SKILLS: tuple[Skill, …] = (
Skill(command=“/mcp”, name=“MCP 状态”, tool=“mcp.status”, safety=“diagnostic”),
Skill(command=“/database”, name=“数据库任务”, tool=“database.assistant”, safety=“readonly”),
Skill(command=“/ops”, name=“运维诊断”, tool=“ops.assistant”, safety=“readonly”),
Skill(command=“/docs”, name=“知识库”, tool=“docs.assistant”),
Skill(command=“/report”, name=“导出报告”, tool=“report.export”),
)
这样做以后，用户不需要关心底层到底是 mcp.schema 还是 mcp.query，他只要知道自己在问数据库问题。真正调用什么工具，交给 Agent 判断。

这个设计也让我后面维护起来轻松一些。如果以后 KES MCP Server 多了新的工具，我不一定要再加一个新的用户入口。只要把工具注册进来，再让 /database 或 /ops 根据意图调用就行。用户看到的还是那几个场景入口，不会越用越乱。

这里其实和我之前做智能体的经验类似：外层提示词或菜单不要太复杂，复杂逻辑尽量放到工作流或工具层。用户看到的入口越简单，越容易理解这个助手到底能做什么。

自然语言怎么命中#
路由这里也遇到过问题。最早我想得比较简单，用一些关键词去判断用户要做什么，比如看到“表结构”就进 /database，看到“MCP Server”就进 /mcp。刚开始能跑几个示例，但很快就暴露问题。

最典型的就是 kes_mcp_demo.orders。这里的 mcp 明明只是 schema 名的一部分，如果只靠字符串匹配，就很容易误判成 MCP 状态检查。还有一些口语输入也很麻烦，比如“直接看下表不就得了”“看下数据库状态”“订单列表按手机号查很慢”，这些不是靠多补几个 if 就能稳定解决的。

所以后面我把这块重写成了模型结构化识别。模型不直接回答问题，而是先输出一个意图对象，里面包括要进入哪个 skill、要调用哪个 tool、有没有表名、SQL、健康检查类型这些字段。大概是这个思路：

class IntentResult(BaseModel):
intent: str = “chat”
skill_command: str = “/chat”
tool_name: str | None = None
confidence: float = 0.0
entities: IntentEntity = Field(default_factory=IntentEntity)
readonly_sql: str | None = None
needs_clarification: bool = False
safety: str = “diagnostic”
这样用户输入一句话以后，第一步不是直接调数据库，而是先让 IntentClassifier 给出结构化结果。例如：

{
“skill_command”: “/database”,
“tool_name”: “mcp.schema”,
“entities”: {
“schema”: “kes_mcp_demo”,
“table”: “orders”
},
“safety”: “readonly”
}
这里还有一个关键点：模型可以识别意图，但不能完全相信它。比如模型有时候会把 object_type 返回成数组，也可能把 conditions 返回成字符串。为了不因为这种 JSON 小偏差直接降级成普通聊天，我在解析层做了容错，把这些常见结构先规范掉。

@field_validator(“conditions”, mode=“before”)
def _coerce_conditions(cls, value: Any) -> dict[str, Any]:
if value is None:
return {}
if isinstance(value, dict):
return value
return {“raw”: str(value)}
如果第一次分类结果是 /chat，但用户其实是在说“直接看下表不就得了”这种口语表达，我还会让模型做一次重试。第二次提示里会明确要求它从最接近的安全只读工具里选，而不是轻易要求用户澄清。这个改动以后，很多日常说法就顺了。

MCP 调用#
工具注册#
MCP 工具发现以后，统一注册到工具层。这样 /database 后面可以根据用户意图去调用不同 MCP 能力，而不是把每个 MCP 工具都暴露给用户。

self.tools.register(_tool(“mcp.schema”, “通过 MCP 采集 schema/table/column/index/constraint”, “readonly”, …))
self.tools.register(_tool(“mcp.query”, “通过 MCP 执行只读查询”, “readonly”, …))
self.tools.register(_tool(“mcp.explain”, “通过 MCP 分析执行计划”, “readonly”, …))
self.tools.register(_tool(“mcp.health”, “通过 MCP 执行健康检查”, “readonly”, …))
self.tools.register(_tool(“mcp.top_queries”, “通过 MCP 获取 Top SQL”, “readonly”, …))
self.tools.register(_tool(“mcp.index_advice”, “通过 MCP 分析索引建议”, “readonly”, …))
我比较喜欢这种方式。菜单上只看到场景入口，代码里保留清晰的工具来源。后面回答时，也可以告诉用户这次依据来自哪个工具。

还有一点是失败处理。MCP 调用失败时，不能让模型自己编一个结果，也不能只抛一串异常。比较好的方式是把失败也作为证据返回，比如连接失败、工具不可用、权限不足、SQL 不安全。这样最终回答里可以明确告诉用户这一步没有采集到结果，而不是误导用户说数据库没有问题。

证据返回#
/database 背后会继续分发到具体工具：

def _database_assistant_tool(ctx: ToolContext) -> ToolEvidence:
tool_name = ctx.intent.tool_name if ctx.intent is not None else None
if tool_name == “mcp.schema”:
return mcp_schema_tool(ctx.settings, ctx.message, ctx.intent)
if tool_name == “mcp.query”:
return mcp_query_tool(ctx.settings, ctx.message, ctx.intent)
if tool_name == “mcp.explain”:
return mcp_explain_sql_tool(ctx.settings, ctx.message, ctx.intent)
if tool_name == “mcp.index_advice”:
return mcp_index_advice_tool(ctx.settings, ctx.message, ctx.intent)
if tool_name == “mcp.health”:
return mcp_health_tool(ctx.settings, ctx.message, ctx.intent)
这里的重点是证据先落地，再交给模型总结。比如用户问表结构，先通过 MCP 拿字段、主键、外键和索引；用户问执行计划，先拿数据库返回的计划；用户问健康检查，先拿 MCP 的诊断结果。这样回答就不是模型拍脑袋，而是有真实依据。

如果工具调用失败，也要保留证据来源。比如手册检索依赖本地 Milvus，Milvus 没启动时不能只返回一个笼统的 docs.assistant 失败，而是要明确告诉用户 manual_search_tool 失败。这样后面排查时才能知道是路由没命中，还是外部依赖没起来。

安全边界#
数据库工具最怕的是看起来很智能，实际上很危险。所以我默认只让只读任务自动执行。SELECT、SHOW、WITH、EXPLAIN 可以走自动链路，UPDATE、DELETE、CREATE、DROP、VACUUM 这些都不自动执行。

这里还有一个细节：自然语言查询可以由模型生成 readonly_sql，但模型只负责“提出候选 SQL”，不能决定是否执行。真正执行前还要经过本地 SQL guard，确认它是单条只读 SQL。这样即使模型输出了 DELETE、CREATE INDEX 之类的内容，也会在工具层被拦住。

READ_ONLY_PREFIXES = (“select”, “show”, “with”, “explain”)
BLOCKED_KEYWORDS = (
“alter”, “analyze”, “call”, “copy”, “create”, “delete”, “drop”,
“execute”, “grant”, “insert”, “merge”, “reindex”, “revoke”,
“truncate”, “update”, “vacuum”,
)

def assert_read_only_sql(sql: str) -> None:
if not is_read_only_sql(sql):
raise UnsafeSqlError(“只允许执行单条只读 SQL：SELECT / SHOW / WITH / EXPLAIN”)
服务端编程这一块我没有单独做成大功能，而是放在安全边界里处理。存储过程、函数、触发器可以生成 SQL 草稿、解释参数和返回值、提醒 search_path、异常处理、事务边界和锁影响；如果是性能问题，也可以提示行级触发器频繁访问大表、函数里重复查询、缺少索引这些风险。但最终创建函数、替换过程、启用触发器，都必须人工确认后再执行。

这块我宁愿保守一点。数据库里的变更操作跟普通代码生成不一样，执行错了就可能影响真实数据。尤其是函数和触发器，很多问题不是创建时立刻暴露，而是在后续业务写入时才出现。所以默认只读是这次工具的底线，先把查询和诊断做好，再考虑更复杂的自动化。

效果验证#
查结构和查数据#
第一个验证场景就是基础查库。输入：

看下我都有哪些表
看下 orders 有哪些字段，顺便说明主键、外键和索引
查一下 orders 前 5 条
统计一下 orders 有多少条数据
image

image

image

image

这里最明显的变化是，不再靠模型猜字段。Agent 会通过 MCP 工具先采集结构，再组织回答。查数据也是一样，安全范围内能确定是只读查询，就直接执行，不再只给一段 SQL 让用户自己复制。

我特意把自然语言查询收得比较窄，只支持前几条、数量统计、简单等值条件。复杂业务口径不强行猜，宁愿让用户补充表名和字段。数据库工具保守一点，比为了显得聪明乱跑 SQL 更可靠。

比如开发一个接口前，我先问有哪些表，再问 orders 表有哪些字段，然后查几条样例数据。整个过