KingbaseES MCP Server

KingbaseES 数据库的 MCP Server，使 AI 助手具备数据库结构探索、SQL 执行、执行计划分析、索引优化和健康检查能力。

版本要求

组件	版本要求	说明
Python	3.12+（建议 3.12.x）	ksycopg2 驱动最高支持 Python 3.13；创建 uv 环境时注意版本
MCP Python SDK	2.x（mcp[cli]>=2.2.0,<3）	已适配 2.x；FastMCP 在 2.x 中更名为 MCPServer
MCP 协议	2025-11-25（SDK 2.x 协商）	经 stdio/SSE/streamable-http 三种传输验证
KingbaseES	V8R6+	本仓库在 V9（V009R200C013PS002）实测通过；暂不支持容器部署测试
数据库驱动	ksycopg2 ≥ 2.9.0	Mac / Alpine 平台不支持该驱动，可用 psycopg2 伪装访问
可选扩展	sys_hypo、sys_stat_statements	非必须；不安装则假设索引、慢查询/工作负载分析不可用

目录

• 版本要求
• 架构说明
• 产品特性
• 功能说明
• 前提条件
• 安装部署
• 配置说明
• 使用说明
• 安全说明
• 项目结构
• 许可证

架构说明

AI 从静态推理向动态交互演进，Agent 能调用 LLM、访问数据库、调用 API、执行任务。当前 LLM 与数据库之间缺少标准化交互协议，每个数据源都需要自定义实现。

MCP（Model Context Protocol，模型上下文协议）为解决这一问题设计的标准化框架，使 LLM 可以与外部数据库、API 和工具高效交互。

KingbaseES + MCP + LLM 架构

MCP Server 作为 AI 工具与 KingbaseES 之间的桥梁：

• AI 客户端 发送自然语言请求（如"找出最慢的查询"）
• MCP Server 解析请求，调用对应工具
• KingbaseES 执行查询并返回结果
• AI 客户端 展示返回结果并提供相关分析

产品特性

• 结构探索：列出 schema、表、视图，查看列/约束/索引详情
• SQL 执行：执行任意 SQL；restricted 为 AST 白名单受限模式
• 执行计划分析：EXPLAIN（unrestricted 模式下可用 EXPLAIN ANALYZE），支持假设索引模拟
• 索引优化：基于代价模型 DTA 算法，从查询负载中推荐索引
• 健康检查：7 项检查 — 索引、连接、vacuum、序列、复制、缓存、约束
• 慢查询分析：按总耗时/均值耗时/资源消耗排序的 Top N 慢查询
• 多传输方式：Stdio、SSE、Streamable HTTP

功能说明

Prompts

本项目不预定义 Prompt 模板，AI 通过自然语言对话即可触发数据库操作。

Tools

Tool	说明	参数
list_schemas	列出所有 schema，并区分系统 schema 与用户 schema	无
list_objects	列出指定 schema 下的表、视图、序列和扩展	schema_name, object_type
get_object_details	查看表/视图的列定义、约束和索引详情	schema_name, object_name, object_type
execute_sql	执行 SQL；restricted 仅允许白名单内的只读语句（如 SELECT、不带 ANALYZE 的 EXPLAIN、SHOW），禁止 VACUUM、ANALYZE 和 CREATE EXTENSION	sql
explain_query	执行 EXPLAIN，支持假设索引模拟。analyze=True 需 unrestricted 模式：restricted 下 EXPLAIN ANALYZE 被 AST 白名单拒绝	sql, analyze, hypothetical_indexes
analyze_workload_indexes	基于 sys_stat_statements 的工作负载推荐索引。默认仅纳入 calls >= 50 且平均耗时 >= 5ms 的语句（最多取 100 条）；阈值均可调，调低可避免低频库返回空结果。结果为空时返回 workload_diagnostics 诊断（扫描条数、通过阈值条数、按原因分类的跳过数），便于判断是"无语句达标"还是"语句达标但不可分析"	max_index_size_mb, method, min_calls, min_avg_time_ms, limit
analyze_query_indexes	对指定 SQL 列表推荐索引（最多 10 条）	queries, max_index_size_mb, method
analyze_db_health	执行 7 项数据库健康检查；支持单项、多项（逗号分隔）或全部检查	health_type（如 index、index,buffer 或 all）
get_top_queries	返回 Top N 慢查询（按 resources、mean_time 或 total_time 排序）	sort_by, limit（1–100）
analyze_db_config	分析数据库配置参数并给出优化建议（内存/并行/WAL/autovacuum/planner），不指定参数时分析全部已知参数	parameter（可选，如 shared_buffers）

