1. 引言

金仓数据库（KingbaseES）在国产数据库领域有着广泛的应用，尤其在政务、金融等关键行业，其稳定性和兼容性有目共睹。但日常运维和开发中，我总觉得还差那么一点“智能”——能不能像聊天一样查询数据库？能不能让 AI 帮我自动发现慢查询并给出优化建议？

最近 MCP（Model Context Protocol）很火，它让 AI 模型能安全地连接外部工具和数据源。于是，我动手给金仓写了一个 MCP Server，把“自然语言问数”和“AI 自动调优”这两个能力作为标准工具暴露出来，让 LLM 可以直接调用。这篇文章，就是我的实践记录。

2. 什么是 MCP Server？为什么需要它？

MCP 是 Anthropic 提出的一种开放协议，用于标准化 AI 模型与外部工具、数据源的交互方式。一个 MCP Server 可以暴露以下三类能力：

• Resources：向模型提供数据（例如数据库表结构、查询日志）。
• Tools：模型可以调用的函数（例如执行 SQL、获取执行计划）。
• Prompts：预定义的提示模板。

简单来说，MCP 就像 USB 协议，让任何 AI 客户端（如 Claude Desktop、支持 MCP 的 IDE）都能即插即用地接入你的服务。我给金仓写的 MCP Server，核心就是通过 Tools 让模型能够“问数”和“调优”。

3. 金仓数据库的痛点与机会

在日常使用金仓过程中，我经常遇到这几类问题：

• 业务人员想查数据，但不会写 SQL，需要找开发人员翻译。
• 开发人员写了一条复杂 SQL，性能差，但不知道从哪里优化。
• 数据库有很多慢查询，需要人工定期分析，效率低。

这些痛点，恰好可以用 AI 来解决：自然语言转 SQL 解决“问数”；自动分析执行计划、索引建议解决“调优”。而 MCP Server 就是连接 AI 和金仓的桥梁。

4. 整体设计思路

我将 MCP Server 的能力分为两个核心模块：

1. 自然语言问数：通过大模型将用户问题转为 SQL，执行并返回结果，同时附带解释。
2. AI 自动调优：分析慢查询日志、获取执行计划、检查索引/统计信息，给出优化建议，甚至可以自动执行简单的 DDL。

整个系统流程如下：

MCP Server 本身是一个 Python 服务，通过 psycopg2 连接金仓，使用 mcp 库实现协议。

5. MCP Server 架构与实现

5.1 环境准备

首先安装依赖：

pip install mcp psycopg2-binary python-dotenv

5.2 核心框架

MCP Server 的核心是定义一个 Server 并注册工具。代码骨架如下：

from mcp.server import Server, NotificationOptions
from mcp.server.models import InitializationCapabilities
from mcp.server.stdio import stdio_server
import asyncio
import psycopg2
import os

server = Server("kingbase-mcp")

# 数据库连接配置
DB_CONFIG = {
    "dbname": os.getenv("KINGBASE_DB", "test"),
    "user": os.getenv("KINGBASE_USER", "system"),
    "password": os.getenv("KINGBASE_PASSWORD", "123456"),
    "host": os.getenv("KINGBASE_HOST", "localhost"),
    "port": os.getenv("KINGBASE_PORT", "54321")
}

def get_connection():
    return psycopg2.connect(**DB_CONFIG)

这里我们使用 psycopg2，因为金仓兼容 PostgreSQL 协议，驱动可以直接使用。

5.3 注册工具

MCP 工具通过装饰器 @server.list_tools() 和 @server.call_tool() 注册。我们注册两个工具：natural_language_query 和 auto_tune。

@server.list_tools()
async def handle_list_tools() -> list:
    return [
        Tool(
            name="natural_language_query",
            description="将自然语言问题转换为 SQL 查询并执行，返回结果及解释",
            inputSchema={
                "type": "object",
                "properties": {
                    "question": {"type": "string", "description": "用户自然语言问题"}
                },
                "required": ["question"]
            }
        ),
        Tool(
            name="auto_tune",
            description="分析慢查询或指定 SQL 的执行计划，给出索引、SQL 改写等优化建议",
            inputSchema={
                "type": "object",
                "properties": {
                    "sql": {"type": "string", "description": "待优化的 SQL 语句（可选，不填则从慢查询日志提取）"}
                }
            }
        )
    ]

工具调用核心逻辑在 handle_call_tool 中实现，下面分别展开。

6. 自然语言问数实现

这一步需要用到 LLM 将自然语言转为 SQL。由于 MCP Server 本身不内置大模型，我们可以通过调用外部 API（如 OpenAI 兼容接口）来完成。为了简化，这里展示一个调用本地 LLM 的示例，实际部署时可按需替换。

