用 Python 自建 MySQL MCP 服务器,让 Claude Code 获得数据库能力

在上一篇Claude Code MCP 入门中，我们了解了 MCP 的三层架构、两种传输方式和作用域概念，并通过 Playwright 体验了 MCP 让 Claude Code "操控浏览器"的能力。

MCP 的本质是一个标准化的"工具接口协议"，只要有人按照 MCP 规范写一个服务器，Claude Code 就能通过它调用对应能力。

那么，如果我想让 Claude Code 直接查询和操作本地或远程的 MySQL 数据库呢？能不能自己写一个 MCP 服务器来暴露数据库能力？答案是可以的。

本文将带你从零开始，用 Python 搭建一个属于自己的 MySQL MCP 服务器，让 Claude Code 获得"数据库助手"的能力。

自建过程能让你直观理解 MCP 的三个核心能力概念(也叫原语)——Tools、Resources、Prompts是如何协同工作的，以及大模型是如何通过多轮工具调用逐步"探索"数据库结构的。

01 环境准备

第 1 步：安装 MySQL

首先需要在本地安装 MySQL 数据库。

推荐使用官方 Installer，下载地址为 https://dev.mysql.com/downloads/installer/（选择 mysql-installer-community）。

安装过程中注意以下几点：

• 安装类型：选择 "Server only" 或 "Developer Default"
• 端口配置：Type and Networking 环节，端口保持默认 3306
• 认证方式：选择 Use Strong Password Encryption
• 账户密码：设置 root 密码并牢记
• Windows 服务：勾选开机自启，服务名默认为 MySQL80

安装完成后，验证服务是否正常启动：

Get-Service | Where-Object { $_.Name -like "*mysql*" }

另外需要将 MySQL 的 bin 目录（一般为 C:\Program Files\MySQL\MySQL Server 8.0\bin）添加到系统环境变量 Path 中，以便在任意位置使用 mysql 命令行工具。

第 2 步：建库建表 + 创建只读账号

MySQL 安装完成后，登录并创建测试数据库和表：

-- 建库CREATEDATABASE testdb DEFAULTCHARACTERSET utf8mb4;USE testdb;-- 建表CREATETABLEusers (idINT PRIMARY KEY AUTO_INCREMENT,nameVARCHAR(50),    department VARCHAR(50),    age INT,    created_at DATETIME DEFAULTCURRENT_TIMESTAMP);-- 插入测试数据INSERTINTOusers (name, department, age) VALUES('张三', '技术部', 28),('李四', '市场部', 32),('王五', '技术部', 25),('赵六', '财务部', 40);

接下来创建一个只读账号供 MCP 服务器使用，这样即使 MCP 服务器被恶意利用，攻击者也只能读取数据而不能修改或删除。

CREATEUSER'mcp_ro'@'localhost'IDENTIFIEDBY'McpRead@123';GRANTSELECTON testdb.* TO'mcp_ro'@'localhost';FLUSHPRIVILEGES;

PS：'mcp_ro'@'localhost' 只允许本地回环连接。如果用机器名或局域网 IP（如 192.168.x.x）连接，会报 1130 Host '...' is not allowed 错误。连接时 host 必须使用 127.0.0.1，不要用局域网 IP。

第 3 步：Python 环境

Python 的包安装是全局的，如果不使用虚拟环境，所有项目都共享同一套第三方包。

当不同项目依赖不同版本的同一个库时（比如一个项目需要 mcp 1.x，另一个需要 2.x），就会产生冲突。

虚拟环境为每个项目创建独立的 Python 运行空间和包目录，互不干扰。

创建虚拟环境并安装依赖

mkdir D:\mcp_mysqlcd D:\mcp_mysqlpython -m venv .venv.\.venv\Scripts\Activate.ps1        # 激活虚拟环境

如果遇到"禁止运行脚本"的错误，需要先设置执行策略：

Set-ExecutionPolicy -Scope CurrentUser RemoteSigned

然后安装所需依赖：

pip install "mcp[cli]" pymysql

这里需要两个核心依赖：

如果提示找不到 mcp[cli]，可能是 Python 版本太旧（如 3.8）。mcp 要求 Python >= 3.10，建议安装新版 Python 后重试。

02 整体架构

在编写代码之前，先理解整个系统的架构。以下是数据流向示意图：

整个系统涉及四个角色：

• Claude Code：宿主/客户端，既连接大模型也连接 MCP 服务
• 大模型（LLM）：决策大脑，决定何时调用哪个工具、生成什么 SQL
• MCP 服务（server.py）：暴露 Tools、Resources、Prompts 三大原语
• MySQL：真实的数据源