参数约束：list_objects / get_object_details 的 object_type 仅接受 table、view、sequence、extension；get_top_queries 的 sort_by 仅接受 resources、mean_time、total_time，limit 取值 1–100；analyze_workload_indexes 的 min_calls ≥ 1、min_avg_time_ms ≥ 0、limit 取值 1–200。以上均为 schema 级校验（@validate_call + Literal），非法入参在触达数据库前即被拒绝。

Resources

除工具外，服务器还暴露两个只读资源（JSON），供客户端直接读取：

Resource URI	说明
kingbase://server/info	服务器元信息：当前访问模式、工具清单、可选扩展依赖映射
kingbase://schema/overview	Schema 概览（含系统/用户 schema 分类），等价于 list_schemas 的 JSON 形态

此外，服务器在初始化时向客户端下发 instructions，说明推荐的工作流（探索 → 查询 → 诊断 → 索引建议）以及 restricted 模式的限制与可选扩展（sys_stat_statements、sys_hypo）依赖。

前提条件

• KingbaseES V8R6+ 运行实例

  注意：

  当前版本暂不支持使用容器部署测试，数据库需要先手动安装部署。

  数据库安装包可从 Kingbase 官网下载页面获取，数据库可按照 KingbaseES 产品手册 说明安装部署。

  相关功能需要以下扩展(非必须，不安装会导致部分性能优化功能不可用)：

  扩展	依赖的功能	安装命令
  sys_hypo	假设索引模拟（explain_query、索引分析）	CREATE EXTENSION sys_hypo;
  sys_stat_statements	慢查询追踪、工作负载分析	CREATE EXTENSION sys_stat_statements;

• Python 3.12+

安装部署

1. 克隆仓库

git clone https://gitee.com/king-db/kes-mcp-server.git
cd kes-mcp-server

2. 安装 uv 包管理器

Linux：

curl -LsSf https://astral.sh/uv/install.sh | sh

Windows：

irm https://astral.sh/uv/install.ps1 | iex

pip 安装（备选）：

pip install uv

验证安装：

uv --version

3. 创建虚拟环境并安装依赖

uv venv --python 3.12

# Linux / macOS
source .venv/bin/activate

# Windows
.venv\Scripts\activate

# 安装依赖
uv pip install .

注意：

当前 Kingbase-MCP 依赖的数据库驱动 ksycopg2，在 Pypi 发布平台只发布了 Linux x86_64/Aarch64/Windows版本，其他平台版本是否支持可参考 Kingbase 官网手册说明。非 Pypi 发布包，用户需要自行从 Kingbase 官网下载页面的 Python 模块 手动下载安装部署 Ksycopg2驱动，手动部署安装可参考 KingbaseES 产品手册相关说明。

当前 Mac 或者 Alpine 平台明确不支持，因对应 KingbaseES 数据库驱动不支持相关平台，可使用 psycopg2 来伪装成 Ksycopg2来访问 KingbaseES 数据库。

Ksycopg2最高支持到 Python3.13 版本，创建 uv 环境时需要注意 Python 版本。

当前代码已适配 MCP Python SDK 2.x（mcp[cli]>=2.2.0,<3，FastMCP 已更名为 MCPServer）。

4. 配置 MCP Server

Kingbase-MCP 提供了三种传输方式 Stdio/SSE/Streamable HTTP，配置 MCP Server 时根据需要三者选其一即可：

• Stdio

  客户端启动本地子进程，经标准输入/输出传输 MCP 消息；无需开端口，配置简单，适合本机开发。

• SSE

  MCP Server 以 HTTP 提供服务，客户端连接 /sse 接收 Server-Sent Events 推送，请求经 HTTP 发送。

  可远程访问，属较早的 HTTP 传输方案。

