1. 为什么不让 Claude 直接连 MySQL:FastMCP 封装数据库的真实场景
很多人第一次听到「让 Claude 查数据库」,脑子里浮现的画面是:Claude 里填个 MySQL 地址、账号、密码,然后直接SELECT。我试过在本地这么干,能跑通,但只敢在自己电脑上玩。原因很简单——你把数据库的完整读写权限交给了一个会「自由发挥」的模型,它今天心情好帮你查订单,明天可能因为一句模糊指令生成DELETE FROM customer WHERE 1=1。
所以真实可用的架构一定是分层的:Claude / Cursor 作为 MCP Client,通过 MCP 协议调用你写的 FastMCP Server,Server 再去连 MySQL。数据库账号密码只存在于 Server 的环境变量里,AI 客户端永远看不到。这一层 FastMCP Server 既是「工具层」,也是「安全层」和「业务封装层」。
这篇文章要解决的核心检索词就是FastMCP + MySQL 让 Claude 和 Cursor 直接查询数据库。适合谁看?三类人:一是想让 AI 助手查自己业务库的后端/全栈开发者;二是正在搭 AI Agent、需要给模型接内部数据源的工程师;三是用 Cursor 写代码、希望 AI 能顺手查一下测试库表结构的同学。
整条链路长这样:
用户提问 ↓ Claude / Cursor(MCP Client) ↓ MCP 协议 FastMCP Server(你写的 Python 服务) ↓ Service 层(业务逻辑 + 固定 SQL) ↓ MySQL ↓ JSON 结果 ↓ LLM 整理成自然语言关键点在于:AI 不写 SQL,只传参数。你提前把「查客户」「查订单」「查库存」这些动作写成固定工具,模型只负责决定调哪个工具、传什么参数。这样即使模型抽风,最坏结果也只是查错一条数据,而不是删库。
另外还有一个容易被忽略的痛点:Claude Desktop、Cursor、Cline 这些客户端各自要配一套 API Key 和 endpoint,管理起来很碎。本文会把 MCP Server 本身跑在本地,但把模型通道统一收到 TaoToken 上,一个 Key 管所有客户端,后面会给出具体配置。
先把结论放这:FastMCP 负责「让 AI 能安全地碰数据库」,TaoToken 负责「让所有 AI 客户端共用一个通道」。两件事分开做,互不干扰。
2. TaoToken 前置准备:统一 Key 与 MCP 环境搭建
在写 MCP Server 之前,先把两件前置事情做完:一是拿到统一的模型通道 Key,二是把 Python 侧的 FastMCP 环境装好。这两件事都不难,但顺序别搞反,否则后面调试时会分不清是 MCP 的问题还是 Key 的问题。
2.1 为什么要把 endpoint 统一到 TaoToken
Claude Desktop 用 Anthropic 的通道,Cursor 可能用另一家,Cline 又是第三家。每个客户端一套 Key、一套计费、一套额度,时间一长自己都记不清哪个 Key 快到期了。把 endpoint 统一改到 TaoToken 之后,所有客户端共用同一个 Base URL 和同一个 Key,换模型、看用量、控成本都在一个地方。
TaoToken 的 API 地址是https://taotoken.net/api,注意这个地址不带任何查询参数,直接作为 Base URL 用。官网入口是https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content=,注册和看文档都从这里进。
你需要提前准备好三样东西,后面配置里会反复用到:
| 项目 | 说明 | 示例 |
|---|---|---|
| Base URL | 统一模型通道地址 | https://taotoken.net/api |
| API Key | 在控制台创建,形如sk-... | sk-xxxxxxxx |
| Model ID | 具体模型标识,按需选 | claude-sonnet-4-5等 |
API Key 在控制台的 API Keys 页面创建,地址是https://taotoken.net/console/api-keys?utm_source=taotoken_aicg_blog_end&utm_content=api_keys&utm_campaign=rewrite。创建后只显示一次,记得立刻复制存到密码管理器里。
2.2 安装 FastMCP 与 MySQL 驱动
Python 版本建议 3.10 以上,FastMCP 对类型注解依赖比较重。建一个干净的虚拟环境,别在系统 Python 里装:
python -m venv venv source venv/bin/activate # Windows 用 venv\Scripts\activate pip install "mcp[cli]" pymysql python-dotenvmcp[cli]会带上 FastMCP 和调试用的命令行工具,pymysql是纯 Python 的 MySQL 驱动,装起来没有编译依赖,比 mysqlclient 省心。python-dotenv用来读.env文件,避免把数据库密码写死在代码里。
装完验证一下:
python -c "from mcp.server.fastmcp import FastMCP; print('FastMCP OK')"能打印出FastMCP OK就说明环境没问题。如果报ModuleNotFoundError,八成是虚拟环境没激活,或者 pip 装到了别的 Python 上,用which python和which pip确认一下路径一致。
2.3 项目结构
建议一开始就分好层,别把所有代码堆在一个文件里。后面加订单、库存、财务模块时你会感谢自己:
mcp-mysql/ ├── server.py # MCP Server 入口 ├── database.py # 数据库连接管理 ├── config.py # 读环境变量 ├── tools/ │ └── customer.py # 暴露给 AI 的工具 ├── services/ │ └── customer_service.py # 业务逻辑 + 固定 SQL ├── .env # 敏感配置(别提交 Git) └── requirements.txt职责划分很清楚:tools只做参数校验和调用转发,services写真正的 SQL,database.py管连接。AI 看到的是tools里的函数签名,看不到services里的 SQL,这就是安全边界。
3. 可复制配置:FastMCP Server 连接 MySQL 的完整代码
这一节是全文的核心,所有代码都可以直接复制改改就用。我会按「配置 → 连接 → Service → Tool → 启动」的顺序给全,每一步都说明为什么这么写。
3.1 环境变量配置
先建.env文件,把数据库信息和模型通道 Key 都放进去:
# MySQL 配置 MYSQL_HOST=127.0.0.1 MYSQL_PORT=3306 MYSQL_USER=demo_reader MYSQL_PASSWORD=your_db_password MYSQL_DATABASE=crm MYSQL_CHARSET=utf8mb4 # TaoToken 统一通道(供 MCP Server 内部调用模型时使用) TAOTOKEN_BASE_URL=https://taotoken.net/api TAOTOKEN_API_KEY=sk-xxxxxxxxxxxxxxxx TAOTOKEN_MODEL=claude-sonnet-4-5注意数据库账号用只读账号demo_reader,别用 root。这是第一道防线:即使 MCP Server 被攻破,也只能读不能写。
config.py负责把这些变量读进来:
import os from dotenv import load_dotenv load_dotenv() class Config: MYSQL_HOST = os.getenv("MYSQL_HOST", "127.0.0.1") MYSQL_PORT = int(os.getenv("MYSQL_PORT", "3306")) MYSQL_USER = os.getenv("MYSQL_USER") MYSQL_PASSWORD = os.getenv("MYSQL_PASSWORD") MYSQL_DATABASE = os.getenv("MYSQL_DATABASE") MYSQL_CHARSET = os.getenv("MYSQL_CHARSET", "utf8mb4") TAOTOKEN_BASE_URL = os.getenv("TAOTOKEN_BASE_URL", "https://taotoken.net/api") TAOTOKEN_API_KEY = os.getenv("TAOTOKEN_API_KEY") TAOTOKEN_MODEL = os.getenv("TAOTOKEN_MODEL", "claude-sonnet-4-5") config = Config()3.2 数据库连接管理
database.py用一个简单的连接工厂,配合上下文管理器保证连接释放:
import pymysql from contextlib import contextmanager from config import config @contextmanager def get_connection(): conn = pymysql.connect( host=config.MYSQL_HOST, port=config.MYSQL_PORT, user=config.MYSQL_USER, password=config.MYSQL_PASSWORD, database=config.MYSQL_DATABASE, charset=config.MYSQL_CHARSET, cursorclass=pymysql.cursors.DictCursor, autocommit=True, ) try: yield conn finally: conn.close()cursorclass=pymysql.cursors.DictCursor是关键,它让查询结果直接返回字典而不是元组。前面 excerpt 里提到过,LLM 对 JSON 的理解远好于元组,{"id": 1001, "name": "张三"}比(1001, "张三")好处理得多。
3.3 Service 层:固定 SQL
services/customer_service.py里写死 SQL,只接受参数,不接受拼接:
from database import get_connection class CustomerService: def query_by_id(self, customer_id: int): sql = """ SELECT id, name, phone, level, created_at FROM customer WHERE id = %s """ with get_connection() as conn: with conn.cursor() as cursor: cursor.execute(sql, (customer_id,)) row = cursor.fetchone() if not row: return {"success": False, "message": f"客户 {customer_id} 不存在"} return {"success": True, "data": row} def top_customers(self, days: int = 30, limit: int = 10): sql = """ SELECT c.id, c.name, SUM(o.amount) AS total_amount FROM customer c JOIN orders o ON o.customer_id = c.id WHERE o.created_at >= DATE_SUB(NOW(), INTERVAL %s DAY) GROUP BY c.id, c.name ORDER BY total_amount DESC LIMIT %s """ with get_connection() as conn: with conn.cursor() as cursor: cursor.execute(sql, (days, limit)) rows = cursor.fetchall() return {"success": True, "data": rows}注意%s占位符是 pymysql 的参数化写法,它会把参数安全转义,杜绝 SQL 注入。永远不要用 f-string 拼 SQL,这是底线。
3.4 Tool 层:暴露给 AI 的接口
tools/customer.py里定义 AI 能看到的工具,函数签名和 docstring 就是给模型看的「说明书」:
from mcp.server.fastmcp import FastMCP from services.customer_service import CustomerService service = CustomerService() def register_customer_tools(app: FastMCP): @app.tool() def query_customer(customer_id: int) -> dict: """根据客户 ID 查询客户基本信息(姓名、电话、等级)。 Args: customer_id: 客户编号,整数 """ try: return service.query_by_id(customer_id) except Exception as e: return {"success": False, "message": str(e)} @app.tool() def top_customers(days: int = 30, limit: int = 10) -> dict: """查询最近 N 天成交金额最高的前 M 位客户。 Args: days: 统计天数,默认 30 limit: 返回条数,默认 10 """ try: return service.top_customers(days, limit) except Exception as e: return {"success": False, "message": str(e)}docstring 一定要写清楚参数含义,模型就是靠这个决定传什么值的。异常统一 catch 后返回{"success": False, ...},别让异常直接抛出去,否则客户端只会看到一个看不懂的堆栈。
3.5 Server 入口
server.py把所有工具注册进来并启动:
from mcp.server.fastmcp import FastMCP from tools.customer import register_customer_tools app = FastMCP("crm-mysql-server") register_customer_tools(app) if __name__ == "__main__": app.run()app.run()默认走 stdio 传输,这是 Claude Desktop 和 Cursor 最常用的方式。启动后进程会挂在标准输入输出上等客户端连接,不会打印一堆日志,这是正常的。
3.6 客户端配置片段
Claude Desktop 的配置文件在~/Library/Application Support/Claude/claude_desktop_config.json(macOS)或%APPDATA%\Claude\claude_desktop_config.json(Windows):
{ "mcpServers": { "crm-mysql": { "command": "/path/to/venv/bin/python", "args": ["/path/to/mcp-mysql/server.py"], "env": { "MYSQL_HOST": "127.0.0.1", "MYSQL_PORT": "3306", "MYSQL_USER": "demo_reader", "MYSQL_PASSWORD": "your_db_password", "MYSQL_DATABASE": "crm" } } } }Cursor 的 MCP 配置在~/.cursor/mcp.json,结构类似:
{ "mcpServers": { "crm-mysql": { "command": "/path/to/venv/bin/python", "args": ["/path/to/mcp-mysql/server.py"] } } }command一定要写虚拟环境里 Python 的绝对路径,别写python,否则客户端可能用系统 Python 启动,找不到你装的mcp包。
4. 验证请求:在 Claude 和 Cursor 里跑通一次真实查询
配置写完不算完,得真跑一次查询才算数。这一节给出完整的验证步骤和预期结果。
4.1 先用 MCP Inspector 本地自测
在接客户端之前,先用官方调试工具确认 Server 本身没问题:
mcp dev server.py它会启动一个本地 Web 界面,列出所有注册的工具。点开query_customer,填customer_id=10086,点运行。如果返回:
{ "success": true, "data": { "id": 10086, "name": "张三", "phone": "138****8888", "level": "VIP", "created_at": "2025-03-12T10:20:00" } }说明 Server 到 MySQL 这条链路是通的。如果这里就报错,先别急着配客户端,把错误解决掉。
4.2 在 Claude Desktop 中验证
重启 Claude Desktop,让它重新加载配置。在对话框右下角能看到一个工具图标,点开应该能看到crm-mysql这个 Server 和它下面的两个工具。
然后直接问:
帮我查一下 10086 号客户的信息
Claude 会先弹出一个工具调用确认框,显示它准备调用query_customer,参数是{"customer_id": 10086}。点允许,它会执行并返回:
客户 10086 的姓名是张三,会员等级为 VIP,联系电话 138****8888。
整个过程 Claude 没有写一行 SQL,它只是决定「调哪个工具、传什么参数」。这就是分层设计的价值。
4.3 在 Cursor 中验证
Cursor 里按Cmd/Ctrl + L打开 Chat,切到 Agent 模式。同样问:
查一下最近 30 天成交额最高的 5 位客户
Cursor 会调用top_customers,参数{"days": 30, "limit": 5},返回一个列表。你可以让它把结果整理成表格,它会直接输出 Markdown 表格。
4.4 把模型通道切到 TaoToken
上面两步验证的是 MCP 链路。如果你还想让 Cursor 或 Cline 这类客户端的模型请求也走统一通道,就在客户端设置里改 Base URL。以 Cursor 为例,在 Settings → Models 里把 OpenAI/Anthropic 的 Base URL 改成https://taotoken.net/api,API Key 填 TaoToken 控制台创建的那个,Model 填claude-sonnet-4-5。
Cline 的配置在扩展设置里,同样是三件套:
| 配置项 | 值 |
|---|---|
| Base URL | https://taotoken.net/api |
| API Key | sk-xxxxxxxx(TaoToken 控制台创建) |
| Model ID | claude-sonnet-4-5 |
改完之后,MCP 工具调用走本地 FastMCP Server,模型推理走 TaoToken,两条链路互不影响。想验证模型通道是否生效,可以在模型对话页面发一条测试消息,地址是https://taotoken.net/chat?utm_source=taotoken_aicg_blog_end&utm_content=model_chat&utm_campaign=rewrite。
5. 本篇常见错误排查:401、local proxy failed、reading choices 全解析
配置 MCP + 数据库这类项目,报错基本集中在几个固定位置。这一节按真实报错信息对照排查,遇到问题直接对号入座。
5.1 401 Unauthorized
现象:客户端调用模型时报 401,或者 MCP Server 内部调用 TaoToken 时报 401。
原因:API Key 错误、过期、或者带了多余空格。复制 Key 时经常把首尾空格一起复制进去。
排查:
curl -X POST https://taotoken.net/api/v1/chat/completions \ -H "Authorization: Bearer sk-xxxxxxxx" \ -H "Content-Type: application/json" \ -d '{"model":"claude-sonnet-4-5","messages":[{"role":"user","content":"hi"}]}'如果这条命令返回 401,说明 Key 本身有问题,去控制台重新创建一个。如果返回正常,说明 Key 没问题,是客户端配置里写错了。注意Bearer和 Key 之间是一个空格,别多别少。
5.2 local proxy failed / connection refused
现象:Claude Desktop 启动后工具图标是灰的,日志里出现local proxy failed或connection refused。
原因:MCP Server 进程没起来,或者command路径写错。
排查:先在终端手动跑一遍:
/path/to/venv/bin/python /path/to/mcp-mysql/server.py如果报ModuleNotFoundError: No module named 'mcp',说明这个 Python 不是装依赖的那个。用which python确认虚拟环境路径,把配置里的command改成绝对路径。如果手动跑没报错但客户端还是连不上,检查配置文件 JSON 格式是否合法,多一个逗号都会导致整个配置加载失败。
5.3 reading 'choices' / undefined is not an object
现象:客户端报Cannot read properties of undefined (reading 'choices')。
原因:模型通道返回的结构和客户端预期的不一致。常见于 Base URL 写成了https://taotoken.net(少了/api),或者写成了带/v1的完整路径导致重复。
排查:Base URL 统一写https://taotoken.net/api,不要自己加/v1,客户端会自动补。用 curl 测一下:
curl https://taotoken.net/api/v1/models \ -H "Authorization: Bearer sk-xxxxxxxx"能返回模型列表就说明地址对了。
5.4 OAuth / authentication_error
现象:Claude Code 或某些客户端报 OAuth 相关错误。
原因:客户端默认走 OAuth 流程,但你用的是 API Key 模式。
排查:在客户端设置里明确选择「API Key」认证方式,别选 OAuth。Claude Code 的话,检查~/.claude/settings.json里的env段:
{ "env": { "ANTHROPIC_BASE_URL": "https://taotoken.net/api", "ANTHROPIC_API_KEY": "sk-xxxxxxxx" } }5.5 工具调用返回空 / 模型说「我没有这个工具」
现象:模型回复「我无法访问数据库」或「没有可用工具」。
原因:MCP Server 没被客户端识别,或者工具注册失败。
排查:用mcp dev server.py确认工具列表里有query_customer。如果 Inspector 里能看到但客户端看不到,重启客户端。Claude Desktop 对配置变更不敏感,必须完全退出再启动,不是关窗口。
5.6 数据库连接超时
现象:工具调用卡住很久然后返回(2003, "Can't connect to MySQL server")。
原因:MySQL 没启动、端口不对、或者账号没有远程访问权限。
排查:
mysql -h 127.0.0.1 -P 3306 -u demo_reader -p crm -e "SELECT 1"能连上说明数据库没问题,问题在 MCP Server 的环境变量。检查.env里的MYSQL_HOST是不是写成了localhost(某些系统下会走 socket 而不是 TCP),统一用127.0.0.1。
6. 长期编码与 Agent 场景:把通道和工具都管起来
MCP Server 跑通只是起点。真正长期用起来,你会遇到两个管理问题:一是模型通道的额度和成本要统一看,二是 MCP 工具会越加越多,得有地方管。
模型通道这块,把所有客户端的 Base URL 都指向https://taotoken.net/api之后,用量和成本在控制台一处可见。如果你打算长期跑编码 Agent、让 Cursor 或 Claude Code 持续调用,可以了解下 Coding Plan,地址是https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding_plan&utm_campaign=rewrite。它适合高频编码场景,比按量付费更可控。
MCP 工具这块,建议按业务域拆文件。客户、订单、库存各一个tools/xxx.py,各自对应一个services/xxx_service.py。server.py里统一注册:
from tools.customer import register_customer_tools from tools.order import register_order_tools from tools.inventory import register_inventory_tools app = FastMCP("crm-mysql-server") register_customer_tools(app) register_order_tools(app) register_inventory_tools(app)工具多了之后,docstring 的质量直接决定模型选工具的准确率。写清楚「什么时候用这个工具」,比写清楚「这个工具做什么」更重要。比如top_customers的 docstring 里加一句「当用户问『成交额最高』『大客户』『Top N』时使用」,模型命中率会明显提升。
安全上还有几个长期要守的规矩:数据库账号永远只读;SQL 永远参数化;工具永远只暴露固定业务动作,不暴露execute_sql;每次工具调用记一条日志,包含时间、工具名、参数、结果状态。日志不用多复杂,写到一个本地文件就够排查用了:
import logging logging.basicConfig( filename="mcp_audit.log", level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s" )在 Tool 里调用前后各记一条,出问题时能快速定位是模型传错了参数,还是数据库返回了异常。
最后留一个实用技巧:调试 MCP 工具时,把mcp dev server.py一直开着,改完代码它会自动重载,比每次重启客户端快得多。等 Inspector 里验证通过了,再去客户端点确认,能省掉大量来回折腾的时间。