stdio 模式的 MCP 服务器不是常驻后台服务，而是由 Claude Code 在每次会话启动时按需拉起的子进程，会话结束即被杀掉。

这意味着每次启动 Claude Code 时，MCP 服务器都会重新初始化。

03 MCP Server

项目结构

D:\mcp_mysql\├── .venv\├── server.py            # MCP 服务器主文件（Tools + Resources + Prompts）└── test_conn.py         # MySQL 连接测试脚本

完整的 MCP 服务器

使用 FastMCP 框架，同时包含三种核心原语：

# server.py —— 本地 MySQL 的 MCP 服务器import osimport pymysqlfrom mcp.server.fastmcp import FastMCPmcp = FastMCP("mysql-server")DB_CONFIG = {"host": os.getenv("MYSQL_HOST", "127.0.0.1"),  # 本地localhost"port": int(os.getenv("MYSQL_PORT", "3306")),"user": os.getenv("MYSQL_USER", "mcp_ro"),"password": os.getenv("MYSQL_PASSWORD", "McpRead@123"),"database": os.getenv("MYSQL_DATABASE", "testdb"),"charset": "utf8mb4","cursorclass": pymysql.cursors.DictCursor,}defget_conn():return pymysql.connect(**DB_CONFIG)def_rows_to_text(rows) -> str:"""将查询结果格式化为可读文本。"""ifnot rows:return"查询成功，但无数据返回。"    headers = list(rows[0].keys())    lines = [" | ".join(headers), "-" * len(" | ".join(headers))]for r in rows:        lines.append(" | ".join(""if r[h] isNoneelse str(r[h]) for h in headers        ))return"\n".join(lines)

上述代码完成了服务器实例化和配置管理。

数据库连接参数通过环境变量读取，既方便配置又避免了硬编码密码。

_rows_to_text 辅助函数将查询结果格式化为易读的表格文本。

接下来是三个核心部分的实现。

第一部分：Tools

Tools 是大模型自主决定调用的能力，类比 HTTP 的 POST 请求：

# ---- Tools（AI 主动调用）----@mcp.tool()defrun_query(sql: str) -> str:"""执行只读 SQL（仅 SELECT/SHOW/DESC/EXPLAIN）。"""    stripped = sql.strip().lower().rstrip(";")ifnot stripped.startswith(("select", "show", "desc", "explain")):return"仅允许 SELECT/SHOW/DESC/EXPLAIN。"try:        conn = get_conn()with conn.cursor() as cur:            cur.execute(sql)            rows = cur.fetchmany(200)        conn.close()return _rows_to_text(rows)except Exception as e:returnf"查询出错：{e}"@mcp.tool()deflist_tables() -> str:"""列出所有表名。"""    conn = get_conn()with conn.cursor() as cur:        cur.execute("SHOW TABLES")        rows = cur.fetchall()    conn.close()return"\n".join(str(list(r.values())[0]) for r in rows) or"无表"@mcp.tool()defdescribe_table(table: str) -> str:"""查看某张表的字段结构。"""    conn = get_conn()with conn.cursor() as cur:        cur.execute(f"DESC `{table}`")        rows = cur.fetchall()    conn.close()return _rows_to_text(rows)

这里注册了三个工具：run_query/list_tables/describe_table

run_query：执行只读 SQL 语句。内置安全校验，只允许 SELECT、SHOW、DESC、EXPLAIN 开头的语句，防止数据被意外修改。

结果限制最多 200 行，避免大查询阻塞。

list_tables：列出数据库中所有表名，是大模型"探索"数据库的第一步。

describe_table：查看指定表的字段结构（类型、是否可空等），帮助大模型理解表结构以生成正确的 SQL。

工具的 docstring（即 """..."""）就是给大模型看的"说明书"。写得越清楚，大模型用得越准确。

第二部分：Resources

Resources 由客户端主动加载，作为背景知识"喂"给大模型，类比 HTTP 的 GET 请求：

# ---- Resources（只读上下文，客户端主动加载）----@mcp.resource("schema://tables")defall_tables_schema() -> str:"""全库所有表结构，作为背景知识。"""    conn = get_conn()    out = []with conn.cursor() as cur:        cur.execute("SHOW TABLES")        tables = [list(r.values())[0] for r in cur.fetchall()]for t in tables:            cur.execute(f"DESC `{t}`")            out.append(f"### 表 `{t}`\n" + _rows_to_text(cur.fetchall()))    conn.close()return"\n\n".join(out)@mcp.resource("schema://table/{name}")defone_table_schema(name: str) -> str:"""指定表的结构（带 URI 参数）。"""    conn = get_conn()with conn.cursor() as cur:        cur.execute(f"DESC `{name}`")        cols = cur.fetchall()    conn.close()returnf"表 `{name}` 的结构：\n" + _rows_to_text(cols)