• Streamable HTTP

  MCP Server 以 HTTP 提供服务，客户端连接 /mcp，在同一条 HTTP 会话中双向传输 MCP 消息。

  可远程访问，是当前推荐的 HTTP 传输方式。

Stdio 模式

本地输入输出流交互，MCP Server 参考配置如下：

{
  "mcpServers": {
    "mydb-kingbase-mcp": {
      "command": "uv",
      "args": [
        "--directory",
        "path/to/kingbase_mcp",
        "run",
        "mydb-kingbase-mcp",
        "--access-mode",
        "restricted"
      ],
      "env": {
        "DATABASE_URI": "kingbase://user:password@host:port/dbname"
      }
    }
  }
}

启动命令有两种等价写法，任选其一：

• 控制台脚本：mydb-kingbase-mcp [database_url] --access-mode restricted（pyproject.toml 的 [project.scripts]）
• 模块方式：python -m kingbase_mcp [database_url] --access-mode restricted（由 src/kingbase_mcp/__main__.py 提供）

例如把上面的 "run", "mydb-kingbase-mcp" 换成 "run", "python", "-m", "kingbase_mcp" 同样可用。

SSE 模式

远端启动 MCP Server 服务（需先设置 DATABASE_URI ）：

# Linux
export DATABASE_URI="kingbase://user:password@host:port/dbname"

# Windows
$env:DATABASE_URI = "kingbase://user:password@host:port/dbname"

# 启动 MCP Server 服务
uv run mydb-kingbase-mcp --transport sse --sse-host 0.0.0.0 --sse-port 8000

安全提示： 上例绑定 0.0.0.0 表示监听所有网卡，且当前版本未内置认证、DNS rebinding 防护在此绑定下不生效（详见安全说明）。生产环境建议改为默认的 localhost 绑定并置于反向代理之后。

MCP Server 参考配置（type 必须为 sse，路径为 /sse）：

{
  "mcpServers": {
    "kingbase-sse": {
      "type": "sse",
      "url": "http://127.0.0.1:8000/sse"
    }
  }
}

Streamable HTTP 模式

远端启动 MCP Server 服务（需先设置 DATABASE_URI ）：

# Linux
export DATABASE_URI="kingbase://user:password@host:port/dbname"

# Windows
$env:DATABASE_URI = "kingbase://user:password@host:port/dbname"

# 启动 MCP Server 服务
uv run mydb-kingbase-mcp --transport streamable-http --streamable-http-host 0.0.0.0 --streamable-http-port 8000

安全提示： 同上一节 —— 绑定 0.0.0.0 会对外暴露且不享受 DNS rebinding 防护，生产环境请改用 localhost 绑定 + 反向代理。

MCP Server 参考配置（type 必须为 streamableHttp，路径为 /mcp）：

{
  "mcpServers": {
    "kingbase-http": {
      "type": "streamableHttp",
      "url": "http://127.0.0.1:8000/mcp"
    }
  }
}

配置说明

连接参数

Kingbase-MCP 通过两类参数进行配置：

环境变量（配置数据库连接）

变量	必填	说明
DATABASE_URI	是	连接串：kingbase://user:password@host:port/dbname
KSYCOPG2_LIB_PATH	Windows不需要，Linux可选配置	libkci.so* 的路径( Ksycopg2驱动包默认提供)

环境变量（LLM 配置，可选）

变量	默认值	说明
KINGBASE_MCP_LLM_MODEL	gpt-4o	LLM 模型名称
KINGBASE_MCP_LLM_BASE_URL		LLM API 基础 URL（留空使用 OpenAI 默认地址）
KINGBASE_MCP_LLM_API_KEY		LLM API Key（留空使用 OPENAI_API_KEY 环境变量）

连接串格式：

kingbase://user:password@host:port/dbname

注意：

KSYCOPG2_LIB_PATH 非必须配置，Linux 下可能出现 libkci.so* 找不到的报错，此时可修改配置参数或者环境变量解决。

对应 libkci 库会通过 Ksycopg2驱动包提供，Ksycopg2一般会安装在对应 Python安装路径/site-packages/ksycopg2，uv 环境下 Ksycopg2会安装在 path/to/mcp_server/mydb-kingbase-mcp/.venv/lib/python3.12/site-packages/ksycopg2。

