基于AI MCP协议，用Python写一个连接数据库执行SQL的MCP服务：TaoToken统一Key接入与config.toml配置骨架

1. 从「AI 只会聊天」到「AI 能查库」MCP 服务到底解决什么问题你可能遇到过这种场景让 AI 帮你分析一张订单表它讲得头头是道但数据是它编的你想让它直接跑一条 SQL 看看真实结果它只能说「我无法访问你的数据库」。MCPModel Context Protocol就是补上这块短板的协议层——它把 AI 客户端和外部工具、数据源用统一标准连起来让模型从「只会说」变成「能动手」。MCP 服务本质上是一个独立进程通过 stdio 或 HTTP 跟 AI 客户端通信。客户端把可用工具列表告诉模型模型决定调用哪个工具、传什么参数服务端执行完把结构化结果回传。用 Python 写一个连接数据库执行 SQL 的 MCP 服务核心就三件事定义工具函数、管理数据库连接配置、把服务注册到 AI 客户端。这篇面向想自己动手的开发者从零写一个 MySQL 查询 MCP 服务给出可复制的config.toml配置骨架并用 TaoToken 统一 Key 接入 AI 工具链最后跑一次完整的 SQL 查询验证。适合有 Python 基础、想让 AI 直接操作数据库做分析或排障的人。2. TaoToken 前置准备统一 Key 与接入地址在写 MCP 服务之前先把 AI 侧的接入准备好。TaoToken 提供统一的 API Key兼容 Anthropic 风格的接口MCP 客户端比如 Claude Code、Cursor 等配置时只需要填一个 base URL 和一个 Key不用为每个模型单独维护凭证。你需要做两件事拿到 API Key记住接入地址。官网入口https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentAPI 基地址https://taotoken.net/api获取 Keyhttps://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi_keysutm_campaignrewrite接入文档https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite注意API 基地址不带 UTM 参数直接写https://taotoken.net/api即可。Key 只在客户端配置里出现不要硬编码进 MCP 服务脚本。如果你用的是 Claude Code 这类命令行工具接入配置可以走~/.claude/settings.json或环境变量如果是 Cursor在设置里填 Anthropic API 的 base URL 和 Key。具体路径参考接入文档不同客户端字段名略有差异。3. 可复制配置config.toml 骨架与 MCP 服务脚本3.1 config.toml 配置骨架把数据库连接信息和 MCP 服务参数抽到config.toml脚本只读配置不写死密码。下面这份骨架可以直接复制改# config.toml [mcp] name mysql-sql-service transport stdio # 本地调试用 stdio远程部署改 streamable-http host 0.0.0.0 port 8000 path /sql-service [security] default_limit 10 # SELECT 未写 LIMIT 时自动追加 max_rows 500 # 单次查询返回上限 allow_write false # 是否允许 INSERT/UPDATE/DELETE [databases.test] host 127.0.0.1 port 3306 user test password test123 charset utf8mb4 database test [databases.analytics] host 127.0.0.1 port 3307 user analytics password analytics123 charset utf8mb4 database analyticsallow_write false时服务端只放行 SELECT/SHOW/DESCRIBE/EXPLAIN写操作直接拒绝。生产环境建议保持 false需要写时单独开一个受限账号。3.2 MCP 服务脚本安装依赖pip install fastmcp pymysql pydantic tomliPython 3.11 以下用tomli读 TOML3.11 可以直接import tomllib。脚本如下# mysql_mcp_server.py import json import tomllib from pathlib import Path from typing import Any import pymysql from fastmcp import FastMCP from pydantic import BaseModel CONFIG_PATH Path(__file__).parent / config.toml with CONFIG_PATH.open(rb) as f: CONFIG tomllib.load(f) DATABASES: dict[str, dict[str, Any]] CONFIG.get(databases, {}) SECURITY CONFIG.get(security, {}) DEFAULT_LIMIT SECURITY.get(default_limit, 10) MAX_ROWS SECURITY.get(max_rows, 500) ALLOW_WRITE SECURITY.get(allow_write, False) mcp FastMCP(CONFIG[mcp][name]) class SQLRequest(BaseModel): database: str sql: str def _is_query(sql: str) - bool: head sql.strip().upper() return head.startswith((SELECT, SHOW, DESCRIBE, EXPLAIN)) def _apply_limit(sql: str) - str: if _is_query(sql) and LIMIT not in sql.upper(): return f{sql.rstrip(;)} LIMIT {DEFAULT_LIMIT} return sql mcp.tool def execute_sql(request: dict | str | SQLRequest) - dict: 在指定数据库中执行 SQL 并返回结果。 Args: request: {database: test, sql: SELECT * FROM orders} Returns: {success: bool, data: [...], affected_rows: int, error: str} try: if isinstance(request, str): request json.loads(request) if isinstance(request, dict) and request in request: request json.loads(request[request]) if isinstance(request[request], str) else request[request] req request if isinstance(request, SQLRequest) else SQLRequest(**request) except Exception as e: return {success: False, error: f请求解析失败: {e}} if req.database not in DATABASES: return {success: False, error: f数据库 {req.database} 未配置} sql _apply_limit(req.sql.strip()) if not sql: return {success: False, error: SQL 不能为空} if not ALLOW_WRITE and not _is_query(sql): return {success: False, error: 当前配置禁止写操作} cfg dict(DATABASES[req.database]) cfg.setdefault(database, req.database) conn None try: conn pymysql.connect(cursorclasspymysql.cursors.DictCursor, **cfg) with conn.cursor() as cur: cur.execute(sql) if _is_query(sql): rows cur.fetchall() return {success: True, data: rows[:MAX_ROWS], row_count: len(rows)} conn.commit() return {success: True, affected_rows: cur.rowcount} except pymysql.Error as e: return {success: False, error: f数据库错误: {e}} finally: if conn: conn.close() if __name__ __main__: transport CONFIG[mcp].get(transport, stdio) if transport stdio: mcp.run(transportstdio) else: mcp.run( transportstreamable-http, hostCONFIG[mcp][host], portCONFIG[mcp][port], pathCONFIG[mcp][path], )关键点_apply_limit给裸 SELECT 自动补 LIMIT避免 AI 一次拉全表_is_query判断语句类型写操作在allow_writefalse时被拦MAX_ROWS做二次截断防止 LIMIT 写很大。3.3 注册到 AI 客户端以 JSON 配置方式为例在客户端的 MCP 配置里加{ mcpServers: { mysql_tool: { command: python, args: [D:\\work\\mysql_mcp_server.py], env: {} } } }Windows 下command用py或python都行路径用绝对路径。配置完重启客户端工具列表里应该能看到execute_sql。4. 验证请求跑一次完整的 SQL 查询4.1 准备测试表在test库里建一张表并插几条数据CREATE TABLE orders ( id INT PRIMARY KEY AUTO_INCREMENT, user_name VARCHAR(64), amount DECIMAL(10,2), created_at DATETIME DEFAULT CURRENT_TIMESTAMP ); INSERT INTO orders (user_name, amount) VALUES (alice, 120.50), (bob, 88.00), (carol, 256.30);4.2 在 AI 客户端里发起查询在接入了 MCP 的客户端里输入帮我查一下 test 库里 orders 表的前 5 条数据模型会调用execute_sql参数大致是{database: test, sql: SELECT * FROM orders}。服务端自动补 LIMIT 10返回{ success: true, data: [ {id: 1, user_name: alice, amount: 120.50, created_at: ...}, {id: 2, user_name: bob, amount: 88.00, created_at: ...}, {id: 3, user_name: carol, amount: 256.30, created_at: ...} ], row_count: 3 }4.3 直接调服务验证不经过 AI想单独测服务是否正常可以用 stdio 方式手动发一条 JSON-RPC。更简单的是临时把 transport 改成streamable-http启动后用 curl 测python mysql_mcp_server.py # 另开终端 curl -X POST http://127.0.0.1:8000/sql-service \ -H Content-Type: application/json \ -d {jsonrpc:2.0,id:1,method:tools/call,params:{name:execute_sql,arguments:{request:{database:test,sql:SELECT * FROM orders}}}}返回里能看到data数组就说明服务通了。这一步过了再回到 AI 客户端里用自然语言触发链路就完整了。5. 本篇常见错排查报错ModuleNotFoundError: No module named fastmcp客户端启动 MCP 服务时用的 Python 解释器和你装包的未必是同一个。在配置里把command写成解释器绝对路径比如C:\\Python311\\python.exe再用这个解释器pip install。报错Database test not configuredconfig.toml里[databases.test]的段名要和请求里的database字段完全一致大小写敏感。检查 TOML 有没有写错层级。AI 说「工具调用失败」但服务日志没输出stdio 模式下服务日志走 stderr客户端可能吞掉了。临时改成streamable-http看 HTTP 返回或者把异常写进文件。SELECT 返回空但表里明明有数据_apply_limit补的 LIMIT 位置不对或者 SQL 末尾有分号导致拼接成...; LIMIT 10。脚本里已经用rstrip(;)处理如果你改过逻辑要留意。写操作被拒绝allow_write false是默认行为。确认要写时改成 true并确保数据库账号有对应权限别用 root 跑。中文乱码连接配置里charset utf8mb4必须带上建表也用 utf8mb4。6. 把 MCP 服务接进你的 AI 工具链服务跑通后接入方式按你的使用场景分流只是想让 AI 帮你查数据、做分析在模型对话里直接问MCP 工具会自动被调用。模型对话入口https://taotoken.net/model-chat?utm_sourcetaotoken_aicg_blog_endutm_contentmodel_chatutm_campaignrewrite长期在编辑器里写代码、让 AI 反复查库辅助开发用 Coding Plan 更划算。入口https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding_planutm_campaignrewrite需要管理多个 Key、看调用量控制台https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewrite接入配置遇到问题API Keys 页和接入文档https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi_keysutm_campaignrewrite 、https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite一个实用技巧把config.toml里的allow_write默认关掉需要写时临时开改完立刻关回去。MCP 服务本身不替代数据库权限体系账号该给的最小权限还是要给别让 AI 拿着高权限账号裸奔。