开发者日记

外观

{title} ​

把 DBA 经验封装成 MCP 技能：CI 拦截慢查询实战 ​

数据库变更不要只靠 DBA 在合并前肉眼扫一遍。我的做法是把 EXPLAIN 分析、元数据查询、Online DDL 风险判断和迁移幂等性检查封装成一组 MCP 可复用技能，再让智能体在 CI 中编排调用；上线两个迭代后，被拦截的高风险变更从“发版夜惊魂”变成了 PR 阶段的自动评论，线上由慢 SQL 引发的连接打满事件下降了约 80%。这篇文章记录我在真实 agent-workflow 中的封装方式、可运行代码和踩过的坑。

为什么固定规则不够：数据库变更审核的真实难点 ​

很多团队一开始会引入 SQL 审核平台，配置一批正则规则，比如禁止 SELECT *、强制索引名、限制 VARCHAR 长度。这类规则能挡住低级问题，但挡不住真正导致事故的变更：

  • 一张 2 亿行的表上直接 ALTER TABLE ... ADD COLUMN，在 MySQL 5.7 上可能触发表重建；
  • DELETE FROM orders WHERE status = 3 没有走索引，单次锁扫描拖垮整库；
  • 迁移脚本重复执行时报“字段已存在”，CI/CD 重试后把发布流程卡死；
  • 新索引看似没问题，但列区分度极低，优化器最终选择全表扫描。

这些判断都需要上下文：表多大、执行计划是什么、数据库版本支持什么 DDL 算法、语句是否可重入。让 LLM 直接“看 SQL 给意见”并不可靠，因为它看不到真实元数据；把查询能力通过 MCP 暴露给智能体，才能让模型基于事实推理，而不是凭经验编答案。

数据库变更审核智能体工作流
1
提交迁移脚本
开发者在 PR 中提交 SQL
2
MCP 技能编排
agent-workflow 串联多个工具
3
EXPLAIN 与元数据
评估执行计划与表规模
4
Online DDL 规则
判断锁表与幂等性
5
CI 门禁裁决
通过或阻断合并并给出修改建议

方案对比：人工审核、规则平台与 MCP 智能体 ​

我评估过三条路线，最后选择“MCP 技能 + 智能体编排”，但没有完全抛弃规则平台。固定规则负责确定性检查，MCP 技能负责需要上下文和推理的部分，DBA 只处理智能体标记为高风险或置信度不足的变更。

三种审核方案选型对比
1纯人工 DBA 审核
  • 经验判断准
  • 能处理极端场景
  • 反馈慢
  • 规则难沉淀
  • 夜间发版无人兜底
适合作为兜底，不适合高频迭代
2SQL 审核平台 + 固定规则
  • 稳定可重复
  • 接入简单
  • 缺上下文
  • 误报多
  • 无法结合执行计划推理
规则明确场景可用，泛化弱
推荐
3MCP 技能 + 智能体
  • 工具可复用
  • 可结合元数据推理
  • 能解释原因
  • 需要治理工具权限
  • 提示词需迭代
推荐，兼顾自动化与可解释性

这里的关键取舍是：不要让 LLM 直接生成或执行变更 SQL，只让它调用只读工具、解释结果、给出裁决建议。真正能否决 CI 的是可审计的规则函数和结构化风险等级。这样即便模型偶尔判断激进，也不会绕过安全边界。

MCP 技能设计：把 DBA 判断拆成四个原子能力 ​

我没有做一个巨大的“review_sql”工具，而是拆成四个高内聚技能。工具粒度太粗会让提示词难以约束，返回内容也会混杂；粒度太细则会导致智能体来回调用，CI 时间变长。

MCP 审核技能分层架构
1
接入层
GitHub/GitLab CIPR 评论审核报告
2
智能体层
变更解析风险推理修复建议门禁裁决
3
MCP 技能层
explain.analyzemeta.describe_tableddl.check_onlinemigration.check_idempotent
4
数据与规则层
只读从库information_schemaOnline DDL 规则库团队 SQL 规范

四个技能分别是：

  1. explain.analyze：对 DML 语句执行 EXPLAIN/EXPLAIN ANALYZE，返回访问类型、扫描行数、索引使用情况；
  2. meta.describe_table：从 information_schema 查询表结构、估算行数、索引基数；
  3. ddl.check_online：根据数据库版本和 DDL 类型判断是否可能锁表、是否支持 ALGORITHM=INPLACE/INSTANT；
  4. migration.check_idempotent：检查迁移是否可重复执行，例如是否使用 IF NOT EXISTS、是否有不可逆删除。