配置参数（以 Stdio 配置为例）：

{
  "mcpServers": {
    "mydb-kingbase-mcp": {
      "command": "uv",
      "args": [
        "--directory",
        "path/to/kingbase_mcp",
        "run",
        "mydb-kingbase-mcp",
        "--access-mode",
        "restricted"
      ],
      "env": {
        "DATABASE_URI": "kingbase://user:password@host:port/dbname",
        "KSYCOPG2_LIB_PATH": "path/to/ksycopg2"
       }
     }
  }
}

环境变量：

export KSYCOPG2_LIB_PATH=path/to/ksycopg2

CLI 参数（配置服务行为）

参数	默认值	功能说明	可配置参数值
--access-mode	restricted	SQL 访问模式	unrestricted、restricted
--transport	stdio	MCP 传输方式	stdio、sse、streamable-http
--sse-host	localhost	SSE 服务绑定地址	任意可监听地址
--sse-port	8000	SSE 服务监听端口	1-65535 的整数
--streamable-http-host	localhost	Streamable HTTP 服务绑定地址	任意可监听地址
--streamable-http-port	8000	Streamable HTTP 服务监听端口	1-65535 的整数

访问模式

模式	说明	使用场景
restricted	AST 白名单受限模式：仅允许只读 SQL，禁止 DML、DDL 和维护语句，30s 超时	生产环境受限访问
unrestricted	允许所有 SQL（DDL + DML）	开发调试

配置示例

以 TRAE SOLO CN 版本为例，登录用户后，依次点击左下角的用户名 -> 设置 -> MCP，如下图所示：

点击添加按钮，将以下 MCP Server 配置项修改成实际情况后填入，如下图所示：

连接数据库成功后，会加载对应工具，如下图所示：

使用说明

以下示例基于 TRAE 实际交互演示，展示 Kingbase-MCP 使用示例。

示例 1：连接数据库

Q: 调用mydb-kingbase-mcp，连接数据库

示例 2：列出所有表

Q: 列出 public schema 下所有的表

示例 3：查看表结构

Q: 查看 orders 表的结构

示例 4：自然语言查询

Q: 查询本月销售额前 5 的商品

示例 5：执行计划分析

Q: 分析这个查询：SELECT * FROM orders WHERE user_id = 123 AND status = 'pending'

示例 6：假设索引模拟

Q: 如果在 (user_id, status) 上加索引会怎样？

示例 7：索引优化建议

Q: 分析一下数据库最近的查询负载，帮我推荐索引

示例 8：健康检查

Q: 检查一下数据库健康状况

示例 9：慢查询排查

Q: 找出最近最耗时的 5 条查询

安全说明

• Restricted 模式：AST 白名单验证；仅允许白名单内的只读语句。禁止 DML、DDL 和维护语句，包括 VACUUM、ANALYZE 与 CREATE EXTENSION；执行层另有只读事务约束（SET TRANSACTION READ ONLY）
• 30 秒超时：防止长时间查询阻塞（仅 restricted 模式生效；unrestricted 模式当前无超时）
• 白名单函数：500+ 允许函数
• 连接池上限：最大 5 个连接

说明： restricted 是严格只读模式。需要执行 VACUUM、ANALYZE、CREATE EXTENSION 或其它写操作时，请显式使用 unrestricted 模式并配置具备相应权限的数据库用户。

CREATE EXTENSION 在 restricted 模式下一律被拒绝（在语句类型判定阶段即拦截）。因此扩展白名单在受限模式下不生效，该模式的可用扩展不随 ALLOWED_EXTENSIONS 变化。

建议：

• 生产环境使用 restricted 模式
• 使用最小权限数据库用户；restricted 模式下建议仅授予查询所需权限

HTTP 传输安全（SSE / Streamable HTTP）

Stdio 传输由客户端以子进程方式启动，进程边界即安全边界，默认无需额外配置。HTTP 传输是网络服务，请按以下现状评估部署风险：