这里注册了两个资源：

schema://tables：一次性返回全库所有表的完整结构。大模型加载后就有了全部背景知识，写 SQL 时不再需要多轮探路。

schema://table/{name}：按需返回指定单张表的结构，URI 中带参数 {name}，支持动态查询。

第三部分：Prompts

Prompts 由用户主动触发，返回一段预置的提问文本，类比 HTTP 模板：

# ---- Prompts（用户触发的模板）----@mcp.prompt()defanalyze_table(table: str) -> str:"""标准化表分析流程模板。"""returnf"""请对 `{table}` 表做完整数据分析：1. 调 describe_table 看结构2. run_query 统计总行数3. 看 2~3 个关键字段分布4. 检查空值/异常5. 给出中文结论"""@mcp.prompt()defsql_review(sql: str) -> str:"""SQL 审查模板。"""returnf"请审查这条 SQL 的正确性与性能：\n```sql\n{sql}\n```"

这里注册了两个提示词模板：

analyze_table：将常用的表分析流程固化成标准步骤，用户只需传入表名即可一键启动完整分析。

sql_review：用于审查 SQL 语句的正确性和性能，适合开发阶段的代码审查场景。

启动服务器

# ---- 启动 ----if __name__ == "__main__":    mcp.run(transport="stdio")

04 本地测试

在接入 Claude Code 之前，建议先进行本地验证，确保服务器能正常运行。

启动服务器：

.\.venv\Scripts\python.exe server.py

服务器启动后会通过 stdio 模式等待 Host（Claude Code）发起连接。

此时 Claude Code 还无法直接使用，需要在 Claude Code 中注册这个 MCP 服务器。

05 接入 Claude Code

配置 MCP 服务器

推荐使用环境变量传递数据库配置，避免在代码中写死密码：

claude mcp add mysql-server --scope user `  --env MYSQL_HOST=127.0.0.1 `  --env MYSQL_PORT=3306 `  --env MYSQL_USER=mcp_ro `  --env MYSQL_PASSWORD=McpRead@123 `  --env MYSQL_DATABASE=testdb `  -- D:\mcp_mysql\.venv\Scripts\python.exe D:\mcp_mysql\server.py

参数说明：

• --scope user：写入用户级全局配置，所有项目都可用
• --env：传递环境变量，MCP 服务器启动时会读取这些变量
• -- 之后的部分是启动服务器的完整命令（含虚拟环境路径）

关于作用域的详细对比，可参考之前的文章。

验证连接

配置完成后，执行以下命令确认连接状态：

claude mcp list

如果看到 mysql-server 服务器旁边显示 ✔ Connected，说明连接成功。

实际使用

启动 Claude Code 对话：

claude

在对话中直接输入自然语言：

> 数据库里有哪些表？            → 调 list_tables> users 表的结构是什么？        → 调 describe_table> 查询技术部的所有员工          → 生成SQL → 调 run_query> 统计每个部门的人数            → 生成SQL → 调 run_query

06 MCP 三大核心原语详解

前面展示了 Tools、Resources、Prompts 的实际用法。下面深入理解这三个核心概念。

核心概念对比

注意到在 Claude Code 的 /mcp 主界面中，常常只显示 Tools；Resources 需要用 @ 符号引用、Prompts 需要用 / 斜杠命令才能看到。这是 UI 设计的差异，不代表注册失败。

Tools：

Resources：

Prompts：

下面依次介绍每种原语的执行流程。

执行流程：Tools

这里关键点是大模型自主判断需要调哪个工具、传什么参数。无论哪种原语，都涉及 4 个角色，注意谁发起、数据往哪流：

用户："技术部有几个人"      │      ▼Claude Code：把【问题 + 可用工具清单(name/描述/参数)】发给大模型      │  （工具清单包含：tool name、描述 docstring、参数类型）      ▼大模型：判断"需要查数据库" → 决定调用 run_query，生成 SQL      │  （返回一个"工具调用请求"，不是最终答案）      ▼Claude Code：收到请求 → 调用 MCP 服务的 run_query(sql)      │      ▼MCP 服务：连 MySQL 执行 SQL → 拿到结果      │  （安全校验 → 执行查询 → fetchmany(200) 限制行数）      ▼Claude Code：把查询结果塞回上下文，再次发给大模型      │      ▼大模型：基于结果用自然语言总结"技术部有 3 人"      │      ▼用户：看到最终答案

发起者是大模型，用户只是提了需求，没有指定用哪个工具。

数据流从 MySQL → MCP → Claude Code → 大模型 → 用户

