基于MCP协议构建KES数据库AI操作层：安全架构与工程实践-开发者社区

1. 项目概述：当AI开始“动手”操作数据库

最近在折腾AI应用开发的朋友，估计都绕不开一个核心问题：如何让大模型不只是“纸上谈兵”，而是能真正地、安全地去执行一些具体的操作，比如查询、分析甚至管理数据库。我们总不能让AI每次回答“帮我查一下上个月的销售数据”时，都只能生成一段SQL代码，然后还得我们手动复制粘贴到数据库客户端去执行吧？这体验就太割裂了。这个需求催生了一个关键的技术组件——MCP Server。

MCP，全称是Model Context Protocol，你可以把它理解成AI大模型和外部工具、数据源之间的一座“标准化桥梁”。它定义了一套协议，让像Claude、GPTs这类AI助手能够发现、调用并安全地使用外部的功能，比如读取文件、执行代码，当然，还有我们今天要聊的操作数据库。而一个针对特定数据库的MCP Server，就是这座桥梁在数据库领域的具体实现，它封装了连接、认证、SQL执行、结果处理等一系列复杂且敏感的操作。

那么，当这个数据库是**电科金仓KingbaseES（KES）**时，事情就变得更有趣了。KES作为一款重要的国产数据库，在企业级应用、政务系统中有着广泛的应用。让AI能力无缝集成到KES的运维、数据分析乃至业务开发流程中，无疑能极大提升效率。本文要分享的，就是基于KES构建一个专属MCP Server的完整实践。这不是一个简单的“Hello World” demo，而是从架构设计、安全考量、到具体实现和深度优化的全过程记录，其中踩过的坑、总结的经验，或许能为你正在进行的AI Agent或智能助手项目提供直接的参考。

2. 核心思路与架构设计：为什么是MCP，以及如何为KES量身定制

在开始敲代码之前，我们得先想清楚几个根本问题：为什么选择MCP协议而不是自己写一套API？这个Server的核心职责是什么？整体的技术栈如何选型？

2.1 为何选择MCP协议：标准化与生态优势

最初我们评估过几种方案：一是为AI单独开发一套RESTful API，二是使用像LangChain Tools这样的框架。但最终选择MCP，主要基于以下几点考量：

协议标准化，而非框架绑定：MCP是一个开放协议，不绑定任何特定的AI前端（如Claude Desktop、Cursor）或后端框架。这意味着我们今天为KES写的Server，明天可以同样被集成到其他支持MCP的AI工作流中，可移植性极强。相比之下，如果基于某个特定框架（如LangChain）开发，虽然初期快，但容易被框架的演进所绑架。
强大的工具发现与描述能力：MCP协议要求Server向AI客户端清晰地“自我介绍”，说明自己提供了哪些“工具”（Tools），每个工具需要什么参数，参数是什么类型。这种自描述特性让AI能动态地理解和使用数据库功能，无需在AI模型训练时硬编码。
原生支持复杂数据类型与流式响应：数据库查询结果可能是结构化的表、文本，甚至是二进制数据（如图片）。MCP协议对资源（Resources）和内容（Contents）有良好的抽象，能更好地处理这些复杂情况。对于大数据集，它还支持流式（Streaming）返回，避免一次性传输造成的阻塞。
日益壮大的生态：随着Claude等AI产品大力推广MCP，其生态正在快速发展。使用MCP意味着我们的工作能更容易地融入这个生态，获得兼容性红利。

2.2 KES MCP Server的核心职责与边界

我们的Server目标很明确：成为一个安全、可靠、高效的中间层。它的核心职责包括：

