1. 项目概述当AI开始“动手”操作数据库最近在折腾AI应用开发的朋友估计都绕不开一个核心问题如何让大模型不只是“纸上谈兵”而是能真正地、安全地去执行一些具体的操作比如查询、分析甚至管理数据库。我们总不能让AI每次回答“帮我查一下上个月的销售数据”时都只能生成一段SQL代码然后还得我们手动复制粘贴到数据库客户端去执行吧这体验就太割裂了。这个需求催生了一个关键的技术组件——MCP Server。MCP全称是Model Context Protocol你可以把它理解成AI大模型和外部工具、数据源之间的一座“标准化桥梁”。它定义了一套协议让像Claude、GPTs这类AI助手能够发现、调用并安全地使用外部的功能比如读取文件、执行代码当然还有我们今天要聊的操作数据库。而一个针对特定数据库的MCP Server就是这座桥梁在数据库领域的具体实现它封装了连接、认证、SQL执行、结果处理等一系列复杂且敏感的操作。那么当这个数据库是**电科金仓KingbaseESKES**时事情就变得更有趣了。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 fpostgresql://{settings.kes_user}:{settings.kes_password}{settings.kes_host}:{settings.kes_port}/{settings.kes_database} # 注意asyncpg默认使用PostgreSQL端口5432KES通常是54321需要在DSN或参数中指定 # 更稳妥的方式是使用参数字典 cls._pool await asyncpg.create_pool( hostsettings.kes_host, portsettings.kes_port, usersettings.kes_user, passwordsettings.kes_password, databasesettings.kes_database, min_sizesettings.kes_pool_min_size, max_sizesettings.kes_pool_max_size, # 关键设置语句执行超时防止AI发送死循环查询 command_timeout30.0, ) logger.info(KES数据库连接池创建成功, hostsettings.kes_host, databasesettings.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版本上测试基本功能如连接、简单查询。如果遇到兼容性问题可以回退到使用aiopgpsycopg2的异步封装或同步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, fSQL语句包含禁止的操作关键字: {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(“正在执行查询”, sqlsql[:100]) # 日志只记录前100字符 # 使用fetch获取所有结果。对于超大结果集应考虑分页或流式返回。 rows await conn.fetch(sql) # 将asyncpg.Record对象转换为字典列表 result [dict(row) for row in rows] return json.dumps({status: “success”, “data”: result}, defaultstr) # defaultstr处理日期等不可序列化对象 except Exception as e: logger.error(“查询执行失败”, sqlsql, errorstr(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}, defaultstr) except Exception as e: logger.error(“获取表模式失败”, schemaschema, tabletable_name, errorstr(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}, defaultstr) except Exception as e: logger.error(“列出表失败”, schemaschema, errorstr(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_namename, argumentsarguments) 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”, textresult)] # 使用标准输入输出与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从“顾问”真正转变为能够动手解决问题的“助手”。