基于MCP协议的MySQL智能运维平台设计与实践

做数据库运维的朋友应该都有过这种体验半夜被一条“MySQL 主库线程数过高”的告警叫醒睡眼惺忪爬起来登录跳板机一条条执行 SHOW PROCESSLIST、查慢日志、跑 EXPLAIN最后定位到某个慢 SQL 再通知业务方优化。整个过程说不上难但就是重复、琐碎而且特别依赖“知道下一步该查什么”的经验积累。前段时间我在研究 MCPModel Context Protocol模型上下文协议脑子里突然冒出一个念头如果把 MySQL 的日常巡检和诊断动作全部抽象成标准工具让 AI 助手通过 MCP 协议直接调用是不是就能把这套排查链路自动化说干就干我花了两周时间基于开源 MCP 服务端搭了一个 MySQL 智能运维平台前端对接常见的交互式 AI 客户端用自然语言就能完成大部分日常巡检和初步诊断。这篇博客就把整个思路、架构、核心代码和踩过的坑完整记录下来希望对同样折腾 MCP 运维自动化的朋友有帮助。1. 为什么是 MCP MySQL 这个组合1.1 传统 MySQL 运维的流水线困局先说个普遍痛点。生产环境 MySQL 出故障大多数时候不是“不会处理”而是“排查链路太长”。业务方反馈慢你要先看连接数有没有打满再看有没有锁等待然后翻慢日志定位具体 SQL跑 EXPLAIN分析索引情况最后才能判断问题出在哪。我粗略统计过一次典型故障里真正动手修复的时间可能只有几分钟但前面的检查动作往往要花上半小时甚至更久。关键在于这些检查动作高度重复连接数、锁状态、慢日志、索引、表结构翻来覆去就是那十几条标准 SQL。传统做法是写一批固定脚本定时采样、出问题时人工翻报表但脚本能解决“数据有没有”的问题解决不了“该查什么、怎么解读”的问题。因为故障场景千变万化固定脚本永远覆盖不全判断逻辑还得靠人肉完成。1.2 MCP 补上了“AI 摸到数据库”的最后一公里MCP 是 Anthropic 提出并开源的一套标准化协议核心思路很简单把 AI 模型和外部工具之间的调用关系标准化。服务端以标准接口暴露工具、资源和提示词客户端接入后AI 模型就能自主决定何时调用、以什么参数调用这些能力。这类的价值我用一个类比说过很多次就像打印机驱动。早年每台打印机要装专属驱动换了品牌就抓瞎后来有了标准打印协议插上就能用。MCP 之于 AI 应用做的就是这件事。AI 模型不需要了解 MySQL 协议细节也不需要关心服务端是 Python 还是 Go 写的它只需要知道“这边有几个工具分别能干什么”就能自主编排诊断流程。具体到 MySQL 运维这个组合的价值就体现出来了AI 模型天然擅长“根据现象判断下一步查什么”这类推理而 MySQL 诊断同样高度依赖一套标准查询命令。两者通过 MCP 对接之后AI 不再是纸上谈兵的聊天机器人而是真正能读取数据库状态、执行只读分析的操作者。1.3 适用场景与边界划分虽然这个组合很诱人但我不建议把所有运维工作都交给 AI。我在设计工具集时把场景分成了三类适合交给 AI 的日常巡检、慢查询分析、连接数查看、锁状态诊断、索引使用情况分析、EXPLAIN 结果解读、表结构查看等只读类操作。需要谨慎对待的需要写操作的变更类动作比如建索引、改配置、清理数据。这类操作建议只暴露“评估”能力不暴露“执行”能力。坚决不碰的DROP TABLE、TRUNCATE、大量 UPDATE/DELETE、账号权限修改等高危动作在服务端层面直接做白名单拦截。这个边界会在服务端代码里强制约束而不是指望 AI 自觉。后面讲实现的时候你会看到我是如何在工具层就杜绝写操作的。2. 平台整体架构与核心设计2.1 MCP 服务端到底长什么样、怎么工作先看 MCP 协议层的基本模型它包含四个核心概念Tools工具AI 可调用的具体函数例如“执行只读 SQL”“查看当前连接列表”。模型根据系统提示和用户问题决定是否调用。Resources资源可暴露给模型的静态或动态数据例如服务器配置信息、监控报表。资源通常用于给模型补充上下文。Prompts提示词预置的指令模板帮助模型按特定套路执行任务例如“慢查询分析模板”。Transport传输层客户端与服务端之间的通信方式最常见的是标准输入输出stdio和 SSE/HTTP。我的平台主要依赖 Tools 层因为诊断动作本质上是“模型发起调用、工具执行并返回结果”的循环。调用流程可以简单理解成用户用自然语言提问 → AI 模型判断需要调用某个工具 → 通过 MCP 客户端发起工具调用请求 → 服务端连接 MySQL 执行查询 → 返回格式化结果 → 模型基于结果继续推理和回答。整个过程中模型是决策者服务端是执行者。2.2 数据流与分层设计我在设计时把它分成三层职责清晰接入层MCP 服务端入口负责处理协议握手、工具注册、请求分发。这一层不写任何业务逻辑。工具层定义每一个运维动作比如“查看慢日志”“分析索引”。工具函数做参数校验、调用数据层、格式化返回结果。数据层负责 MySQL 连接管理、SQL 安全过滤、查询超时控制。数据层是安全底线所在。这种分层的好处是将来想加一个新工具只需要在工具层写一个函数并注册不需要改动其他逻辑。我把所有工具做到同一个服务进程里方便部署和管理如果工具量太大也可以拆成多个 MCP 服务按需挂载。2.3 技术选型的理由我选择了 Python 官方 MCP SDK PyMySQL 这套组合理由有三条生态成熟MCP 官方 Python SDK 维护最活跃FastMCP 装饰器模式写起来极快适合快速验证。数据库连接方便PyMySQL 是纯 Python 实现安装简单不需要编译原生依赖在大多数 Linux 服务器上都能直接跑。和 AI 工具链天然契合Python 生态里的类型注解、数据格式化、日志工具都能直接复用后面想扩展解析能力也容易。如果你想用 Node.js 或 GoMCP 社区也有对应 SDK但我在对比后认为 Python 版本对“工具注册 参数校验”的支持最完善尤其是 FastMCP 的 Pydantic 参数校验能帮我们省掉大量防御性代码。3. 从零落地开源服务端核心环节拆解3.1 项目初始化与依赖清单我建议先建一个干净的虚拟环境避免污染系统 Python。依赖非常简单mkdir mysql-mcp-ops cd mysql-mcp-ops python3 -m venv .venv source .venv/bin/activate pip install mcp[cli] pymysql cryptography python-dotenv这里解释一下每个依赖的用途mcp[cli]MCP 官方 Python SDK附带了调试命令行工具 mcp方便本地跑 Dev 模式验证工具是否正常注册。pymysql纯 Python 的 MySQL 驱动不需要编译。cryptographyPyMySQL 连接 MySQL 8 时做认证插件caching_sha2_password必需。python-dotenv从 .env 文件读取数据库连接配置避免把密码写死在代码里。项目结构保持扁平mysql-mcp-ops/ ├── .env # 数据库连接等敏感配置 ├── server.py # MCP 服务端入口 ├── db.py # 数据库连接与安全过滤 ├── tools/ # 工具集目录 │ ├── __init__.py # 注册工具到服务端 │ ├── inspect.py # 连接信息、状态变量 │ ├── slow_query.py # 慢查询分析 │ ├── schema.py # 表结构、索引查看 │ └── processlist.py # 当前连接与锁信息 └── requirements.txt3.2 数据库连接管理与安全设计这是整个平台最核心、也最不能含糊的部分。我的思路是服务端数据库连接使用专用只读账号该账号在 MySQL 侧只授予 SELECT、SHOW VIEW、PROCESS 等只读权限从数据库层面杜绝写操作。# db.py import os import pymysql from pymysql.cursors import DictCursor from dotenv import load_dotenv load_dotenv() DB_CONFIG { host: os.getenv(MYSQL_HOST, 127.0.0.1), port: int(os.getenv(MYSQL_PORT, 3306)), user: os.getenv(MYSQL_USER, ops_reader), password: os.getenv(MYSQL_PASSWORD, ), charset: utf8mb4, cursorclass: DictCursor, connect_timeout: 5, read_timeout: 10, write_timeout: 5, } def get_connection(): return pymysql.connect(**DB_CONFIG)这里有几个细节值得注意read_timeout设置为 10 秒避免慢 SQL 拖垮整个工具服务。万一某个查询超过 10 秒没返回连接会主动断开并报错。charset必须用utf8mb4因为很多业务表里会有 emoji 或生僻字用旧版 utf8 会导致转码异常。每次调用工具都新建连接、用后即关。虽然连接池更高效但运维工具调用频率不高简单连接反而避免连接泄漏问题。然后在数据库侧创建只读账号这一步建议由 DBA 执行CREATE USER ops_reader% IDENTIFIED BY YourStrongPassword; GRANT SELECT, SHOW VIEW, PROCESS ON *.* TO ops_reader%; GRANT SELECT ON performance_schema.* TO ops_reader%; FLUSH PRIVILEGES;注意PROCESS权限是查看 SHOW PROCESSLIST 所必需的但如果你的安全策略较严也可以不授只靠information_schema.PROCESSLIST做部分查看。performance_schema用于读取部分诊断数据按需授权。3.3 核心工具函数的实现工具层是我花时间最多的地方。我最终保留了 6 个工具覆盖了日常巡检和故障排查的大部分场景。逐个看核心实现。第一个工具执行只读查询。这是最基础但也是最危险的工具所以安全校验必须做足# tools/inspect.py import re import json from mcp.server.fastmcp import FastMCP from db import get_connection # SQL 白名单校验只允许以这些关键字开头的语句 ALLOWED_PREFIX ( select, show, explain, describe, desc, with, show full, show create, ) BLOCK_PATTERNS [ r\b(insert|update|delete|drop|truncate|alter|create|grant|revoke)\b, r\b(mysql\.user|mysql\.db)\b, ] def _check_read_only_sql(sql: str): sql_stripped sql.strip().lower().rstrip(;) if not sql_stripped.startswith(ALLOWED_PREFIX): raise ValueError(f仅支持 SELECT/SHOW/EXPLAIN/DESCRIBE 语句) for pattern in BLOCK_PATTERNS: if re.search(pattern, sql_stripped, re.IGNORECASE): raise ValueError(检测到禁止的关键字或系统表访问) return sql_stripped def register_inspect_tools(mcp: FastMCP): mcp.tool() def exec_query(sql: str, limit: int 50) - str: 在 MySQL 上执行只读查询并返回结果。limit 控制最大返回行数默认 50最大 200。 sql _check_read_only_sql(sql) if limit 200: limit 200 try: conn get_connection() with conn.cursor() as cur: cur.execute(sql) # 如果用户没写 LIMIT自动追加 if limit not in sql.lower(): sql_with_limit sql.rstrip(;) f LIMIT {limit} cur.execute(sql_with_limit) rows cur.fetchmany(sizelimit) # 转成 JSON 方便模型阅读 result json.dumps(rows, defaultstr, ensure_asciiFalse) return f查询成功返回 {len(rows)} 行\n{result} finally: if conn in locals() and conn: conn.close()这段代码有三个关键防御点关键字黑名单拦截、LLM 可能会“忘写” LIMIT 自动补、结果序列化成 JSON 让模型更容易解读。这里有个很实际的问题模型生成的 SQL 不一定老实有时会生成select * from 超大表这种危险查询所以我强制补 LIMIT同时上限 200 行双保险。第二个工具查看当前连接和锁状态。这是故障排查高频动作# tools/processlist.py from mcp.server.fastmcp import FastMCP from db import get_connection import json def register_processlist_tools(mcp: FastMCP): mcp.tool() def show_processlist() - str: 查看当前 MySQL 所有活跃连接包括用户、来源 IP、状态、正在执行的 SQL。用于排查连接数过高和锁问题。 conn get_connection() try: with conn.cursor() as cur: cur.execute(SHOW FULL PROCESSLIST) rows cur.fetchall() return f当前活跃连接数{len(rows)}\n json.dumps(rows, defaultstr, ensure_asciiFalse) finally: conn.close() mcp.tool() def show_innodb_status() - str: 查看 InnoDB 引擎状态主要用于分析死锁、锁等待和事务信息。 conn get_connection() try: with conn.cursor() as cur: cur.execute(SHOW ENGINE INNODB STATUS) row cur.fetchone() # 原始输出很长截取 Status 字段 status_text row.get(Status, ) if row else return status_text[:3000] finally: conn.close()SHOW FULL PROCESSLIST有个坑某条连接正在执行长 SQL 时Info字段可能是 NULL因为 MySQL 返回的是正在执行的语句语句太长还会被截断。所以我在返回时做了defaultstr避免 NULL 导致 JSON 序列化失败。第三个工具慢查询分析。这个最复杂因为不同 MySQL 版本的慢日志存储方式不一样# tools/slow_query.py from mcp.server.fastmcp import FastMCP from db import get_connection import json def register_slow_query_tools(mcp: FastMCP): mcp.tool() def show_slow_log(limit: int 20) - str: 查看最近产生的慢查询记录返回 SQL 文本、耗时、锁等待时间、扫描行数等关键指标。 conn get_connection() try: with conn.cursor() as cur: # MySQL 8.0 且开启慢查询日志到表时使用 mysql.slow_log try: cur.execute( SELECT start_time, query_time, lock_time, rows_examined, rows_sent, sql_text FROM mysql.slow_log ORDER BY start_time DESC LIMIT %s, (limit,) ) rows cur.fetchall() return f最近 {len(rows)} 条慢查询\n json.dumps(rows, defaultstr, ensure_asciiFalse) except Exception: # 如果没授权访问 slow_log 表退化为 information_schema 信息 cur.execute( SELECT VARIABLE_VALUE FROM information_schema.GLOBAL_STATUS WHERE VARIABLE_NAME LIKE Slow_queries ) row cur.fetchone() return f累计慢查询数{row.get(VARIABLE_VALUE, N/A)}。提示如需查看具体慢 SQL请开启慢日志或授予对应权限。 finally: conn.close()这里我做了降级处理不同环境权限不一样服务端不应该因为访问失败就罢工。我在代码里 catch 住异常后退化成返回慢查询总量的全局状态至少给 AI 提供部分信息让它知道系统里可能有多少慢查询需要关注。第四个工具查看表结构和索引。这是我实际使用频率最高的工具因为 AI 分析 SQL 性能时经常需要确认某个字段是否有索引# tools/schema.py from mcp.server.fastmcp import FastMCP from db import get_connection import json def register_schema_tools(mcp: FastMCP): mcp.tool() def show_table_schema(table_name: str, database_name: str None) - str: 查看指定表的建表语句包含所有字段定义和索引信息。用于分析 SQL 是否走了正确索引。 conn get_connection() try: with conn.cursor() as cur: schema_sql SHOW CREATE TABLE if database_name: schema_sql f {database_name}.{table_name} else: schema_sql f {table_name} cur.execute(schema_sql) row cur.fetchone() if not row: return f未找到表 {table_name}请检查表名或库名是否正确。 # 结果是一个 dict第二个字段是 Create Table create_stmt list(row.values())[1] if len(row) 1 else str(row) return create_stmt finally: conn.close()这个工具的妙处在于模板字符串里对库名和表名都加了反引号能有效避免 SQL 注入。虽然我们只读但表名拼进 SQL 时如果不加过滤模型可能生成奇怪的输入反引号方案成本最低、效果最好。3.4 注册所有工具并启动服务端最后在入口文件里把所有工具注册好然后用 FastMCP 启动# server.py from mcp.server.fastmcp import FastMCP from tools.inspect import register_inspect_tools from tools.processlist import register_processlist_tools from tools.slow_query import register_slow_query_tools from tools.schema import register_schema_tools import os mcp FastMCP( mysql-ops-server, instructions你是 MySQL 智能运维助手。根据用户问题调用工具进行诊断所有操作必须走只读工具严禁尝试写操作。, ) # 注册工具集 register_inspect_tools(mcp) register_processlist_tools(mcp) register_slow_query_tools(mcp) register_schema_tools(mcp) if __name__ __main__: # 默认走 stdio 传输适合对接本地客户端 mcp.run(transportstdio)启动之后可以用 SDK 自带的调试命令验证工具是否注册成功mcp dev server.py这个命令会启动一个开发调试界面你能看到所有注册的工具列表、参数定义还能手动模拟调用。我强烈建议在接 AI 客户端之前先用这个工具把所有函数手动调一遍确认返回格式符合预期不然很有可能客户端那边报半天错结果是服务端工具本身出了问题。4. 对接交互式 AI 助手让运维对话化4.1 客户端选择与配置流程服务端就绪后接下来就是接入交互式 AI 客户端让用户通过自然语言操作。我测试了两种主流客户端桌面版 Claude 和命令行 Codex以及一种自己用 MCP SDK 写的极简客户端。原理都一样只需要在客户端配置里加上我们服务端的启动命令。以 Claude Desktop 为例配置文件在claude_desktop_config.json里加一段{ mcpServers: { mysql-ops: { command: python, args: [/绝对路径/mysql-mcp-ops/server.py], env: { MYSQL_HOST: 127.0.0.1, MYSQL_PORT: 3306, MYSQL_USER: ops_reader, MYSQL_PASSWORD: YourStrongPassword } } } }如果你的服务端跑在远程机器上就不能用 stdio 了需要改成 SSE 模式# server.py 中改成 mcp.run(transportsse, host0.0.0.0, port8000)客户端配置随之改为{ mcpServers: { mysql-ops: { url: http://你的服务器:8000/sse } } }但我要提醒一下远程模式意味着 AI 客户端可以跨网络连接你的数据库运维服务端安全风险显著提升必须在服务端加鉴权层比如 API Key否则不建议直接暴露到公网。4.2 实战演示用自然语言完成一次慢查询分析配置好之后我做的第一件事就是模拟一次真实故障排查。我在测试库造了几十万条数据故意在某个表里删掉一个关键索引然后执行一条很慢的查询把这个查询塞进慢日志。接着用 AI 客户端问“帮我看看最近有没有慢查询分析一下原因并给出优化建议。”AI 会这样工作它先调用show_slow_log工具拿到最近慢查询列表发现有一条SELECT * FROM orders WHERE user_id 123 AND amount 100耗时 4.2 秒然后它调用show_table_schema查看orders表结构确认user_id字段没有索引再调用exec_query跑一条EXPLAIN SELECT ...看到全表扫描的标记最后综合这些信息给出优化建议为user_id和amount建立联合索引。整个过程我一行 SQL 都没手写。AI 自己决定调用什么工具、传什么参数、如何解读结果。这就是 MCP 的威力所在模型是决策者服务端是执行者两者之间通过标准协议高效协作。4.3 权限与审计运维的最后底线接入 AI 助手后权限控制反而更重要了。我在整个项目里贯彻了“三层防守”的思路数据库层使用只读账号从根上杜绝写操作。这一步最有效因为就算服务端代码出了漏洞数据库权限也能兜底。应用层SQL 关键字白名单 黑名单校验防止模型生成危险语句。前面代码里的_check_read_only_sql就是干这个的。使用层AI 客户端的对话记录天然可审计建议开启会话日志。一旦出了事故可以回溯到是哪次调用、什么参数导致了问题。另外我还给服务端加了一个简单的请求日志功能每次工具调用都会记录时间、工具名、参数、返回行数# 在工具函数里统一加装饰器 import logging, time, functools def audit_log(func): functools.wraps(func) def wrapper(*args, **kwargs): start time.time() result func(*args, **kwargs) logging.info( f[AUDIT] tool{func.__name__} params{kwargs} elapse{time.time()-start:.2f}s result_len{len(str(result))} ) return result return wrapper建议生产环境把审计日志接入统一日志平台方便事后排查和告警联动。5. 实操踩坑记常见问题与排查技巧5.1 MCP 服务端启动失败或客户端连不上这是新手最容易遇到的第一道坎。我遇到的情况主要有三种Python 解释器路径不对客户端配置里command写的是python但你的虚拟环境 Python 不在 PATH 里。解决方案是在配置里写绝对路径比如/root/mysql-mcp-ops/.venv/bin/python。依赖没装进同一个环境如果你在虚拟环境里装了依赖但客户端用的是系统 Python 启动服务端会直接报 ModuleNotFoundError。建议启动前先用which python确认路径。SSE 端口被防火墙挡住远程模式下客户端连不上第一反应curl http://你的服务器:8000/sse看看通不通不通就先查防火墙和安全组。排查技巧不要瞎猜先在命令行手动跑一次python server.py看有没有报错这能排除 80% 的环境问题。5.2 模型生成的 SQL 不理想或者结果解析失败模型虽然聪明但生成的 SQL 偶尔还是让我哭笑不得。比如它会在SHOW CREATE TABLE语句里多加一个分号或者在EXPLAIN后面跟一整条 INSERT 语句。我发现最有效的防御不是反复 prompt 纠正而是在代码里做模糊容忍# 统一清理掉末尾分号、多余空白 sql sql.strip().rstrip(;).strip() # 如果同时出现多个分号只保留第一段 if ; in sql: sql sql.split(;)[0].strip()另外MySQL 返回的Decimal、datetime类型不能被标准 JSON 序列化必须加defaultstr。我在第一版代码里忽略了这个问题导致 AI 客户端报 “JSON serialization error”排查了半天才发现是日期字段没处理。5.3 返回数据量过大导致会话上下文爆炸AI 客户端的上下文窗口是有限的。如果你执行一条SELECT * FROM 大表虽然服务端只返回 200 行但这 200 行可能包含非常长的文本字段、二进制字段直接把上下文塞爆。我的解决方案是在工具层做两层截断服务端限制行数上限不超过 200。单次返回内容超过 3000 字符时自动做截断并加提示“结果过长已截断前 N 字符。”这样能保证 AI 客户端始终有足够上下文处理最核心的信息而不是被一堆无关字段淹没。5.4 安全方面的踩坑权限被低估了我在测试时曾偷懒用 root 账号连接服务端结果 AI 在一次推理中“自作主张”生成了一条ALTER TABLE语句。虽然我们的 SQL 白名单拦截了它但这件事给我提了个醒永远不要依赖 AI 的自觉永远在工具层强制校验永远使用最小权限账号。这条金标准现在写在项目 README 的最前面。另一个容易被忽略的点是PROCESS权限虽然只是查看进程列表但它能看到其他用户正在执行的 SQL 文本如果生产库有敏感查询这个权限可能涉及合规问题。建议根据实际安全策略评估是否授予。6. 后续扩展思路与个人体会做到这一步平台已经能覆盖日常巡检和初步诊断了。但我觉得它还有很大的扩展空间列几个我认为值得继续做的方向扩展数据源MCP 服务端不止可以接 MySQL还可以接 Redis查看 key 分布、慢日志、PG长事务、死锁、Elasticsearch慢查询、集群状态做成一个统一的“数据库智能助手”。对接告警平台把 MCP 服务端挂在告警回调里让 AI 在收到告警时自动做初步诊断并输出分析报告和安全建议。增加运维知识库把内部的 SQL 规范、索引规范、故障复盘文档通过 MCP Resources 暴露给模型让回答更贴合团队实践。Webhook 联动AI 分析完后可以再通过另一个工具对接工单系统自动创建优化任务形成闭环。最后说一点个人体会。我最初做这个项目只是想省掉半夜爬起来手敲 SQL 的麻烦。做完之后最大的感受是MCP 给运维自动化的想象力不在哪一条具体的 SQL而是它把“AI 编排工具”这件事真正标准化了。以前我们写自动化脚本是在对抗不确定的工具接口现在只需要暴露能力剩下的判断和编排可以交给 AI。当然前提是安全边界做扎实。如果你也在研究 MCP 和数据库运维的结合建议先从只读巡检这类低风险场景开始跑通全链路踩过一轮坑之后再考虑扩大工具集。这套链路跑通之后你会明显感觉到深夜看告警这件事终于不再那么孤独了。