1. 为什么 DBA 巡检需要 MCP 协议来接管数据库巡检这件事做过 DBA 的都懂每天早上到工位先连跳板机敲一遍SHOW PROCESSLIST、SHOW ENGINE INNODB STATUS、翻慢查询日志、看磁盘水位然后整理成巡检邮件发出去。这套动作重复三年五年闭着眼都能敲但问题在于——它占的是你最清醒的那段时间而真正需要你判断力的架构优化、容量规划反而被挤到下午。MCP 协议Model Context Protocol在 2026 年成为 DBA 圈子的新宠核心原因不是它多玄乎而是它把「AI 能调用工具」这件事标准化了。以前你想让大模型帮你查数据库得自己写 Function Calling 的 JSON Schema、处理参数解析、拼返回值每个模型厂商的格式还不一样。MCP 把这层抽象掉了你写一个 MCP Server暴露几个工具接口任何支持 MCP 的客户端Claude Desktop、Cline、Cursor 等都能直接调用。放到巡检场景里这意味着什么意味着你可以把「查当前连接数」「查表大小排名」「查慢查询」这些 DBA 高频动作封装成三个工具然后对着 AI 说一句「库今天状态怎么样」它自己去连 MySQL、执行巡检 SQL、把异常会话和慢查询列表整理成人话甩给你。你不需要写定时任务发邮件也不需要业务方每次来找你手工查——AI 成了那个 7x24 小时在线的巡检值班员。这篇文章要交付的东西很具体一个可复制的 MCP Server 配置片段、一份巡检任务清单、一次本地验证动作。你跟着做能在自有环境跑通「AI 自动巡检数据库」的闭环。TaoToken 在这里的角色是提供统一的 Key 和 API 通道让你在配置 MCP 客户端时不用为每个模型单独折腾鉴权。官网入口在 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 后面配置环节会用到。先说清楚适合谁如果你手头有 MySQL 5.7 或 8.0不管是本地 Docker 还是测试环境只要你能连上去就能跟着跑。Python 3.10 是硬要求因为 MCP 的 Python SDK 依赖较新的 asyncio 特性。Claude Desktop 作为客户端最省事但如果你用 Cline 或别的 MCP 客户端配置逻辑一样只是配置文件路径不同。巡检任务清单我建议先定这三条别贪多第一当前负载与锁等待。查Threads_connected、Threads_running再看SHOW PROCESSLIST里非 Sleep 状态的线程重点盯执行时间超过 60 秒的。这条对应「数据库是不是堵了」。第二慢查询 Top N。拉最近 24 小时执行时间超过 1 秒的 SQL按耗时降序。这条对应「哪个 SQL 在拖后腿」。第三表大小排名。从information_schema.tables统计各库各表的data_length index_length按总大小降序取前 15。这条对应「磁盘会不会被打满」。这三条覆盖了 DBA 日常巡检 80% 的诉求而且都是只读操作不会误删误改。MCP Server 里只暴露这三个工具AI 就算想执行DROP TABLE也没入口——这是权限隔离的关键设计后面会展开。2. TaoToken 前置统一 Key 与 MCP 客户端的接入准备在写 MCP Server 之前得先把「AI 那一端」的通道准备好。MCP 协议本身只负责工具调用的标准化但 AI 模型从哪来、用什么 Key 调、走哪个 API 端点这些是客户端配置的事。TaoToken 在这里提供的是统一 Key 和 API 通道让你在 Claude Desktop、Cline 这些客户端里填一次配置就能用。先明确一个概念MCP Server 是你本地跑的一个进程它通过 stdio标准输入输出和客户端通信。客户端比如 Claude Desktop负责把用户的自然语言转成工具调用请求发给 MCP ServerMCP Server 执行完数据库查询把结果返回给客户端客户端再让模型把结果整理成人话。所以整条链路是用户 → 客户端 → 模型 API → 客户端 → MCP Server → MySQL → 原路返回。TaoToken 作用在「客户端 → 模型 API」这一段。你需要在 TaoToken 控制台创建一个 API Key然后在客户端配置里把 Base URL 指向 TaoToken 的 API 端点把 Key 填进去。这样模型请求就走 TaoToken 的通道不用你自己去每个模型厂商单独开账号。具体操作路径访问 https://taotoken.net/api 拿到 API 端点注意这个地址不带 UTM 参数是纯 API 入口。然后去控制台创建 Key控制台地址是 https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 。创建完 Key 之后在 API Keys 页面可以管理地址是 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 。如果你用的是 Claude Code 或者需要 Anthropic 兼容格式的客户端TaoToken 有对应的接入文档地址是 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 。Claude Code 的专用接入页在 https://taotoken.net/ClaudeCodeAnthropic?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 里面会写清楚 Base URL 和 Model ID 怎么填。这里要强调「三件套」的概念Base URL、API Key、Model ID。不管你是配 Claude Desktop、Cline 还是 Codex 的 auth.json这三个东西缺一不可。Base URL 决定请求发到哪API Key 决定你有没有权限Model ID 决定用哪个模型。TaoToken 的统一通道让你在这三个参数上不用来回切换一个 Key 可以走多个模型。如果你打算长期跑编码类或 Agent 类任务比如让 AI 持续巡检、自动生成报告可以看看 Coding Plan地址是 https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 。这个适合需要稳定调用额度的场景比按次计费更划算。模型对话的入口在 https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 你可以先在网页上验证 Key 能不能正常调模型再去配 MCP 客户端。这一步别跳过因为后面排障时如果 MCP 工具调用失败你得先确认是模型通道的问题还是 MCP Server 的问题。准备工作清单Python 3.10 环境python3 --version确认一下。MySQL 5.7 或 8.0本地或远程都行但建议先用本地测试库别直接连生产。Claude Desktop 或其他 MCP 客户端Claude Desktop 在设置里能看到 Developer 选项就说明支持 MCP。TaoToken 的 API Key 和 Base URL提前准备好。还有一个容易忽略的点MCP 客户端和 MCP Server 之间的通信是本地 stdio不走网络。所以你的数据库连接信息host、user、password是写在 MCP Server 代码里的不会经过 TaoToken 通道。TaoToken 只负责模型 API 那一层。这个边界要清楚不然会担心数据库密码泄露——实际上它只存在你本地。3. 可复制配置MCP Server 代码与客户端 settings 片段这一节是全文的核心直接给可复制的代码和配置。我按「建项目 → 造演示数据 → 写 Server → 配客户端」的顺序来每一步都有完整命令和文件内容。先建项目目录并装依赖mkdir mcp-mysql-inspector cd mcp-mysql-inspector pip install mcp pymysql touch server.pymcp是官方 Python SDKpymysql是 MySQL 驱动。装完之后pip list | grep mcp确认一下版本建议用 1.0 以上的。接下来造演示数据。实际生产中我们直接查information_schema和performance_schema但为了让你跑起来立刻看到效果先建一张模拟慢查询表。创建init_db.py# init_db.py import pymysql from datetime import datetime, timedelta import random conn pymysql.connect( hostlocalhost, userroot, password你的密码, databasetest_db, charsetutf8mb4 ) cursor conn.cursor() cursor.execute( CREATE TABLE IF NOT EXISTS slow_query_log_demo ( id INT AUTO_INCREMENT PRIMARY KEY, sql_text TEXT, query_time DECIMAL(10,2), rows_examined INT, start_time DATETIME ) ) for i in range(20): sql_text fSELECT * FROM orders WHERE user_id {random.randint(1,1000)} query_time round(random.uniform(0.5, 8.0), 2) rows_examined random.randint(100, 50000) start_time datetime.now() - timedelta(minutesrandom.randint(1, 1440)) cursor.execute( INSERT INTO slow_query_log_demo (sql_text, query_time, rows_examined, start_time) VALUES (%s, %s, %s, %s), (sql_text, query_time, rows_examined, start_time) ) conn.commit() conn.close() print(模拟慢查询数据初始化完成)跑之前先建库CREATE DATABASE test_db;然后python3 init_db.py。看到「初始化完成」就对了。现在写核心的server.py。我把工具拆成三个专用接口而不是一个通用 SQL 执行器——这是关键设计后面排障会讲为什么。# server.py - MySQL 数据库智能巡检助手 import pymysql import json from mcp.server import Server from mcp.server.stdio import stdio_server from mcp.types import Tool, TextContent DB_CONFIG { host: localhost, user: root, password: 你的密码, database: test_db, charset: utf8mb4 } app Server(mysql-inspector) app.list_tools() async def list_tools() - list[Tool]: return [ Tool( nameget_connection_status, description获取当前数据库连接数和线程状态。用于检查数据库是否拥堵是否存在大量堆积。, inputSchema{type: object, properties: {}, required: []} ), Tool( nameget_table_size_ranking, description统计所有数据库或指定库的表大小排名包含数据量、索引大小。用于排查磁盘增长过快的表。, inputSchema{ type: object, properties: { database_name: { type: string, description: 指定要查询的数据库名如 test_db。不传则查所有库。 } }, required: [] } ), Tool( nameget_recent_slow_queries, description查询最近1天内执行时间超过1秒的慢查询SQL按执行时间降序排列。帮助快速定位性能瓶颈。, inputSchema{ type: object, properties: { limit: { type: integer, description: 返回的慢查询条数默认10条。 } }, required: [] } ) ] app.call_tool() async def call_tool(name: str, arguments: dict) - list[TextContent]: conn pymysql.connect(**DB_CONFIG) cursor conn.cursor() try: if name get_connection_status: cursor.execute(SHOW STATUS LIKE Threads_connected) threads_connected cursor.fetchone()[1] cursor.execute(SHOW PROCESSLIST) processes cursor.fetchall() active_threads [p for p in processes if p[4] ! Sleep] cursor.execute(SHOW VARIABLES LIKE max_connections) max_conn cursor.fetchone()[1] result { 当前连接数: int(threads_connected), 最大允许连接数: int(max_conn), 活跃执行线程数: len(active_threads), 活跃线程详情(截取前5条): [ {ID: p[0], 用户: p[1], 主机: p[2], 命令: p[4], 执行时间(秒): p[5], 状态: p[7]} for p in active_threads[:5] ] } return [TextContent(typetext, textjson.dumps(result, ensure_asciiFalse, indent2))] elif name get_table_size_ranking: db_name arguments.get(database_name) sql SELECT table_schema AS 数据库, table_name AS 表名, ROUND((data_length index_length) / 1024 / 1024, 2) AS 总大小(MB), ROUND(data_length / 1024 / 1024, 2) AS 数据大小(MB), ROUND(index_length / 1024 / 1024, 2) AS 索引大小(MB), table_rows AS 估算行数 FROM information_schema.tables WHERE table_schema NOT IN (information_schema, performance_schema, mysql, sys) params [] if db_name: sql AND table_schema %s params.append(db_name) sql ORDER BY (data_length index_length) DESC LIMIT 15 cursor.execute(sql, params) rows cursor.fetchall() columns [desc[0] for desc in cursor.description] data [dict(zip(columns, row)) for row in rows] return [TextContent(typetext, textjson.dumps(data, ensure_asciiFalse, indent2))] elif name get_recent_slow_queries: limit arguments.get(limit, 10) try: cursor.execute( SELECT DATE_FORMAT(start_time, %H:%i:%s) AS 开始时间, ROUND(query_time, 2) AS 耗时秒, rows_examined AS 扫描行数, sql_text AS SQL摘要 FROM performance_schema.events_statements_history_long WHERE query_time 1000000000 ORDER BY query_time DESC LIMIT %s , (limit,)) rows cursor.fetchall() if rows: columns [desc[0] for desc in cursor.description] return [TextContent(typetext, textjson.dumps([dict(zip(columns, row)) for row in rows], ensure_asciiFalse, indent2))] else: cursor.execute( SELECT DATE_FORMAT(start_time, %H:%i:%s) AS 开始时间, query_time AS 耗时秒, rows_examined AS 扫描行数, sql_text AS SQL摘要 FROM slow_query_log_demo ORDER BY query_time DESC LIMIT %s , (limit,)) rows cursor.fetchall() columns [desc[0] for desc in cursor.description] return [TextContent(typetext, textjson.dumps([dict(zip(columns, row)) for row in rows], ensure_asciiFalse, indent2))] except Exception as e: return [TextContent(typetext, textf慢查询查询失败请检查权限或数据库版本{str(e)})] else: return [TextContent(typetext, textf未知巡检工具{name})] except Exception as e: return [TextContent(typetext, textf巡检执行异常{str(e)})] finally: conn.close() async def main(): async with stdio_server() as (read_stream, write_stream): await app.run(read_stream, write_stream) if __name__ __main__: import asyncio asyncio.run(main())代码里有三个设计点值得说。第一去掉了通用 SQL 执行器只暴露三个专用巡检函数对 AI 来说更「傻瓜式」也避免了 AI 生成危险 SQL 的可能。第二get_table_size_ranking里加了WHERE table_schema NOT IN (...)排除系统库这是踩过坑的优化几百张表的环境下速度能差 10 倍。第三慢查询做了双重兜底先查performance_schema查不到再降级查演示表确保你没开慢查询日志也能跑通。现在配 Claude Desktop。配置文件路径在 macOS 是~/Library/Application Support/Claude/claude_desktop_config.jsonWindows 是%APPDATA%\Claude\claude_desktop_config.json。内容如下{ mcpServers: { mysql-inspector: { command: python3, args: [/Users/你的绝对路径/mcp-mysql-inspector/server.py], cwd: /Users/你的绝对路径/mcp-mysql-inspector/ } } }注意args和cwd里的路径必须换成你机器上的绝对路径末尾不要加多余斜杠。JSON 里不能写注释写完用在线校验器过一遍。保存后重启 Claude Desktop在输入框旁边能看到一个小插头图标点开显示mysql-inspector就说明连上了。如果你用的是 Cline配置在 VS Code 的 settings 里格式类似但字段名可能是mcpServers下的command和args。Cline 还支持 MCP 的 SSE 模式但本地巡检用 stdio 就够了。Codex 的auth.json配置逻辑不同它主要管模型鉴权MCP Server 的配置在另一个文件里具体看 TaoToken 的接入文档。4. 验证请求一句话触发巡检并拿到成功结果配置完重启 Claude Desktop插头图标亮起后就可以用自然语言触发巡检了。这一节给三个验证动作每个都有预期结果你对照着看。第一个验证查当前负载。在 Claude 输入框里打「帮我看看数据库现在忙不忙有没有卡住的查询」预期行为Claude 会调用get_connection_status工具返回一个 JSON里面有当前连接数、最大允许连接数、活跃执行线程数以及前 5 条活跃线程的详情。如果某条查询执行时间超过 600 秒Claude 通常会在回复里加一句「检测到一条长时间运行的查询建议立即终止」。成功标志你能看到具体的连接数数字而不是「我无法访问数据库」之类的回复。如果 Claude 说它没有工具可用说明 MCP Server 没连上去检查配置文件路径和 JSON 格式。第二个验证定位磁盘空间。输入「我的服务器磁盘报警了帮我查一下哪个表占空间最大」预期行为Claude 调用get_table_size_ranking返回 Top 15 的表每张表列出数据库名、表名、总大小、数据大小、索引大小、估算行数。因为我们在 SQL 里排除了系统库所以返回的都是业务表。成功标志能看到test_db下的表以及slow_query_log_demo的大小。如果返回空数组说明你的test_db里没有表回去跑一下init_db.py。第三个验证拉慢查询。输入「把最近慢查询排前三的 SQL 发给我看看。」预期行为Claude 调用get_recent_slow_queries传limit3。如果performance_schema里有数据就返回真实的没有就降级查slow_query_log_demo。返回结果包含开始时间、耗时秒、扫描行数、SQL 摘要。成功标志能看到三条 SQL耗时从高到低排列。如果返回「慢查询查询失败」看错误信息里是不是权限问题——performance_schema需要 SELECT 权限。这三个验证跑通说明整条链路是通的Claude 理解自然语言 → 转成 MCP 工具调用 → MCP Server 执行 SQL → 返回结果 → Claude 整理成人话。你可以把这三个问题存成快捷指令每天早上点一下30 秒完成巡检。再进一步你可以让 Claude 把三次结果汇总成一份巡检报告。输入「把刚才的连接状态、表大小、慢查询汇总成一份巡检报告用 Markdown 表格输出。」Claude 会调用三次工具然后把结果拼成一份结构化报告。这份报告你可以直接贴到钉钉或飞书群里。实测下来从输入问题到拿到报告整个过程在 10 秒以内取决于模型响应速度和数据库查询耗时。如果表特别多information_schema查询可能慢一点但加了系统库排除之后一般也在 2 秒内。这里有个小技巧你可以在 Claude 的 Project 功能里把这三个问题设成预设提示词每次打开项目直接点不用重新打字。或者用 Cline 的 Custom Instructions把巡检指令写进去每次对话自动带上。5. 本篇常见错排查401、local proxy failed、reading choices、OAuth这一节按真实报错来每个错误给现象、原因、解决。这些是我在配 MCP TaoToken 通道时踩过的坑你大概率也会遇到其中一两个。报错一401 Unauthorized现象Claude 回复「模型请求失败401」或者 MCP 工具能调用但模型不返回结果。原因TaoToken 的 API Key 没填对或者 Base URL 配错了。401 是鉴权失败说明请求到了服务端但 Key 无效。解决去 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 重新生成一个 Key确认复制时没有多余空格。然后检查客户端的 Base URL 是不是https://taotoken.net/api注意不要带 UTM 参数。Model ID 也要填对不同模型 ID 不一样具体看接入文档 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 。报错二local proxy failed现象客户端提示「local proxy failed」或「connection refused」。原因MCP Server 进程没起来或者 Python 路径不对。Claude Desktop 启动时会去拉起server.py如果command填的python3不在 PATH 里或者args路径写错就会失败。解决先在终端手动跑python3 /你的路径/server.py看能不能正常启动。如果报ModuleNotFoundError: No module named mcp说明依赖没装到当前 Python 环境用pip install mcp pymysql重装。如果手动能跑但 Claude 里不行检查配置文件里的command是不是绝对路径比如/usr/local/bin/python3。报错三reading choices 相关错误现象模型返回结果时解析失败提示「reading choices」或类似字段缺失。原因这通常是模型 API 返回格式和客户端预期不一致。如果你用的是 OpenAI 兼容格式的客户端但 Model ID 填了一个不兼容的模型就会出这个错。解决确认你用的 Model ID 和客户端协议匹配。Claude Desktop 走 Anthropic 格式Cline 可以走 OpenAI 兼容格式。TaoToken 的统一通道支持多种格式但 Model ID 要对应。去 https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 先用网页版验证同一个 Model ID 能不能正常对话能的话再回客户端排查。报错四OAuth 相关错误现象提示「OAuth token expired」或「invalid_grant」。原因如果你用的是 Claude Code 或某些需要 OAuth 的客户端Token 过期了。TaoToken 的 API Key 是长期有效的但客户端自己的 OAuth 会话可能过期。解决重新登录客户端或者清除客户端的缓存 Token 重新授权。Claude Code 的接入方式看 https://taotoken.net/ClaudeCodeAnthropic?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 里面写了 Base URL、Key、Model ID 三件套怎么填。报错五performance_schema 查询返回空现象慢查询工具返回空数组但你知道有慢查询。原因MySQL 默认可能没开events_statements_history_long消费者。解决执行UPDATE performance_schema.setup_consumers SET ENABLED YES WHERE NAME LIKE %history_long%;然后调整performance_schema_events_statements_history_long_size大小。这也是我在代码里加降级查演示表的原因——确保你没开慢查询日志也能跑通演示。报错六information_schema 查询超时现象表特别多时get_table_size_ranking响应很慢甚至超时。原因information_schema.tables在几百张表的环境下查询会慢。解决代码里已经加了WHERE table_schema NOT IN (information_schema, performance_schema, mysql, sys)排除系统库。如果还慢可以再加AND table_rows 0过滤空表。另外可以给information_schema查询加超时设置在pymysql.connect里传read_timeout5。报错七JSON 配置文件格式错误现象Claude 重启后没有插头图标甚至闪退。原因claude_desktop_config.json里有注释、多余逗号、或者路径里的反斜杠没转义。解决JSON 不支持注释删掉所有//和/* */。Windows 路径里的\要写成\\。写完用在线 JSON 校验器过一遍。cwd路径末尾不要加斜杠。排障的通用思路先确认 MCP Server 能手动跑起来再确认客户端能连上 MCP Server最后确认模型通道正常。这三层任何一层出问题表现都是「AI 不干活」但排查方向完全不同。手动跑server.py能排除第一层插头图标能排除第二层网页版对话能排除第三层。6. 语义一致 CTA把巡检闭环跑起来之后跑通这个 Demo 之后你手里就有了一个 7x24 小时在线的巡检值班员。它不会累、不会忘、不会因为早上开会就漏掉巡检。你只需要在钉钉或 Slack 里 一下 AI问一句「库今天状态怎么样」它自己去连 MySQL、执行巡检 SQL、把异常会话和慢查询列表整理好发给你。这套方案的价值不在于技术多新而在于它把 DBA 从重复劳动里解放出来。以前你要写 Shell 脚本、配 cron、搞邮件模板现在只需要一个 MCP Server 加一个客户端配置。权限隔离也更干净——AI 只能调用你暴露的只读巡检接口永远无法执行 DELETE 或 DROP因为工具列表里根本没写。如果你想把这条链路固化下来建议做三件事。第一把server.py里的数据库连接改成从环境变量读取别硬编码密码。第二把三个巡检工具扩展成五个加上主从延迟检查和 InnoDB 缓冲池命中率。第三用 Cline 或 Claude Code 的定时任务功能每天早上 8 点自动触发一次巡检把报告推到群里。TaoToken 在这条链路里负责模型通道的稳定性。你不需要为每个模型单独开账号一个 Key 走统一 API。模型对话入口在 https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 接入文档在 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content API Keys 管理在 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 。长期跑编码或 Agent 任务的话Coding Plan 在 https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 。最后留一个实用技巧把巡检问题存成 Claude 的 Project 预设提示词或者 Cline 的 Custom Instructions。这样每次打开客户端不用重新打字点一下就能触发巡检。如果你用的是 Claude Code可以把巡检指令写进CLAUDE.md每次启动自动加载。数据库巡检这件事从手动敲命令到 AI 自动执行中间隔的不是技术鸿沟而是一个 MCP Server 的配置。你现在就可以打开终端从mkdir mcp-mysql-inspector开始。