@server.call_tool()
async def handle_call_tool(name: str, arguments: dict) -> list:
    if name == "natural_language_query":
        question = arguments["question"]
        sql = await nl_to_sql(question)  # 自然语言转 SQL
        if not sql:
            return [TextContent(type="text", text="无法生成有效 SQL，请检查问题描述。")]
        # 执行 SQL
        conn = get_connection()
        cur = conn.cursor()
        try:
            cur.execute(sql)
            rows = cur.fetchall()
            columns = [desc[0] for desc in cur.description]
            result = f"查询成功，返回 {len(rows)} 行。\nSQL: {sql}\n"
            result += format_table(columns, rows)  # 格式化表格
            return [TextContent(type="text", text=result)]
        except Exception as e:
            return [TextContent(type="text", text=f"SQL 执行失败: {str(e)}")]
        finally:
            cur.close()
            conn.close()

其中 nl_to_sql 的实现依赖于 LLM，我们可以这样做：

async def nl_to_sql(question: str) -> str:
    # 获取表结构信息作为上下文
    schema = get_schema_info()
    prompt = f"""你是一个 SQL 专家，请根据以下表结构，将用户问题转为一条 KingbaseES 兼容的 SQL 查询。
表结构：
{schema}
用户问题：{question}
只返回 SQL 语句，不要解释。"""
    # 调用 LLM API（这里以 OpenAI 格式为例）
    response = await openai_call(prompt)
    return response.strip()

get_schema_info 直接从金仓系统表中查询所有用户表及其字段，拼成提示，这样模型就能写出准确的 SQL。

7. AI 自动调优实现

自动调优工具的核心是分析执行计划，并结合索引、统计信息给出建议。金仓支持 EXPLAIN 命令，我们可以通过解析执行计划来发现性能瓶颈。

    elif name == "auto_tune":
        sql = arguments.get("sql")
        if not sql:
            # 从慢查询日志提取一条耗时最长的 SQL
            sql = await get_slowest_query()
            if not sql:
                return [TextContent(type="text", text="未找到慢查询记录。")]
        # 获取执行计划
        plan = await get_explain_plan(sql)
        if not plan:
            return [TextContent(type="text", text="无法获取执行计划。")]
        # 分析执行计划
        advice = analyze_plan(plan)
        result = f"优化建议：\n{advice}"
        return [TextContent(type="text", text=result)]

其中 get_explain_plan 执行 EXPLAIN (ANALYZE, FORMAT JSON) 并返回 JSON 格式的执行计划，方便程序化解析。analyze_plan 可以基于规则来判断：

• 出现 Seq Scan 且扫描行数很大 → 建议建索引。
• 出现 Nested Loop 且内表扫描次数多 → 建议改写为 Hash Join 或建索引。
• 统计信息过期 → 建议执行 ANALYZE。

示例建议输出：

表 orders 的 order_date 列进行了全表扫描，建议创建索引：
CREATE INDEX idx_orders_order_date ON orders(order_date);

如果希望自动执行，可以增加一个 enable_auto_execute 参数，但出于安全考虑，默认只给出建议。

8. 与金仓的集成细节

金仓默认使用 pg_catalog 兼容模式，因此大部分 SQL 可以直接使用。但需要注意：

• 连接驱动用 psycopg2，端口可能不是默认的 5432。
• 系统表查询如 information_schema.tables 表现与 PostgreSQL 一致。
• 慢查询日志需要金仓开启 log_min_duration_statement 参数，MCP Server 通过读取日志文件或查询 pg_stat_statements 扩展来获取慢查询。

我选择使用 pg_stat_statements 扩展（需提前安装），因为它能提供更结构化的查询统计信息。

9. 效果展示

将 MCP Server 配置到 Claude Desktop 后，我可以直接用自然语言提问：

用户：
“查询最近一周销售额最高的前10个商品。”

模型会自动调用 natural_language_query 工具，生成 SQL 并返回表格结果：

查询成功，返回 10 行。
SQL: SELECT product_id, SUM(amount) AS total_sales FROM orders WHERE order_date >= CURRENT_DATE - INTERVAL '7 days' GROUP BY product_id ORDER BY total_sales DESC LIMIT 10;
| product_id | total_sales |
|------------|-------------|
| 1001       | 45600.00    |
| 1002       | 38900.00    |
...

用户：
“帮我优化一下这条 SQL。”

模型自动调用 auto_tune，获取执行计划并给出建议：

优化建议：
- orders 表缺少 order_date 索引，建议创建：
  CREATE INDEX idx_orders_order_date ON orders(order_date);
- 该查询涉及大量聚合，建议开启并行查询：
  SET max_parallel_workers_per_gather = 4;

整个交互过程流畅，运维和开发效率大幅提升。

10. 总结与展望

这个 MCP Server 打通了金仓数据库与 AI 之间的壁垒，让自然语言交互和智能调优成为现实。通过标准化协议，任何支持 MCP 的客户端都能直接接入，降低了使用门槛。

未来我还计划扩展以下能力：

• 支持更多数据库对象（视图、存储过程）的查询。
• 与监控系统集成，实现自动巡检和告警。
• 允许 AI 在安全沙箱内执行 DDL，自动完成索引创建等操作。

代码已开源在 GitHub，欢迎试用和提 issue。如果你也在用国产数据库，不妨试试用 MCP 给它加点“智能”，或许会有意想不到的收获。

项目地址：https://github.com/your-username/kingbase-mcp-server （示例）