连接管理：维护与一个或多个KES数据库实例的连接池，处理连接的生命周期（创建、验证、复用、释放）。
SQL执行与安全隔离：接收AI客户端发来的SQL语句或自然语言请求（由AI转换为SQL），执行它，并返回结果。这里的安全隔离至关重要，Server必须严格限制AI的操作范围，例如，禁止执行DROP DATABASE、TRUNCATE TABLE这类高危语句。
结果格式化与适配：将KES数据库返回的原始数据（可能是Python的tuple、list或dict）转换为MCP协议规定的、AI易于理解的格式（通常是JSON）。
工具（Tools）暴露：将数据库操作封装成一个个具体的“工具”。例如：
- execute_query: 执行一个SELECT查询，返回表格数据。
- get_table_schema: 获取指定表的字段名、类型等结构信息。
- list_tables: 列出当前数据库中的所有表（可限定模式）。
- （高级）analyze_query_performance: 对某条SQL执行EXPLAIN分析。

注意：我们刻意不在Server端实现“自然语言转SQL”（NL2SQL）的功能。这个能力应该由前端的AI大模型来负责。Server的输入应该是明确的、经过AI初步处理的SQL语句或结构化请求。这样职责分离更清晰，也便于调试和审计。

2.3 技术栈选型：Python + psycopg2 + mcp

基于快速原型开发和KES的官方驱动支持，我们选择了以下技术栈：

语言：Python。生态丰富，异步支持好，与MCP开发库集成方便。
KES驱动：psycopg2或kingbase官方Python驱动。psycopg2是PostgreSQL协议的事实标准，由于KES高度兼容PostgreSQL，使用psycopg2通常兼容性最好，社区资源也最丰富。如果遇到特定版本的不兼容问题，再考虑切换至官方驱动。
MCP框架：官方提供的mcpPython SDK。它提供了构建Server所需的底层协议通信、工具和资源注册等基础能力，让我们能专注于业务逻辑。
异步框架：asyncio。MCP协议通信本质上是异步的，使用异步IO能更好地处理并发请求，提高Server的吞吐量。
配置管理：使用pydantic-settings管理数据库连接参数、安全规则等配置，支持从环境变量、配置文件读取，安全且灵活。
日志与监控：structlog用于结构化日志，方便后续接入ELK等监控系统。

这个组合在保证功能强大的同时，也兼顾了开发效率和运行性能。

3. 从零到一：构建KES MCP Server的详细步骤

理论说得再多，不如一行代码。接下来，我们一步步搭建这个Server。假设你已经有一个可用的KES数据库实例（版本V8R6或以上），并且准备好了Python 3.9+的环境。

3.1 环境准备与依赖安装

首先，创建一个干净的虚拟环境并安装核心依赖。

# 创建项目目录并进入 mkdir kes-mcp-server && cd kes-mcp-server python -m venv venv # 激活虚拟环境 (Linux/macOS) source venv/bin/activate # Windows: venv\Scripts\activate # 安装核心依赖 pip install mcp psycopg2-binary pydantic-settings structlog # 可选：安装开发工具 pip install black isort mypy

这里选择psycopg2-binary是为了避免编译依赖，简化部署。pydantic-settings用于管理配置。

3.2 核心配置与连接池管理

数据库连接参数和安全规则不应该硬编码在代码里。我们创建一个config.py来管理。

# config.py from pydantic_settings import BaseSettings from typing import List class Settings(BaseSettings): # 数据库连接配置 kes_host: str = "localhost" kes_port: int = 54321 kes_database: str = "testdb" kes_user: str = "mcp_user" kes_password: str = "" # 连接池大小 kes_pool_min_size: int = 2 kes_pool_max_size: int = 10 # 安全规则：禁止执行的SQL关键字列表 forbidden_sql_keywords: List[str] = [ "DROP DATABASE", "DROP SCHEMA", "TRUNCATE TABLE", "ALTER SYSTEM", "VACUUM FULL", # 可以根据需要扩展 ] # 允许访问的模式（数据库schema），为空表示不限制 allowed_schemas: List[str] = ["public", "sales"] class Config: env_file = ".env" # 从.env文件加载配置 settings = Settings()

然后，我们实现一个带连接池的数据库管理器。直接为每个请求创建新连接开销巨大，连接池是必须的。