下面是一个可运行的 Python 版 MCP 工具定义示例。为了便于本地试验，我用 FastMCP 风格编写，数据库访问层只保留核心逻辑；生产环境应连接只读从库，并设置语句超时。

python
# mcp_db_review.py
# 安装依赖：pip install mcp pymysql
from dataclasses import dataclass
import re
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("db-change-review")

@dataclass
class ReviewContext:
    db_version: str = "8.0"
    lock_timeout_ms: int = 1500

def fake_explain(sql: str) -> dict:
    """
    演示用执行计划；生产环境替换为只读从库 EXPLAIN 查询。
    严禁在主库执行来自 PR 的未审核写 SQL。
    """
    return {
        "type": "ALL" if "where" not in sql.lower() else "ref",
        "rows": 2_000_000,
        "key": None if "where" not in sql.lower() else "idx_status",
        "filtered": 10.0 if "where" in sql.lower() else 100.0,
    }

@mcp.tool()
def explain_analyze(sql: str) -> dict:
    """分析 SQL 执行计划，识别全表扫描和预计扫描行数。"""
    plan = fake_explain(sql)
    risks = []
    if plan["type"] == "ALL":
        risks.append("访问类型为 ALL，可能发生全表扫描")
    if plan["rows"] > 100_000 and plan["filtered"] < 5:
        risks.append("扫描行数大且过滤率低，容易产生长事务")
    if plan["key"] is None:
        risks.append("没有命中索引")
    return {"plan": plan, "risks": risks, "level": "HIGH" if risks else "PASS"}

@mcp.tool()
def check_online_ddl(sql: str, db_version: str = "8.0") -> dict:
    """
    检查 DDL 是否可能锁表。
    规则保守处理：无法证明安全时返回 HIGH，而不是默认放行。
    """
    statement = sql.strip().lower()
    if not statement.startswith("alter table"):
        return {"level": "PASS", "risks": []}

    risks = []
    if "algorithm=instant" not in statement:
        risks.append("未显式指定 ALGORITHM=INSTANT，需确认是否触发表重建")
    if "lock=none" not in statement:
        risks.append("未指定 LOCK=NONE，数据库可能选择元数据锁或更严格锁级别")
    if re.search(r"\b(add primary key|modify|change)\b", statement):
        risks.append("主键或列定义变更通常成本较高，需要评估表规模与主从延迟")
    if tuple(map(int, db_version.split(".")[:2])) < (8, 0):
        risks.append("MySQL 5.7 对 INSTANT 支持有限，禁止仅凭经验放行")
    return {"level": "HIGH" if risks else "PASS", "risks": risks}

@mcp.tool()
def check_idempotent(sql: str) -> dict:
    """检查迁移脚本是否可重复执行，避免 CI/CD 重试时失败。"""
    s = sql.lower()
    risks = []

    if "create table" in s and "if not exists" not in s:
        risks.append("CREATE TABLE 缺少 IF NOT EXISTS，重复执行会失败")
    if "add column" in s and "if not exists" not in s:
        risks.append("ADD COLUMN 缺少 IF NOT EXISTS，建议使用框架的列存在性检查")
    if re.search(r"\bdrop column\b|\bdrop table\b", s):
        risks.append("DROP 类变更不可逆，必须拆成备份、灰度、清理三个迁移")
    if "delete from" in s and "limit" not in s:
        risks.append("大批量 DELETE 未分批，建议按主键小批量提交")

    return {"level": "BLOCK" if risks else "PASS", "risks": risks}

if __name__ == "__main__":
    mcp.run(transport="stdio")

智能体工作流：让模型基于证据给结论，而不是拍脑袋 ​

CI 侧我用一个轻量 agent-workflow 串起这些 MCP 技能。智能体拿到迁移文件后，先做语句分类，再决定调用哪些工具；最后将工具返回的结构化结果汇总成 Markdown 评论。提示词中我明确要求“只依据工具结果裁决”，并把 BLOCK、HIGH、WARN、PASS 的含义写死。

python
# ci_review_agent.py
import json
from mcp_client import MCPClient  # 项目内封装的 stdio MCP 客户端

SYSTEM_PROMPT = """
你是数据库变更审核智能体。你只能依据 MCP 工具返回的事实给出结论。
裁决规则：
- BLOCK：存在非幂等迁移、不可逆 DROP 或明确锁表风险；
- HIGH：执行计划全表扫描、扫描行数超过 10 万且无索引；
- WARN：风险不明确，需要 DBA 复核；
- PASS：未发现风险。
输出必须是 JSON：{"level":"...","evidence":[...],"suggestion":"..."}。
不要臆造表行数、索引名或数据库版本。
"""