整个过程可能多轮，大模型可能需要多次调用不同工具才能得出答案。

在编写 MCP 服务器时，花心思写好每个工具的说明文档，能显著提升交互体验。

执行流程：Prompts

Prompts 是一种"预置好的提问模板"，由用户主动触发。

MCP 服务将模板填入参数后，返回一段结构化的提示词文本，这段文本会被当作用户的输入发送给大模型：

用户：输入 /mysql-server:analyze_table，填参数 table=users      │      ▼Claude Code：调用 MCP 服务的 analyze_table("users")      │      ▼MCP 服务：把模板填入参数 → 返回一段提示词文本      │   "请对 users 表做分析：1.看结构 2.统计行数 3.看分布..."      ▼Claude Code：把这段文本当作【用户的输入】发给大模型      │      ▼大模型：按模板里的步骤执行      │   （步骤里让它 describe_table / run_query → 于是又触发 Tools 流程）      ▼用户：看到按标准流程产出的分析结论

发起者是用户，主动选斜杠命令，不是大模型自主决定。

Prompt 本身不查数据库，它只负责"生成一段高质量提问"，只是提问中可能会引导大模型去调 Tools，这才能真正拿到数据。

把常用工作流固化成一键模板，避免每次手打一长串提问。

在 Claude Code 中，Prompts 通过输入 / 斜杠命令触发。例如输入 /mysql-server:analyze_table 并填入表名，就能一键启动完整的表分析流程。

执行流程：Resources

Resources 提供的是只读的背景数据，由客户端（用户）主动加载。

加载完成后，这些数据会作为"背景上下文"出现在对话中，大模型在后续回答问题时就能直接利用这些信息，不再需要多轮探路。

用户：输入 @mysql-server:schema://tables      │      ▼Claude Code：调用 MCP 服务读取该资源内容      │      ▼MCP 服务：查全库表结构 → 返回只读文本（各表字段清单）      │      ▼Claude Code：把这段内容作为"背景上下文"加载进对话      │   （此时还没问问题，只是先"喂资料"）      ▼用户：接着提问"技术部有几个人"      │      ▼大模型：上下文里"已经有"表结构 → 直接生成正确 SQL      │   （跳过 list_tables / describe_table 探路！）      ▼Claude Code → MCP 服务调 run_query → 返回结果 → 大模型总结 → 用户

发起者是用户/客户端，不是大模型主动去"拿"，而是用户先加载好的。

在提问之前先加载好，作为背景知识，这省掉大模型的探路轮次，使 SQL 更快更准。

在 Claude Code 中，Resources 通过输入 @ 引用触发。例如输入 @mysql-server:schema://tables 就能一次性加载全库表结构。加载后，大模型在后续对话中"已经知道"了表结构，可以直接写出正确的 SQL，不再需要多轮探路。

三者对比总结

一个大模型一开始并不知道数据库里有哪些表、每张表有哪些字段，那它是怎么写出正确的 SQL 的呢？答案是靠多轮工具调用探路。

你问："技术部有几个人"第①轮：大模型不知有哪些表 → 调 list_tables → 返回 users第②轮：不知字段 → 调 describe_table('users') → 返回 department 等字段第③轮：现在知道结构了 → 调 run_query("SELECT COUNT(*) FROM users WHERE department='技术部'")第④轮：自然语言回答"技术部 3 人"

大模型启动时只拿到工具的说明书（name + description + 参数），不含实际数据。

所以要真正调用工具才知道表里有什么，通过list_tables → describe_table → run_query，逐步探索。

这正是 Agent 的「思考→行动→观察→再思考」循环。

而 Resources 和 Prompts 的价值是减少探路的轮次

• Resources：提前把表结构喂进上下文 → 大模型跳过 list_tables 和 describe_table，直接写对 SQL（更快、省 token、少出错）。
• Prompts：把"该怎么探路"固化成标准步骤模板，避免遗漏关键环节。

写在最后

本文从零开始，演示了如何用 Python 自建一个 MySQL MCP 服务器。通过 FastMCP 框架的 @tool、@resource、@prompt 三大核心装饰器，将 MySQL 的查询、表管理、结构查看等能力以 MCP 协议的标准格式暴露出来，并通过 claude mcp add 命令接入 Claude Code。

MCP 协议的强大之处就在于这种"即插即用"的扩展能力。不需要修改 Claude Code 本身，只需要按照规范写一个服务器，就能无限扩展它的能力边界。

如果本文对你有帮助的话，别忘记点赞、转发！也欢迎在评论区交流~

关注【伍肆聊AI】，持续更新Claude Code教程、实战模板、避坑干货。