# db_manager.py import asyncpg # 这里使用asyncpg，因为它对异步和连接池支持更原生。KES兼容PostgreSQL协议。 from contextlib import asynccontextmanager from config import settings import structlog logger = structlog.get_logger() class KESDatabaseManager: _pool = None @classmethod async def get_pool(cls): """获取数据库连接池（单例）""" if cls._pool is None: dsn = f"postgresql://{settings.kes_user}:{settings.kes_password}@{settings.kes_host}:{settings.kes_port}/{settings.kes_database}" # 注意：asyncpg默认使用PostgreSQL端口5432，KES通常是54321，需要在DSN或参数中指定 # 更稳妥的方式是使用参数字典 cls._pool = await asyncpg.create_pool( host=settings.kes_host, port=settings.kes_port, user=settings.kes_user, password=settings.kes_password, database=settings.kes_database, min_size=settings.kes_pool_min_size, max_size=settings.kes_pool_max_size, # 关键：设置语句执行超时，防止AI发送死循环查询 command_timeout=30.0, ) logger.info("KES数据库连接池创建成功", host=settings.kes_host, database=settings.kes_database) return cls._pool @classmethod @asynccontextmanager async def get_connection(cls): """从连接池获取一个连接，用完后自动归还""" pool = await cls.get_pool() conn = await pool.acquire() try: yield conn finally: await pool.release(conn) @classmethod async def close_pool(cls): """关闭连接池""" if cls._pool: await cls._pool.close() logger.info("KES数据库连接池已关闭")

实操心得：关于驱动选择。虽然开始提到了psycopg2，但在异步环境下，asyncpg的性能通常更优。由于KES高度兼容PostgreSQL协议，asyncpg在大多数场景下工作良好。但在使用前，务必在你的KES版本上测试基本功能（如连接、简单查询）。如果遇到兼容性问题，可以回退到使用aiopg（psycopg2的异步封装）或同步psycopg2配合线程池。

3.3 实现MCP工具（Tools）：封装数据库操作

这是Server的核心。我们将创建几个最常用的工具。首先，需要一个安全的SQL执行器。

# security.py import re from config import settings class SQLSecurityChecker: @staticmethod def is_sql_safe(sql: str) -> tuple[bool, str]: """ 检查SQL语句是否安全。 返回 (是否安全, 错误信息) """ sql_upper = sql.upper().strip() # 1. 检查是否包含禁止的关键字 for keyword in settings.forbidden_sql_keywords: # 使用单词边界正则匹配，避免误伤（如‘information’中包含‘drop’） pattern = r'\b' + re.escape(keyword.upper()) + r'\b' if re.search(pattern, sql_upper): return False, f"SQL语句包含禁止的操作关键字: {keyword}" # 2. 检查是否试图访问未授权的模式（简化版，通过解析FROM/JOIN后的表名） # 注意：这是一个简化的实现，复杂的嵌套子查询或CTE可能解析不全。 # 生产环境应考虑使用更完善的SQL解析库（如sqlglot）或依赖数据库自身的权限系统。 if settings.allowed_schemas: # 一个简单的表名提取正则（不处理带空格的情况） table_pattern = r'\bFROM\s+(\w+\.\w+|\w+)\b' matches = re.findall(table_pattern, sql_upper, re.IGNORECASE) for match in matches: if '.' in match: schema, _ = match.split('.') if schema.upper() not in [s.upper() for s in settings.allowed_schemas]: return False, f"试图访问未授权的模式: {schema}" # 如果没有指定模式，默认为public或其他，这里可以根据数据库默认模式判断 # 为安全起见，可以要求所有表必须显式指定模式 # else: # return False, “表名必须包含模式前缀，例如 public.table_name” # 3. 可以添加更多检查，如是否包含多个分号（尝试执行多条语句）等 if sql_upper.count(';') > 1: return False, “SQL语句包含多个分号，可能试图执行多条语句” return True, “”

接下来，在tools.py中实现具体的MCP工具。

