ARTICLE DETAIL

资讯详情

深耕网站建设与运营推广的一线实战洞察。

FastMCP实践开发应用:用Python+SSE+MySQL搭建TaoToken统一Key接入的MCP服务

FastMCP实践开发应用:用Python+SSE+MySQL搭建TaoToken统一Key接入的MCP服务 1. 从一次真实踩坑说起为什么我要把 MySQL 工具塞进 FastMCP上周帮朋友调一个 Cline 里的数据库助手他本地跑着 MySQL 8想让 AI 直接查表结构和慢查询锁等待结果每次都要手动贴 SQL、复制结果来回折腾十几轮。我第一反应是写个 Python 脚本包一层但脚本没法被 AI 客户端直接调用于是转向 MCPModel Context Protocol——它本质上是给 AI 助手定义一套「可调用的工具清单」客户端按协议去请求服务端返回结构化结果。FastMCP 是 Python 下构建 MCP 服务端最省事的框架几行装饰器就能把普通函数暴露成工具。但官方示例默认走 stdio本地调试还行一旦要让 Cline、CC Switch 这类客户端远程连过来就得切到 SSEServer-Sent Events传输。再叠加上 MySQL 作为工具数据源、TaoToken 统一 Key 做模型通道整条链路才算完整。这篇就按我实际跑通的顺序写先讲清楚 FastMCP SSE MySQL 这套组合解决什么问题、适合谁再给可复制的 config.toml 与 settings.json 骨架然后一步步配 SSE 端点、写 MySQL 工具、用 curl 验证最后把几个高频报错摊开讲。目标很明确——你照着敲完能拿到一个可被 Cline 调用的 MCP 服务并且模型请求统一走 TaoToken 的 Key。2. TaoToken 前置统一 Key 与 API 通道怎么接MCP 服务本身不负责调模型它只负责「暴露工具」。真正让 AI 客户端能对话、能推理的是客户端背后的模型通道。我这边习惯把模型请求统一收敛到 TaoToken好处是 Key 只维护一份换客户端不用重新配。你需要先拿到一个 API Key。登录官网后进控制台在 API Keys 页面创建一个复制出来形如sk-xxxxxxxx。这个 Key 后面会写进客户端的 settings.json而不是写进 MCP 服务代码里——MCP 服务只管数据库不碰模型鉴权职责分开后面排错才不打架。TaoToken 的 API 基地址是https://taotoken.net/api兼容 OpenAI 风格的/v1/chat/completions。如果你用的是 Claude Code 这类走 Anthropic 协议的客户端也有对应的接入文档可查。我实测下来把 base_url 和 key 填对之后Cline 里选模型、发请求都正常不需要额外装什么中间层。注意API Key 属于敏感凭证别写进会提交到 Git 的 config.toml 里。我的做法是 MCP 服务用.env管数据库密码客户端 Key 放本地 settings.json两边物理隔离。3. 可复制配置config.toml 与 settings.json 骨架先把两个配置文件摆出来后面所有步骤都围绕它们展开。3.1 MCP 服务侧.env 与依赖数据库连接信息走.env避免硬编码# .env MYSQL_HOST127.0.0.1 MYSQL_PORT3306 MYSQL_USERroot MYSQL_PASSWORDyour_password MYSQL_DATABASEtest依赖安装一条命令pip install fastmcp mysql-connector-python python-dotenv3.2 客户端侧settings.json 骨架以 Cline 为例MCP 服务配置通常长这样注意transport选sseurl指向你本机或内网的 SSE 端点{ mcpServers: { mysql-tools: { transport: sse, url: http://127.0.0.1:9000/sse, disabled: false, autoApprove: [get_table_name, get_table_desc] } } }autoApprove里放的是只读类工具查询表名、表结构这种不会改数据自动批准能省掉每次弹窗确认。execute_sql这种能写库的我故意不放进 autoApprove保留人工确认。3.3 模型通道config.toml 片段如果你用的客户端支持 TOML 配置比如某些 CLI 工具模型通道部分这样写[model] provider openai-compatible base_url https://taotoken.net/api api_key sk-你的Key model claude-sonnet-4-5base_url 末尾不要多加/v1客户端一般会自己拼。填错这个是最常见的 404 来源后面排错章节会细说。4. 写 SSE 服务端把 MySQL 工具挂上去4.1 最小可跑的 SSE 骨架FastMCP 默认mcp.run()走 stdio要切 SSE 只需传transportsse并指定端口from fastmcp import FastMCP mcp FastMCP(operateMysql, port9000) mcp.tool() def ping() - str: 连通性测试 return pong if __name__ __main__: mcp.run(transportsse)跑起来后访问http://127.0.0.1:9000/sse浏览器会挂住不断接收事件流这是正常的——SSE 是长连接不是普通网页。4.2 接入 MySQL三个实用工具下面这段是我实际在用的比手写原生 MCP 协议少了近百行。核心是execute_sql做通用查询再包两个语义化工具方便 AI 理解from fastmcp import FastMCP from mysql.connector import connect, Error from dotenv import load_dotenv import os mcp FastMCP(operateMysql, port9000) def get_db_config(): load_dotenv() config { host: os.getenv(MYSQL_HOST, localhost), port: int(os.getenv(MYSQL_PORT, 3306)), user: os.getenv(MYSQL_USER), password: os.getenv(MYSQL_PASSWORD), database: os.getenv(MYSQL_DATABASE), } if not all([config[user], config[password], config[database]]): raise ValueError(缺少必需的数据库配置) return config mcp.tool() def execute_sql(query: str) - list: 执行SQL查询语句多条以分号分隔 config get_db_config() try: with connect(**config) as conn: with conn.cursor() as cursor: statements [s.strip() for s in query.split(;) if s.strip()] results [] for statement in statements: try: cursor.execute(statement) if cursor.description: columns [d[0] for d in cursor.description] rows cursor.fetchall() formatted [ ,.join(NULL if v is None else str(v) for v in row) for row in rows ] results.append(\n.join([,.join(columns)] formatted)) else: conn.commit() results.append(f执行成功影响行数: {cursor.rowcount}) except Error as stmt_error: results.append(f语句 {statement} 出错: {stmt_error}) return [\n---\n.join(results)] except Error as e: return [f执行查询时出错: {e}] mcp.tool() def get_table_name(text: str) - list: 根据表的中文注释搜索表名 config get_db_config() sql ( SELECT TABLE_SCHEMA, TABLE_NAME, TABLE_COMMENT FROM information_schema.TABLES fWHERE TABLE_SCHEMA {config[database]} fAND TABLE_COMMENT LIKE %{text}%; ) return execute_sql(sql) mcp.tool() def get_table_desc(text: str) - list: 获取指定表的字段结构多个表名逗号分隔 config get_db_config() names [n.strip() for n in text.split(,)] condition ,.join(names) sql ( SELECT TABLE_NAME, COLUMN_NAME, COLUMN_COMMENT FROM information_schema.COLUMNS fWHERE TABLE_SCHEMA {config[database]} fAND TABLE_NAME IN ({condition}) ORDER BY TABLE_NAME, ORDINAL_POSITION; ) return execute_sql(sql) if __name__ __main__: mcp.run(transportsse)启动后终端会打印Uvicorn running on http://0.0.0.0:9000说明 SSE 端点已就绪。4.3 为什么工具描述要写中文mcp.tool()下面那行 docstring 不是装饰客户端会把整段描述喂给模型模型据此判断「什么时候该调这个工具」。我一开始写英文模型在中文对话里经常忽略改成中文后问「帮我找订单相关的表」它会主动调get_table_name。描述里把参数格式也写清楚比如「多个表名逗号分隔」能显著减少模型传错参数的概率。5. 验证请求curl 与客户端双通道确认5.1 用 curl 确认 SSE 端点活着先确认服务端在监听curl -N http://127.0.0.1:9000/sse-N关闭缓冲你会看到类似event: endpoint加一行data: /messages/?session_idxxx的输出。这说明 SSE 通道建立成功客户端后续就是往这个 session 地址发 JSON-RPC 请求。5.2 用 curl 走一次工具调用拿到 session_id 后可以手动发一条工具调用请求验证curl -X POST http://127.0.0.1:9000/messages/?session_id你的session_id \ -H Content-Type: application/json \ -d { jsonrpc: 2.0, id: 1, method: tools/call, params: { name: get_table_name, arguments: {text: 订单} } }返回里如果出现订单表相关的表名和注释说明 MySQL 数据源通了。这一步能过客户端那边基本不会出问题。5.3 在 Cline 里跑通完整链路把第 3.2 节的 settings.json 填好重启 Cline。在对话里输入「列出数据库里所有和订单相关的表」正常表现是Cline 先调get_table_name拿到结果后再组织语言回复。如果它直接说「我无法访问数据库」八成是 MCP 服务没连上回去看 Cline 的 MCP 面板有没有绿色状态点。模型通道那边确认 base_url 是https://taotoken.net/api、Key 是控制台新建的那串。我实测下来Cline 里选好模型后发一条「你好」能正常返回就说明 TaoToken 通道没问题剩下的都是 MCP 侧的事。6. 本篇常见错排查6.1 报错Connection refused或客户端连不上先看服务端是不是真的在跑。mcp.run(transportsse)默认绑0.0.0.0:9000如果你在 Docker 里跑端口没映射出来宿主机自然连不上。另外 Windows 防火墙有时会拦 9000 端口临时放行一下测试。还有一种情况客户端配置里写了http://localhost:9000/sse但服务端只监听了 IPv6 的 localhost。改成127.0.0.1通常能解决。6.2 报错Access denied for user或Unknown database.env里的密码如果含、#这类字符dotenv 解析时可能被截断。我的做法是密码用引号包起来MYSQL_PASSWORDabcd1234。另外确认MYSQL_DATABASE指向的库真实存在information_schema查询虽然不依赖具体库但连接时必须指定一个存在的库名。6.3 模型返回 404 或model not found九成是 base_url 拼错了。TaoToken 的基地址是https://taotoken.net/api有些客户端会自动补/v1有些不会。如果你填成https://taotoken.net/api/v1而客户端又补一次就变成/api/v1/v1/chat/completions直接 404。先按文档给的基地址填报错再逐层排查。6.4 工具调用返回空或模型不调工具检查 docstring 是不是被删了。FastMCP 靠函数签名和 docstring 生成工具的 JSON Schema没有描述模型不知道这个工具干嘛用。另外autoApprove里没列的工具客户端会弹窗等确认如果你没点请求就挂在那里看起来像「没反应」。6.5 SSE 连接频繁断开SSE 是长连接中间如果有反向代理默认超时可能只有几十秒。本地直连一般没事如果套了 Nginx记得把proxy_read_timeout调大并关闭缓冲proxy_buffering off;。7. 把 Key 和工具链固定下来跑通之后我建议做两件事。一是把 MCP 服务用 systemd 或 supervisor 托管别每次手动python server.py二是把 TaoToken 的 Key 和 MCP 配置分开管理Key 放客户端、数据库凭证放服务端.env这样换客户端或换数据库时互不影响。如果你后面要长期在编码场景里用这套组合可以看看 Coding Plan 这类按周期计费的方案比每次单独充 Key 省心。需要新建或轮换 Key 的时候直接进控制台 API Keys 页面操作接入细节拿不准就翻接入文档里面把 base_url、鉴权头、常见客户端配置都列全了。工具链固定下来之后AI 查库、看表结构、排查锁等待这些事基本就是一句话的事。
返回列表