1. 当前版本未内置认证。 SSE 与 Streamable HTTP 端点不校验 token、不提供 TLS。任何能访问该端口的客户端都可以调用全部工具。生产部署请置于反向代理（nginx/haproxy）之后，由反代负责 TLS 终止与访问控制，并绑定到内网地址。

2. DNS rebinding 防护只在绑定 loopback 时自动生效。 服务器依赖 MCP SDK 的行为：仅当绑定地址为 127.0.0.1 / localhost / ::1 时，SDK 才自动开启 Host / Origin 头校验；绑定到其他地址（如 0.0.0.0）时该防护不会自动开启。本版本尚未显式接管这一设置，因此：

   绑定地址	DNS rebinding 防护
   localhost / 127.0.0.1 / ::1（默认）	自动开启
   0.0.0.0 / :: / 其他地址	不生效

   出于此原因，建议保持默认的 loopback 绑定，由反向代理对外提供服务；若确需直接绑定 0.0.0.0，请自行在外层网络或反代上补足 Host/Origin 校验。

3. 绑定非 loopback 地址时请留意日志。 以 0.0.0.0 等非 loopback 地址启动意味着数据库服务对外暴露，请确认已有外层访问控制。

开发与测试

运行测试

项目使用 pytest 运行单元测试。部分测试需要连接真实的 KingbaseES 数据库。

当前用例规模：295 passed, 1 xfailed（配置 DATABASE_URI 连接真实库时）。未配置 DATABASE_URI 时，需连库的用例会被跳过，实测为 286 passed, 1 skipped, 1 xfailed。

仅运行单元测试（无需数据库连接）：

uv run pytest tests/unit --ignore=tests/unit/explain/test_explain_plan_real_db.py --ignore=tests/unit/database_health/test_database_health_tool.py -v

运行全部测试（需配置 DATABASE_URI）：

# Linux / macOS
export DATABASE_URI="kingbase://user:password@host:port/dbname"
uv run pytest -v

# Windows
$env:DATABASE_URI = "kingbase://user:password@host:port/dbname"
uv run pytest -v

代码检查

# Ruff lint + format
uv run ruff check .
uv run ruff format .

# Pyright 类型检查
uv run pyright

快速开发

# 使用 justfile 快捷命令
just test    # 运行测试
just dev     # 启动 MCP Server（stdio 模式）
just build   # 构建发行包

项目结构

mydb-kingbase-mcp/
├── src/
│   └── kingbase_mcp/                      核心源码目录
│       ├── server.py                      MCP Server 入口、工具注册与 CLI 参数解析
│       ├── artifacts.py                   执行计划结果数据模型定义
│       ├── sql/                           数据库驱动与 SQL 安全控制
│       │   ├── sql_driver.py              连接池与 SQL 执行封装
│       │   ├── safe_sql.py                AST 白名单校验与受限执行约束
│       │   ├── bind_params.py             $N 占位符与统计信息解析
│       │   ├── index.py                   索引定义数据结构
│       │   └── extension_utils.py         扩展可用性与版本检测
│       ├── explain/
│       │   └── explain_plan.py            执行计划分析与假设索引模拟
│       ├── index/                         索引推荐与优化实现
│       │   ├── dta_calc.py                DTA 代价模型优化器
│       │   ├── llm_opt.py                 LLM 辅助索引推荐
│       │   ├── index_opt_base.py          索引优化通用基类与流程编排
│       │   └── presentation.py            索引优化结果格式化输出
│       ├── top_queries/
│       │   └── top_queries_calc.py        慢查询提取与排序分析
│       └── database_health/               数据库健康检查（7 类）
│           ├── database_health.py         健康检查路由分发入口
│           ├── index_health_calc.py       无效/重复/膨胀索引检查
│           ├── connection_health_calc.py  连接利用率检查
│           ├── vacuum_health_calc.py      事务 ID 回卷风险检查
│           ├── sequence_health_calc.py    序列耗尽风险检查
│           ├── replication_calc.py        复制延迟与复制槽检查
│           ├── buffer_health_calc.py      缓存命中率检查
│           └── constraint_health_calc.py  无效约束检查
└── tests/
    ├── conftest.py                        测试工具及初始化
    └── unit/                              单元测试用例

许可证

Apache License 2.0，详见 LICENSE 。