# tools.py import mcp.types as types from db_manager import KESDatabaseManager from security import SQLSecurityChecker import structlog import json logger = structlog.get_logger() async def execute_query(arguments: dict) -> str: """执行查询SQL并返回JSON格式结果""" sql = arguments.get("sql") if not sql: return json.dumps({"error": “未提供SQL语句”}) # 安全检查 is_safe, msg = SQLSecurityChecker.is_sql_safe(sql) if not is_safe: return json.dumps({"error": f“安全检查失败: {msg}”}) try: async with KESDatabaseManager.get_connection() as conn: logger.info(“正在执行查询”, sql=sql[:100]) # 日志只记录前100字符 # 使用fetch获取所有结果。对于超大结果集，应考虑分页或流式返回。 rows = await conn.fetch(sql) # 将asyncpg.Record对象转换为字典列表 result = [dict(row) for row in rows] return json.dumps({"status": “success”, “data”: result}, default=str) # default=str处理日期等不可序列化对象 except Exception as e: logger.error(“查询执行失败”, sql=sql, error=str(e)) return json.dumps({"status": “error”, “message”: str(e)}) async def get_table_schema(arguments: dict) -> str: """获取指定表的模式信息""" schema = arguments.get(“schema”, “public”) table_name = arguments.get(“table_name”) if not table_name: return json.dumps({"error": “未提供表名”}) # 检查模式是否授权 if schema not in settings.allowed_schemas: return json.dumps({"error": f“模式 {schema} 未授权访问”}) sql = “”” SELECT column_name, data_type, is_nullable, column_default FROM information_schema.columns WHERE table_schema = $1 AND table_name = $2 ORDER BY ordinal_position; “”” try: async with KESDatabaseManager.get_connection() as conn: rows = await conn.fetch(sql, schema, table_name) result = [dict(row) for row in rows] return json.dumps({"status": “success”, “schema”: schema, “table”: table_name, “columns”: result}, default=str) except Exception as e: logger.error(“获取表模式失败”, schema=schema, table=table_name, error=str(e)) return json.dumps({"status": “error”, “message”: str(e)}) async def list_tables(arguments: dict) -> str: """列出指定模式下的所有表""" schema = arguments.get(“schema”, “public”) if schema not in settings.allowed_schemas: return json.dumps({"error": f“模式 {schema} 未授权访问”}) sql = “”” SELECT table_name, table_type FROM information_schema.tables WHERE table_schema = $1 ORDER BY table_name; “”” try: async with KESDatabaseManager.get_connection() as conn: rows = await conn.fetch(sql, schema) result = [dict(row) for row in rows] return json.dumps({"status": “success”, “schema”: schema, “tables”: result}, default=str) except Exception as e: logger.error(“列出表失败”, schema=schema, error=str(e)) return json.dumps({"status": “error”, “message”: str(e)}) # 工具定义，用于向MCP客户端注册 def get_tools(): return [ types.Tool( name=“execute_query”, description=“执行一个SELECT查询SQL语句，返回JSON格式的结果。请确保SQL语法正确且安全。”, inputSchema={ “type”: “object”, “properties”: { “sql”: { “type”: “string”, “description”: “要执行的SELECT查询语句” } }, “required”: [“sql”] } ), types.Tool( name=“get_table_schema”, description=“获取数据库中指定表的字段结构（模式）信息。”, inputSchema={ “type”: “object”, “properties”: { “schema”: { “type”: “string”, “description”: “模式名，默认为‘public’”, “default”: “public” }, “table_name”: { “type”: “string”, “description”: “表名” } }, “required”: [“table_name”] } ), types.Tool( name=“list_tables”, description=“列出数据库中指定模式下的所有表。”, inputSchema={ “type”: “object”, “properties”: { “schema”: { “type”: “string”, “description”: “模式名，默认为‘public’”, “default”: “public” } } } ), ]

3.4 组装Server并运行

最后，我们创建主程序main.py，将上述模块组装成一个完整的MCP Server。