async def review_migration(sql: str) -> dict:
    client = MCPClient(["python", "mcp_db_review.py"])
    await client.start()

    results = {}
    sql_type = classify_sql(sql)

    if sql_type in {"select", "update", "delete"}:
        results["explain"] = await client.call("explain_analyze", {"sql": sql})
    if sql_type == "ddl":
        results["online_ddl"] = await client.call("check_online_ddl", {"sql": sql})
    results["idempotent"] = await client.call("check_idempotent", {"sql": sql})

    # 确定性门禁优先，避免模型把 BLOCK 淡化成建议
    if any(v.get("level") == "BLOCK" for v in results.values()):
        return {"level": "BLOCK", "results": results}

    llm_decision = await llm_json_chat(
        system=SYSTEM_PROMPT,
        user=f"SQL：{sql}\n工具结果：{json.dumps(results, ensure_ascii=False)}"
    )
    return {"level": llm_decision["level"], "results": results, "llm": llm_decision}

def classify_sql(sql: str) -> str:
    head = sql.strip().lower()
    if head.startswith(("alter", "create", "drop")):
        return "ddl"
    if head.startswith(("update", "delete")):
        return head.split()[0]
    return "select"

async def llm_json_chat(system: str, user: str) -> dict:
    """替换为团队实际使用的 LLM SDK；这里保留接口形状。"""
    raise NotImplementedError

CI 中只需要根据最终 level 决定退出码：BLOCK 直接失败，HIGH 要求 DBA 审批标签，WARN 评论但不阻断，PASS 自动通过。我还会把每次工具调用、模型结论和 PR 编号落库，方便后续复盘误报与漏报。

避坑总结：这些问题我都真实踩过 ​

第一，不要让智能体直接连主库执行审核 SQL。 我早期为了拿到最新执行计划，把 DSN 配成了主库，虽然只调用 EXPLAIN，但某些数据库版本下 EXPLAIN UPDATE/DELETE 仍可能获取锁。修正方式是强制连接只读从库，增加 MAX_EXECUTION_TIME，并在 MCP 服务端拒绝多语句和非 SELECT/EXPLAIN 请求。

第二，幂等性不能只靠 IF NOT EXISTS。 MySQL 对 ALTER TABLE ... ADD COLUMN IF NOT EXISTS 的支持受版本和方言限制，有些语句语法能过但行为不一致。更稳的做法是由迁移框架先查 information_schema.columns，再决定是否执行；或者把变更拆成“存在性检查 + DDL”两个步骤。

第三，Online DDL 规则必须按数据库版本和表规模判断。 同样是加列，MySQL 8.0 的 INSTANT 和 5.7 的 INPLACE 成本完全不同。这个坑我排查了整整一下午：智能体在测试库给出 PASS，生产环境却出现主从延迟。后来我把版本号、估算行数、是否大事务写进工具入参，规则无法确认时一律返回 HIGH。

第四，模型输出只能解释，不能作为唯一门禁。 LLM 会把“建议补充索引”说得很像“必须阻断”，也可能漏掉工具已经明确返回的风险。我的处理是确定性规则先裁决，模型只负责汇总证据和生成修改建议；CI 退出码由代码计算，不由自然语言解析。

第五，MCP 工具返回要精简、结构化。 一次把整张 SHOW CREATE TABLE、完整执行计划和统计信息塞给模型，既浪费 token，也容易让模型抓住无关字段发挥。工具应返回风险判断所需字段，并附上原始证据链接，PR 评论里再展示完整报告。

把 DBA 的隐性经验沉淀成 Mcp 技能后，团队得到的不只是一个自动审核机器人，而是一套可复用、可审计、可持续迭代的 ai-skills 资产。后续我还计划把索引推荐和容量预测也纳入同一个 agent-workflow，但边界依旧不变：智能体负责分析和解释，确定性规则负责最终放行。本文首发于开发者日记，希望这套封装能帮你在下一次线上变更前，提前拦住那条最危险的 SQL。 ===CONTENT===

免责声明

本工具与文章内容仅用于学习测试场景，优化建议仅供参考，请勿直接用于生产环境，由此产生的风险自行承担。

本文来自开发者日记（devdiary.cn）· 记录开发者实战笔记与免费在线工具箱

免责声明

本工具与文章内容仅用于学习测试场景，优化建议仅供参考，请勿直接用于生产环境，由此产生的风险自行承担。