用 Python 手搓 MCP 服务：让 AI 用自然语言直连数据库，TaoToken 统一 Key 接入

1. 为什么我要自己写一个 MCP 数据库服务你可能已经习惯了把表结构复制粘贴给 AI让它帮你写 SQL然后再手动去数据库客户端里执行。这个流程在表少的时候还行一旦库里有几十张表、字段名还都是缩写AI 就开始瞎猜写出来的 SQL 不是字段名对不上就是 JOIN 关系搞错。更麻烦的是每次换一个 AI 工具就要重新配一遍 Key、重新贴一遍表结构通道和凭证散落在各个客户端里管理起来很乱。Model Context ProtocolMCP解决的正是这件事它给 AI 客户端和外部数据源之间定了一套标准协议AI 通过 JSON-RPC 调用你暴露出来的工具而不是靠猜。我这次用 Python 从零搭了一个 MCP 服务把「列出所有表」「查看字段定义」「执行只读查询」这几个能力暴露出去AI 就能用自然语言直接查库了。同时我把模型调用的 Key 统一收敛到 TaoToken 一个入口MCP 服务本身只负责数据库这一侧两边职责分开配置一次就能长期用。这篇适合谁会一点 Python、手上有 MySQL 或 SQLite、想让 AI 工具安全查库的开发者。下面从环境准备讲到配置骨架再到自然语言查询的验证步骤命令都可以直接复制。2. TaoToken 前置把模型 Key 统一收口MCP 服务负责「查库」但 AI 客户端要能理解你的自然语言、决定调用哪个工具背后仍然需要模型。如果每个客户端各配一套 Key就又回到了分散的老问题。我的做法是所有支持自定义 API 地址的客户端统一指向 TaoToken 的 API 入口用同一个 Key。TaoToken 在这里的角色是统一的模型接入层兼容常见的 OpenAI 风格接口你不需要为每个工具单独申请凭证。先到控制台创建一个 API Key控制台https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteAPI Keys 管理https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewrite创建好之后客户端里填的 Base URL 用https://taotoken.net/apiKey 填刚生成的那串。如果你只是想先验证模型通不通可以直接在模型对话页面试一句模型对话https://taotoken.net/model-chat?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewrite注意MCP 服务本身不负责模型调用它只暴露数据库工具。模型 Key 配在 AI 客户端那一侧两边不要混在一起排障时才能快速定位是哪一层的问题。如果你后续要长期跑编码类 Agent反复调用模型可以了解下 Coding Plan额度模型更适合高频场景Coding Planhttps://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewrite3. 可复制配置MCP 服务端骨架与 config.toml3.1 目录结构与依赖先建一个干净的项目目录我习惯这样组织mcp-db-python/ ├── server.py ├── db.py ├── config.toml ├── requirements.txt └── .envrequirements.txt里核心就两个MCP 的 Python SDK 和数据库驱动。mcp1.0.0 pymysql1.1.0 python-dotenv1.0.0安装pip install -r requirements.txt3.2 config.toml 骨架MCP 客户端读取服务的方式通常是在客户端的配置里声明一个 stdio 类型的 server。下面这份config.toml是我实际在用的骨架把命令、参数、环境变量都写清楚客户端启动时会按这个拉起 Python 进程[mcp_servers.db_python] command python args [/absolute/path/to/mcp-db-python/server.py] [mcp_servers.db_python.env] DB_TYPE mysql DB_HOST 127.0.0.1 DB_PORT 3306 DB_USER readonly_user DB_PASS your_password DB_NAME test READ_ONLY true MAX_ROWS 200几个参数值得单独说参数作用建议值DB_TYPE数据库类型mysql / sqliteREAD_ONLY是否强制只读trueMAX_ROWS单次查询返回上限100–500DB_USER数据库账号单独建只读账号注意args里的路径一定写绝对路径。MCP 客户端拉起进程时工作目录不一定是你以为的那个相对路径经常导致「找不到 server.py」。3.3 只读校验与工具注册db.py里最关键的是 SQL 白名单校验只允许 SELECT、SHOW、DESC、EXPLAIN 这类语句其他一律拒绝import re ALLOWED_PREFIX (select, show, desc, describe, explain) def is_read_only(sql: str) - bool: cleaned re.sub(r\s, , sql.strip().lower()) if not cleaned.startswith(ALLOWED_PREFIX): return False forbidden (insert, update, delete, drop, alter, truncate, grant) return not any(word in cleaned for word in forbidden)server.py里用 MCP SDK 注册工具把「列表」「看结构」「查询」三件事暴露出去import asyncio from mcp.server import Server from mcp.server.stdio import stdio_server from mcp.types import Tool, TextContent from db import list_tables, get_table_schema, run_query app Server(db-python) app.list_tools() async def list_tools(): return [ Tool(namelist_tables, description列出数据库中所有表, inputSchema{type: object, properties: {}}), Tool(nameget_table_schema, description查看指定表的字段定义, inputSchema{type: object, properties: {table: {type: string}}, required: [table]}), Tool(namerun_query, description执行只读 SQL 查询, inputSchema{type: object, properties: {sql: {type: string}}, required: [sql]}), ] app.call_tool() async def call_tool(name: str, arguments: dict): if name list_tables: return [TextContent(typetext, textstr(list_tables()))] if name get_table_schema: return [TextContent(typetext, textstr(get_table_schema(arguments[table])))] if name run_query: return [TextContent(typetext, textstr(run_query(arguments[sql])))] raise ValueError(funknown tool: {name}) async def main(): async with stdio_server() as (read, write): await app.run(read, write, app.create_initialization_options()) if __name__ __main__: asyncio.run(main())run_query内部先过is_read_only再执行并限制返回行数def run_query(sql: str): if not is_read_only(sql): return {error: 仅允许只读查询} with get_conn() as conn: with conn.cursor() as cur: cur.execute(sql) rows cur.fetchmany(MAX_ROWS) cols [d[0] for d in cur.description] return [dict(zip(cols, row)) for row in rows]4. 验证请求用自然语言查一次库配置写完后先单独跑一下服务确认能启动python server.py终端没有报错、进程挂起等待输入就说明 stdio 通道正常。接着在 AI 客户端里把上面那份config.toml的 server 配置加进去重启客户端让它加载 MCP 服务。然后直接对 AI 说一句自然语言比如帮我看看 test 库里有哪些表然后告诉我 users 表的字段结构。正常情况下AI 会先调用list_tables再调用get_table_schema把结果整理后回给你。接着再试一句带条件的查询查一下 users 表里最近注册的 10 个用户按创建时间倒序。AI 会生成类似这样的 SQL 并调用run_querySELECT * FROM users ORDER BY created_at DESC LIMIT 10;返回结果会以文本形式回到对话里。如果这一步成功了说明「自然语言 → 工具调用 → SQL → 结果」这条链路已经打通。你可以再故意让它执行一条DELETE观察服务是否返回「仅允许只读查询」以此确认安全校验生效。5. 本篇常见错排查5.1 客户端报 server 启动失败九成是路径问题。检查config.toml里args的server.py是不是绝对路径以及command用的python在当前环境里能不能找到。如果你用的是虚拟环境把command换成虚拟环境里的 python 绝对路径例如/Users/you/venv/bin/python。5.2 连不上数据库先确认数据库账号密码和端口再确认账号有没有对应库的权限。生产环境强烈建议单独建一个只读账号只授予 SELECT 权限这样即使校验逻辑有疏漏也删不掉数据。5.3 查询返回空或字段名对不上多半是表名大小写或库名没选对。MySQL 在部分系统上表名区分大小写get_table_schema返回的字段名以数据库实际为准别用 AI 猜的名字去写 SQL。遇到不确定的表先让它调list_tables。5.4 模型侧报鉴权失败这属于客户端到模型那一层和 MCP 服务无关。检查客户端里填的 Base URL 是不是https://taotoken.net/apiKey 有没有多余空格。想快速确认 Key 是否可用去模型对话页面发一句话即可模型对话https://taotoken.net/model-chat?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewrite5.5 返回行数太多把上下文撑爆把MAX_ROWS调小或者在提示词里要求 AI 先加LIMIT。我一般把上限设在 200够日常排查用了。6. 把 Key 和通道固定下来长期用整套跑通之后你会发现真正省事的地方在于数据库这一侧的能力被固化成了 MCP 工具模型这一侧的凭证被收敛成了一个 Key。以后不管换哪个支持 MCP 的客户端只要把config.toml复制过去、Base URL 和 Key 填同一套就能直接查库不用再重新贴表结构、重新配通道。如果你要接的是编码类 Agent需要频繁调用模型可以看下 Coding Plan 的额度方式接入细节和参数说明都在文档里Coding Planhttps://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewrite接入文档https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteAPI Keyshttps://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewrite最后留一个我踩过的坑别把生产库的写权限账号配进 MCP 服务哪怕校验写得再严账号权限才是最后一道闸。只读账号 只读校验 行数上限这三样配齐再让 AI 碰数据库才踏实。