# main.py import asyncio import mcp.server.stdio from mcp.server import Server import mcp.server.models as models from tools import get_tools, execute_query, get_table_schema, list_tables from db_manager import KESDatabaseManager import structlog logger = structlog.get_logger() async def main(): # 初始化数据库连接池 await KESDatabaseManager.get_pool() # 创建MCP Server实例 server = Server(“kes-mcp-server”) # 注册工具 @server.list_tools() async def handle_list_tools() -> list[models.Tool]: return get_tools() @server.call_tool() async def handle_call_tool(name: str, arguments: dict) -> list[models.TextContent]: logger.info(“收到工具调用请求”, tool_name=name, arguments=arguments) if name == “execute_query”: result = await execute_query(arguments) elif name == “get_table_schema”: result = await get_table_schema(arguments) elif name == “list_tables”: result = await list_tables(arguments) else: result = json.dumps({"error": f“未知工具: {name}”}) return [models.TextContent(type=“text”, text=result)] # 使用标准输入输出与MCP客户端通信 async with mcp.server.stdio.stdio_server() as (read_stream, write_stream): await server.run(read_stream, write_stream, server.create_initialization_options()) if __name__ == “__main__”: asyncio.run(main())

现在，一个最基本的KES MCP Server就完成了。你可以通过标准输入输出流来运行它，这通常由支持MCP的AI客户端（如Claude Desktop）来启动和管理。

4. 安全加固与生产级考量

上面实现的是一个基础版本。要用于实际环境，尤其是让AI直接操作生产数据库，安全是重中之重。以下是我们必须考虑的加固点：

4.1 纵深防御策略

专用数据库账户：绝对不要使用高权限账户（如sa、postgres）。创建一个仅具备必要权限的专用账户。例如，只授予对特定模式（schema）的SELECT权限，可能还有几个视图（VIEW）的SELECT权限。

-- 在KES中创建专用用户并授权 CREATE USER mcp_user WITH PASSWORD ‘StrongPassword123!’; GRANT CONNECT ON DATABASE testdb TO mcp_user; GRANT USAGE ON SCHEMA public TO mcp_user; GRANT SELECT ON ALL TABLES IN SCHEMA public TO mcp_user; -- 未来新建的表默认没有权限，需要手动授权或修改默认权限 ALTER DEFAULT PRIVILEGES IN SCHEMA public GRANT SELECT ON TABLES TO mcp_user;

网络隔离与防火墙：MCP Server应该部署在能与AI客户端和KES数据库都通信的网络位置，但最好是在一个受保护的内部网络。确保KES数据库的监听端口（默认54321）不直接暴露在公网。在Server与数据库之间配置防火墙规则，只允许Server的IP访问数据库端口。
SQL注入与语句过滤：我们之前的SQLSecurityChecker只是一个初级的词法过滤。它不能完全防止SQL注入，因为AI生成的SQL本身可能就是恶意的或错误的。更安全的做法是：
- 白名单机制：对于某些固定操作（如“查询最近7天订单”），可以将其映射为预定义的、参数化的SQL模板，AI只能选择模板和提供参数值。
- 使用参数化查询：我们的execute_query工具目前是直接拼接SQL。虽然AI提供的是一整条语句，但我们可以尝试对其中用户输入的部分（如果未来支持）进行参数化。不过，对于AI生成的完整SQL，参数化意义不大，因为整个语句都是动态的。此时，严格的数据库账户权限是最后也是最关键的防线。
- 查询复杂度与成本限制：在数据库层面设置statement_timeout或max_execution_time，防止AI意外生成一个消耗大量资源的复杂查询（如多表笛卡尔积）拖垮数据库。我们在创建连接池时设置的command_timeout就是为此。
审计与日志：所有经由MCP Server执行的SQL语句、执行结果（可脱敏）、调用者（AI会话ID）、时间戳都必须详细记录。structlog结合JSON格式输出，可以很方便地将日志发送到像Loki或Elasticsearch这样的集中式日志系统，便于事后审计和问题排查。

4.2 性能与稳定性优化

