1. 为什么要把 MySQL 数据校验塞进 MCP Server数据校验这件事做过数据质量巡检的朋友都懂写 SQL 查空值、查重复、查外键孤儿、查金额负数一套下来几十条语句跑完还得手动整理成报告。更麻烦的是每次业务方问「昨天订单表有没有异常」你都得重新翻 SQL 文件、改日期、贴结果。我试过把这套流程直接交给 AI 客户端让它自己生成校验 SQL、执行、汇总效率提升非常明显——前提是你得先有一个安全的 MySQL MCP Server 把校验能力暴露出去。MCPModel Context Protocol说白了就是 AI 世界的 USB-C 接口。你按协议写一次工具Claude Desktop、Cursor、Cline 这些支持 MCP 的客户端都能直接调用。把 MySQL 数据校验封装成 MCP Server本质上是给 AI 装了一个「只读数据质检员」它能看到表结构、能跑 SELECT 校验语句、能把结果整理成人话但碰不了 DROP、UPDATE、DELETE。这篇文章面向三类人一是做数据质量巡检的数据工程师想把重复的校验 SQL 沉淀成 AI 可调用的工具二是做入库前校验的后端开发希望在数据写库前让 AI 先跑一遍规则三是刚接触 MCP、想找一个真实可跟做案例的开发者。全文会给到可复制的 MCP Server 配置片段、MySQL 连接参数、校验 SQL 模板以及在 AI 客户端里触发校验、核对返回结果的完整动作。目标只有一个让你一次跑通从配置到校验的闭环。需要先明确一个边界MCP Server 不是让 AI 直连生产库乱来而是通过一层受控的服务端把「能做什么」限定在只读校验范围内。这个思路和 TaoToken 在模型接入层做的事类似——把复杂的对接收敛成标准入口调用方只管用安全边界由中间层守住。2. TaoToken 前置准备模型入口与 Key 获取在配置 MySQL MCP Server 之前得先解决 AI 客户端背后的模型调用问题。MCP 负责把数据库能力暴露给 AI但 AI 本身要能跑起来需要一个稳定的模型入口。TaoToken 在这里扮演的是模型接入层的角色它提供统一的 API 入口兼容主流模型调用格式你拿到 Key 之后Claude Code、Cline、Cursor 这类客户端都能接。先注册并登录控制台地址是 https://taotoken.net/console 。进去之后左侧菜单找「API Keys」新建一个 Key。建议按用途命名比如mysql-mcp-dev方便后面区分。Key 生成后只显示一次复制到安全的地方别直接写进会提交到 Git 的配置文件。拿到 Key 之后你需要确认两件事Base URL 和 Model ID。Base URL 用 https://taotoken.net/api 这是不带 UTM 的纯 API 地址配置到客户端时用这个。Model ID 根据你用的模型填比如claude-sonnet-4-5或gpt-4o这类具体以控制台「模型对话」页面列出的为准。你可以先在 https://taotoken.net/models 里发一条测试消息确认 Key 和模型都通再去配 MCP。这里有个容易踩的坑很多人把 MCP Server 的配置和模型 API 的配置混在一起。它们是两层东西——MCP Server 负责「AI 能调用哪些工具」模型 API 负责「AI 用哪个大脑思考」。两层都要配但配置文件位置不同。模型 API 配在客户端的模型设置里MCP Server 配在客户端的 MCP 配置里。如果你打算长期跑数据校验这类编码/Agent 任务可以看下 Coding Plan地址是 https://taotoken.net/coding-plan 适合高频调用场景。只是偶尔验证一下模型通不通用「模型对话」页面就够了。另外Claude Code 用户如果要用 Anthropic 格式接入可以参考 https://taotoken.net/claude-code-anthropic 这个页面里面有 Base URL、Key、Model ID 三件套的填法。接入文档在 https://taotoken.net/doc 遇到 401 或模型找不到的报错先翻文档里的排障章节。3. 可复制的 MySQL MCP Server 配置片段这一节是全文的核心操作部分。我会给出一份完整的 MCP Server 配置包含 MySQL 连接参数、校验工具定义、以及客户端侧的 JSON 配置。你可以直接复制改路径和账号密码。先看服务端。用 Python 写一个最小可用的 MySQL 校验 MCP Server依赖mcp和mysql-connector-python。安装命令python -m venv mcp-mysql-env source mcp-mysql-env/bin/activate pip install mcp mysql-connector-python python-dotenv服务端核心逻辑分三块连接管理、SQL 安全校验、工具注册。连接参数从环境变量读避免硬编码。下面这份mysql_check_server.py可以直接用#!/usr/bin/env python3 import asyncio import json import os import logging from typing import Any import mysql.connector from dotenv import load_dotenv from mcp.server import Server, NotificationOptions from mcp.server.models import InitializationOptions import mcp.server.stdio import mcp.types as types load_dotenv() logging.basicConfig(levellogging.INFO) logger logging.getLogger(mysql-check-mcp) DB_CONFIG { host: os.getenv(MYSQL_HOST, 127.0.0.1), port: int(os.getenv(MYSQL_PORT, 3306)), user: os.getenv(MYSQL_USER, readonly_user), password: os.getenv(MYSQL_PASSWORD, ), database: os.getenv(MYSQL_DATABASE, ), use_pure: True, connection_timeout: 10, } FORBIDDEN [DROP, TRUNCATE, ALTER, CREATE, INSERT, UPDATE, DELETE, GRANT, REVOKE] def is_safe_select(sql: str) - bool: s sql.strip().upper() if not s.startswith(SELECT): return False for kw in FORBIDDEN: if kw in s: return False if ; in sql.rstrip(;): return False return True def get_conn(): return mysql.connector.connect(**DB_CONFIG) server Server(mysql-check-mcp) server.list_tools() async def list_tools() - list[types.Tool]: return [ types.Tool( namerun_check_sql, description执行只读校验 SQL仅 SELECT返回结果集。用于数据质量巡检。, inputSchema{ type: object, properties: { sql: {type: string, description: 只读 SELECT 校验语句}, limit: {type: integer, default: 200} }, required: [sql] } ), types.Tool( namelist_tables, description列出当前库所有表名, inputSchema{type: object, properties: {}} ), types.Tool( namedescribe_table, description获取指定表的列定义、类型、是否可空, inputSchema{ type: object, properties: {table: {type: string}}, required: [table] } ), types.Tool( namecheck_null_ratio, description统计指定表指定列的空值率用于空值巡检, inputSchema{ type: object, properties: { table: {type: string}, column: {type: string} }, required: [table, column] } ), ] server.call_tool() async def call_tool(name: str, arguments: dict) - list[types.TextContent]: try: if name run_check_sql: sql arguments.get(sql, ) limit arguments.get(limit, 200) if not is_safe_select(sql): return [types.TextContent(typetext, textjson.dumps( {ok: False, error: 仅允许单条只读 SELECT 语句}, ensure_asciiFalse))] if LIMIT not in sql.upper(): sql f{sql.rstrip(;)} LIMIT {limit} conn get_conn() cur conn.cursor(dictionaryTrue) cur.execute(sql) rows cur.fetchall() cur.close() conn.close() return [types.TextContent(typetext, textjson.dumps( {ok: True, row_count: len(rows), rows: rows}, ensure_asciiFalse, defaultstr))] if name list_tables: conn get_conn() cur conn.cursor() cur.execute(SHOW TABLES) tables [r[0] for r in cur.fetchall()] cur.close() conn.close() return [types.TextContent(typetext, textjson.dumps( {ok: True, tables: tables}, ensure_asciiFalse))] if name describe_table: table arguments[table] conn get_conn() cur conn.cursor(dictionaryTrue) cur.execute(fDESCRIBE {table}) cols cur.fetchall() cur.close() conn.close() return [types.TextContent(typetext, textjson.dumps( {ok: True, columns: cols}, ensure_asciiFalse, defaultstr))] if name check_null_ratio: table arguments[table] column arguments[column] sql (fSELECT COUNT(*) AS total, fSUM(CASE WHEN {column} IS NULL THEN 1 ELSE 0 END) AS null_cnt fFROM {table}) conn get_conn() cur conn.cursor(dictionaryTrue) cur.execute(sql) row cur.fetchone() cur.close() conn.close() total row[total] or 1 ratio round((row[null_cnt] or 0) / total, 4) return [types.TextContent(typetext, textjson.dumps( {ok: True, table: table, column: column, total: row[total], null_count: row[null_cnt], null_ratio: ratio}, ensure_asciiFalse))] return [types.TextContent(typetext, textjson.dumps( {ok: False, error: f未知工具 {name}}, ensure_asciiFalse))] except Exception as e: logger.error(tool call failed: %s, e) return [types.TextContent(typetext, textjson.dumps( {ok: False, error: str(e)}, ensure_asciiFalse))] async def main(): async with mcp.server.stdio.stdio_server() as (read, write): await server.run( read, write, InitializationOptions( server_namemysql-check-mcp, server_version1.0.0, capabilitiesserver.get_capabilities( notification_optionsNotificationOptions(), experimental_capabilities{} ) ) ) if __name__ __main__: asyncio.run(main())环境变量文件.env放在同目录MYSQL_HOST127.0.0.1 MYSQL_PORT3306 MYSQL_USERreadonly_user MYSQL_PASSWORDyour_secure_password MYSQL_DATABASEyour_db客户端侧以 Cline 或 Claude Desktop 为例MCP 配置 JSON 如下。注意command指向虚拟环境里的 pythonargs指向脚本绝对路径{ mcpServers: { mysql-check: { command: /path/to/mcp-mysql-env/bin/python, args: [/path/to/mysql_check_server.py], env: { MYSQL_HOST: 127.0.0.1, MYSQL_PORT: 3306, MYSQL_USER: readonly_user, MYSQL_PASSWORD: your_secure_password, MYSQL_DATABASE: your_db } } } }如果你用 Codex 系客户端认证信息写在auth.json里MCP 配置单独一份两者不要混。Base URL 填 https://taotoken.net/api Key 填控制台生成的Model ID 按实际模型填。三件套齐了客户端才能既连上模型、又连上 MCP Server。MySQL 侧务必建一个只读账号别用 rootCREATE USER readonly_user% IDENTIFIED BY your_secure_password; GRANT SELECT ON your_db.* TO readonly_user%; FLUSH PRIVILEGES;这一步是安全底线。MCP Server 里的 SQL 黑名单只是第二道防线数据库权限才是第一道。4. 验证请求在 AI 客户端触发校验并核对结果配置写完之后重启客户端让它重新加载 MCP Server。以 Cline 为例打开 MCP 面板应该能看到mysql-check这个 server状态是 connected展开后列出四个工具run_check_sql、list_tables、describe_table、check_null_ratio。如果状态是 failed先看客户端日志里的 stderr通常是路径写错或依赖没装。验证分三步走。第一步让 AI 列出表帮我列出当前数据库里所有的表。AI 会调用list_tables返回一个表名数组。这一步验证的是 MCP 通信链路通不通。如果这里就报错后面不用继续。第二步让 AI 查表结构看一下 orders 表的列定义。AI 调用describe_table返回列名、类型、是否可空。这一步验证的是工具参数传递是否正确。第三步跑一条真实校验。假设 orders 表有order_date、amount、status三列你想查昨日金额为负的异常订单帮我检查 orders 表里昨天创建的订单有没有金额小于 0 的异常记录把订单号和金额列出来。AI 会生成类似这样的 SQL 并调用run_check_sqlSELECT order_id, amount, status FROM orders WHERE order_date CURDATE() - INTERVAL 1 DAY AND amount 0返回结果是一个 JSON包含ok、row_count、rows。如果row_count为 0说明没有异常如果大于 0AI 会把 rows 里的订单号念给你听。你可以拿这个结果和手动跑 SQL 的结果对一下确认一致。再验证一个空值巡检场景统计 customers 表里 email 列的空值率。AI 调用check_null_ratio返回total、null_count、null_ratio。这个工具的好处是把「统计空值」这种高频校验固化成了参数化调用AI 不用每次现写 SQL减少出错。实测下来从配置到跑通第一条校验顺利的话十分钟内能完成。卡点通常在两处一是 Python 路径没指向虚拟环境导致mcp模块找不到二是 MySQL 账号权限没给够SHOW TABLES能过但SELECT被拒。这两处排查方法在下一节展开。5. 本篇常见报错排查这一节按真实报错来。你在配 MySQL MCP Server 时大概率会遇到下面几类问题。报错一401 Unauthorized 或 invalid api key这个报错来自模型 API 层不是 MCP 层。说明客户端里的 Key 填错了或者 Base URL 写成了带 UTM 的地址。检查两点Base URL 必须是 https://taotoken.net/api 不要带任何查询参数Key 从控制台重新复制一次注意前后不要有空格。如果用的是 Claude Code 的 Anthropic 格式确认ANTHROPIC_BASE_URL和ANTHROPIC_API_KEY都指向 TaoToken 的地址和 KeyModel ID 填控制台列出的名称。改完重启客户端。报错二local proxy failed 或 connection refused这个通常出现在 MCP Server 启动阶段。客户端尝试用 stdio 拉起 Python 进程但进程没起来。原因可能是command路径写错比如写成了系统 python 而不是虚拟环境里的 python。用绝对路径先手动在终端跑一遍/path/to/mcp-mysql-env/bin/python /path/to/mysql_check_server.py如果手动跑能启动并停在等待输入的状态说明脚本没问题是客户端配置路径的问题。如果手动跑就报ModuleNotFoundError: No module named mcp说明依赖装到了别的环境重新在虚拟环境里pip install mcp mysql-connector-python。报错三reading choices 或 model not found这个报错说明模型名填错了。客户端把 Model ID 原样传给 API如果这个 ID 不在可用列表里就返回找不到。去 https://taotoken.net/models 页面确认当前可用的模型名复制准确的 ID 填进去。注意大小写和连字符claude-sonnet-4-5和claude-sonnet-4.5是两回事。报错四OAuth 相关报错或 unauthorized_client如果你用的是需要 OAuth 的客户端而 MCP Server 这边没配认证会出现这类报错。MCP 协议本身支持 OAuth但本文的 MySQL 校验 Server 走的是本地 stdio不涉及 OAuth。出现这个报错通常是客户端把 MCP 认证和模型认证搞混了。检查客户端的 MCP 配置里有没有多余的 auth 字段删掉模型认证单独在模型设置里配。报错五MySQL 连接超时或 access deniedAccess denied for user readonly_userlocalhost说明账号密码不对或者该账号没有从当前主机连接的权限。检查CREATE USER时的 host 部分本地连接用readonly_userlocalhost远程用%。Connection timed out说明 host 或 port 不通先用mysql -h 127.0.0.1 -P 3306 -u readonly_user -p手动连一次确认网络和端口。报错六工具调用返回「仅允许单条只读 SELECT 语句」这是 Server 里的安全校验拦下来的。检查你的 SQL 是不是以 SELECT 开头有没有分号拼接多条语句有没有包含 UPDATE、DELETE 这类关键词。注意子查询里如果出现这些词也会被拦比如SELECT * FROM (DELETE ...)这种非法写法。正常校验 SQL 不会触发。排查顺序建议先确认模型 API 通模型对话页面发消息再确认 MCP Server 能手动启动最后确认客户端配置路径正确。三层分开排查比一上来就改配置高效。6. 把校验能力沉淀成可复用工具链跑通单条校验之后真正有价值的是把校验规则沉淀下来。你可以把常用的校验 SQL 做成模板让 AI 按模板调用。比如订单金额非负校验、用户邮箱空值率、外键孤儿检查各写一条参数化 SQL存成一个checks.yamlchecks: - name: negative_order_amount sql: SELECT order_id, amount FROM orders WHERE order_date CURDATE() - INTERVAL 1 DAY AND amount 0 - name: orphan_orders sql: SELECT o.order_id FROM orders o LEFT JOIN customers c ON o.customer_id c.id WHERE c.id IS NULL - name: null_email_ratio sql: SELECT COUNT(*) AS total, SUM(CASE WHEN email IS NULL THEN 1 ELSE 0 END) AS null_cnt FROM customers然后在 MCP Server 里加一个run_named_check工具读 yaml 按名字执行。这样 AI 只需要说「跑一下 negative_order_amount 这个检查」不用每次现写 SQL。规则集中管理改一处全生效。再进一步可以把校验结果写回一张data_quality_log表记录检查名、执行时间、异常行数。MCP Server 加一个log_check_result工具但注意这个工具需要写权限得单独用一个有 INSERT 权限的账号和只读账号分开。这是权限最小化原则的体现读校验用只读账号写日志用受限写账号两个账号不混用。如果你团队用 Cline 或 CC Switch 管理多个 MCP Server可以把 MySQL 校验 Server 和文件系统 Server、Git Server 并列配置让 AI 在一次对话里既能查库、又能读代码、又能提交报告。CC Switch 的配置里每个 server 独立一段Base URL、Key、Model ID 三件套在模型层配一次即可。长期跑数据质量巡检的话建议把 MCP Server 用 systemd 或 supervisor 托管避免客户端重启后进程丢失。日志输出到文件方便回溯每次校验的执行记录。校验频率高的场景可以在 Server 里加连接池减少频繁建连的开销。最后提醒一句MCP Server 暴露的是能力不是数据本身。AI 看到的是工具返回的 JSON不是整张表。这个边界设计好了数据安全就有保障。校验 SQL 尽量带 WHERE 条件和 LIMIT避免全表扫描把大结果集塞给模型。工具描述里写清楚用途和限制AI 调用时会更准确。整套流程跑下来你会发现数据校验从「人写 SQL 人看结果」变成了「人说需求 AI 跑校验」重复劳动大幅减少。而安全边界由只读账号、SQL 黑名单、LIMIT 保护三层守住比让 AI 直连数据库靠谱得多。