连接池调优：min_size和max_size需要根据实际并发量调整。设置过小会导致频繁新建连接，过大则浪费资源。可以结合监控观察连接数波动。
结果集分页：execute_query工具目前是返回所有结果。如果查询结果很大，会占用大量内存和网络带宽。应该实现分页功能，在工具定义中增加limit和offset参数，并在SQL中自动添加LIMIT和OFFSET子句（或使用KES的FETCH语法）。
健康检查与熔断：Server应定期检查数据库连接池的健康状况。如果数据库连续不可用，应进入熔断状态，快速失败并向AI客户端返回明确错误，而不是让请求一直挂起。
配置热更新：安全规则（如forbidden_sql_keywords）可能需要动态调整。可以实现一个简单的API端点或信号机制，在不重启Server的情况下重新加载配置。

5. 与AI客户端集成：以Claude Desktop为例

Server写好了，怎么用呢？这里以目前对MCP支持较好的Claude Desktop为例。

配置Claude Desktop：在Claude Desktop的配置文件中（通常位于~/Library/Application Support/Claude/claude_desktop_config.jsonon macOS），添加我们的Server配置。

{ “mcpServers”: { “kes”: { “command”: “/path/to/your/venv/bin/python”, “args”: [“/path/to/your/kes-mcp-server/main.py”], “env”: { “KES_HOST”: “your-kes-host”, “KES_USER”: “mcp_user”, “KES_PASSWORD”: “your-strong-password”, “KES_DATABASE”: “testdb” // 其他环境变量... } } } }

重启Claude Desktop：重启后，Claude应该能自动发现并连接上我们的KES MCP Server。
在对话中使用：现在，你可以在Claude的对话中直接说：“请使用KES工具，列出public模式下的所有表。” Claude会识别出可用的list_tables工具，并调用它，然后将结果返回给你。或者说：“帮我查询一下上个月销售额超过1万的订单详情。” Claude可能会先调用get_table_schema来了解orders表的结构，然后组合条件生成SQL，再调用execute_query执行。

6. 踩坑实录与进阶思考

在实际开发和测试中，我们遇到了几个典型问题：

KES与PostgreSQL的细微差异：虽然兼容性很高，但某些系统视图或函数名可能不同。例如，查询版本信息，PostgreSQL用SELECT version();，KES可能要用SELECT kingbase_version();。我们的get_table_schema工具使用了标准的information_schema，这在KES中通常是可用的，但仍需在目标版本上验证。
AI生成的SQL格式问题：大模型生成的SQL有时会包含Markdown代码块标记（如sql ...），或者末尾有分号，有时又没有。我们的Server需要有一定的容错性，在执行前最好做一个简单的清洗，比如去除首尾的标记和多余的空格。
错误信息处理：数据库返回的错误信息可能包含敏感信息（如数据库内部结构）。直接返回给AI客户端是不安全的。我们需要编写一个错误信息过滤器，将详细的数据库错误转换为更通用、安全的提示，如“查询语法错误”或“权限不足”，同时将详细错误记录在Server端日志中。
会话上下文管理：一个高级需求是，在同一个AI对话会话中，可能希望Server能记住之前的查询上下文。例如，用户说“对刚才查询的结果按金额排序”。这需要Server维护一定的会话状态，并将之前查询的临时结果集或标识符与当前会话关联。MCP协议本身不直接管理会话，这需要在Server内部实现一个简单的会话缓存机制，并为工具增加session_id参数。

进阶方向：

向量检索集成：如果KES中存储了文本向量，可以扩展一个semantic_search工具，接收自然语言问题，在Server端将其转换为向量，并执行向量相似度搜索，将结果返回给AI。
成为更智能的“数据助手”：不仅仅是执行SQL，可以封装更复杂的业务查询作为工具，比如get_sales_trend_last_quarter，背后对应一个预定义的、优化过的存储过程或视图。
多数据库支持：将架构抽象，使Server可以同时连接KES、MySQL等多种数据源，根据请求动态选择。

构建这个KES MCP Server的过程，本质上是在AI的“思考”能力和企业的“数据”宝库之间，铺设一条可控、可审计、高性能的管道。它打开了通往“AI原生应用”的一扇大门，让AI从“顾问”真正转变为能够动手解决问题的“助手”。