1. 为什么要把 MySQL 封装成 MCP SSE 服务如果你正在用 dify 搭智能体大概率遇到过这个尴尬模型能聊天、能写代码但一问「我们库里 2023 级计算机 1 班有多少人」就抓瞎。原因很简单dify 本身不会连你的 MySQL它需要一个标准化的工具入口。MCPModel Context Protocol就是干这个的它把数据库能力包装成模型可调用的工具而 SSEServer-Sent Events传输则让 dify 这类平台可以通过一个 HTTP 长连接持续接收工具列表和调用结果。这篇要聊的就是用 FastMCP 把 MySQL 封装成一个 SSE 传输的 MCP 服务然后在 dify 里完成工具调用。适合谁看适合已经会写点 Python、手里有 MySQL、正在折腾 dify 工作流但卡在「怎么让模型查库」这一步的同学。整套链路我实际跑过一遍从服务端启动到 dify 里 Agent 回显查询结果中间踩的坑都会写出来。核心检索词先摆在这MCP、MySQL、SSE、dify、FastMCP。你要做的事分三段——搭服务端、配 dify 插件、写 Agent 指令并验证。下面按可复制的顺序来命令和配置都能直接抄。2. TaoToken 前置准备先把模型通道打通dify 里的 Agent 要能推理、要能决定调哪个工具背后得有稳定的模型服务。我这边习惯先把模型通道准备好再动数据库那摊事不然调试时分不清是模型不响应还是 MCP 没连上。TaoToken 在这里的角色是提供模型调用入口。你可以先去官网看看它支持哪些模型然后到控制台建一个 API Key。整个流程不复杂注册登录后进控制台找到 API Keys 页面新建一个复制出来存好。这个 Key 后面在 dify 的模型供应商配置里要用到。具体入口我列一下方便你直接跳官网了解整体能力https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content控制台建 Keyhttps://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentAPI Keys 管理页https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content接入文档配 dify 模型时对照https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentAPI 基础地址是 https://taotoken.net/api 注意这个不带 UTM 参数配置里填这个就行。如果你后面要跑长期编码或 Agent 任务可以看看 Coding Planhttps://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 。想先在网页里试试模型对话效果用这个https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 。注意模型通道和 MCP 服务是两条独立的链路。模型通道负责「思考」MCP 负责「查库」。两个都通了Agent 才能既想明白又查得到。3. 可复制配置FastMCP 封装 MySQL 的 SSE 服务端这一章是重头戏。服务端骨架我按 FastMCP 的写法给你包含环境变量、SSE 端点、SQL 安全检查和工具注册。你可以直接建目录、贴代码、改配置。3.1 目录结构与依赖先在本地建一个工程目录比如D:\AITools\mysql-mcp-server-sse结构大概这样mysql-mcp-server-sse/ ├── src/ │ └── server.py ├── .env ├── .env.example └── requirements.txtrequirements.txt内容mcp mysql-connector-python python-dotenv装依赖pip install -r requirements.txt3.2 环境变量配置复制.env.example为.env填上你的 MySQL 信息。关键参数如下参数说明示例HOST服务监听地址0.0.0.0PORT服务监听端口3000MYSQL_HOSTMySQL 地址127.0.0.1MYSQL_PORTMySQL 端口3306MYSQL_USER用户名rootMYSQL_PASSWORD密码your_passwordMYSQL_DATABASE数据库名school_dbENV_TYPE环境类型developmentALLOWED_RISK_LEVELS允许风险等级LOW,MEDIUMENABLE_QUERY_CHECK是否启用 SQL 检查trueALLOW_SENSITIVE_INFO是否允许敏感信息false.env示例HOST0.0.0.0 PORT3000 MYSQL_HOST127.0.0.1 MYSQL_PORT3306 MYSQL_USERroot MYSQL_PASSWORDyour_password MYSQL_DATABASEschool_db ENV_TYPEdevelopment ALLOWED_RISK_LEVELSLOW,MEDIUM ENABLE_QUERY_CHECKtrue ALLOW_SENSITIVE_INFOfalse注意HOST填0.0.0.0是为了让 dify 所在机器能访问到。如果你只在同一台机器上跑127.0.0.1也行但 dify 如果是容器部署就得用宿主机的实际 IP。3.3 服务端骨架代码src/server.py的核心逻辑我拆成几块讲。先是初始化和 SSE 启动import os import re import mysql.connector from dotenv import load_dotenv from mcp.server.fastmcp import FastMCP load_dotenv() HOST os.getenv(HOST, 127.0.0.1) PORT int(os.getenv(PORT, 3000)) ENV_TYPE os.getenv(ENV_TYPE, development) ENABLE_QUERY_CHECK os.getenv(ENABLE_QUERY_CHECK, true).lower() true ALLOWED_RISK_LEVELS os.getenv(ALLOWED_RISK_LEVELS, LOW).split(,) mcp FastMCP(mysql_mcp_server_pro) def get_conn(): return mysql.connector.connect( hostos.getenv(MYSQL_HOST), portint(os.getenv(MYSQL_PORT, 3306)), useros.getenv(MYSQL_USER), passwordos.getenv(MYSQL_PASSWORD), databaseos.getenv(MYSQL_DATABASE), )然后是风险等级判断和 SQL 检查。这块是防止模型乱删库的关键RISK_PATTERNS [ (r^\s*(SELECT|SHOW|DESCRIBE|DESC|EXPLAIN), LOW), (r^\s*INSERT, MEDIUM), (r^\s*UPDATE\s.*WHERE, MEDIUM), (r^\s*UPDATE\s(?!.*WHERE), HIGH), (r^\s*(CREATE|ALTER), HIGH), (r^\s*(DROP|TRUNCATE), CRITICAL), (r^\s*DELETE\s(?!.*WHERE), CRITICAL), ] def check_sql(sql: str): if not ENABLE_QUERY_CHECK: return True, check disabled for pattern, level in RISK_PATTERNS: if re.match(pattern, sql, re.IGNORECASE | re.DOTALL): if level not in ALLOWED_RISK_LEVELS: return False, frisk level {level} not allowed return True, level return True, LOW接着注册工具。基础查询工具mysql_query是核心mcp.tool() def mysql_query(sql: str) - str: 执行 SQL 查询支持参数化。 ok, msg check_sql(sql) if not ok: return fSQL 被拦截{msg} conn get_conn() cursor conn.cursor(dictionaryTrue) try: cursor.execute(sql) if sql.strip().upper().startswith((SELECT, SHOW, DESC, EXPLAIN)): rows cursor.fetchall() return str(rows[:100]) else: conn.commit() return f影响行数{cursor.rowcount} except Exception as e: conn.rollback() return f执行错误{e} finally: cursor.close() conn.close()元数据工具也顺手加上dify 里模型经常需要先看表结构mcp.tool() def mysql_show_tables(pattern: str %) - str: 获取数据库中的表列表。 conn get_conn() cursor conn.cursor() cursor.execute(SHOW TABLES LIKE %s, (pattern,)) tables [row[0] for row in cursor.fetchall()] cursor.close() conn.close() return str(tables) mcp.tool() def mysql_show_columns(table: str) - str: 获取表的列信息。 conn get_conn() cursor conn.cursor(dictionaryTrue) cursor.execute(fSHOW COLUMNS FROM {table}) cols cursor.fetchall() cursor.close() conn.close() return str(cols)最后是 SSE 启动入口if __name__ __main__: mcp.run(transportsse, hostHOST, portPORT)3.4 启动服务在工程根目录执行python -m src.server启动成功后终端会打印类似Uvicorn running on http://0.0.0.0:3000的信息SSE 端点就是http://你的IP:3000/sse。我这边实测的地址是http://10.0.15.153:3000/sse你换成自己的实际 IP。4. 验证请求连通性检查与 dify 工具接入服务端起来了先别急着进 dify用 curl 确认 SSE 端点活着。4.1 连通性验证curl -N http://10.0.15.153:3000/sse如果看到持续输出的事件流类似event: endpoint加data: ...说明 SSE 通道正常。按 CtrlC 退出即可。这一步很关键很多人 dify 里连不上其实是服务端根本没监听对地址。4.2 dify 安装 MCP SSE 插件进 dify 插件市场搜索 MCP SSE安装。然后在「工具」里找到 MCP SSE点「去授权」填入服务地址{mysql_mcp_server_pro:{url:http://10.0.15.153:3000/sse}}保存后dify 会去拉取工具列表。如果配置正确你能看到mysql_query、mysql_show_tables等工具出现在列表里。4.3 工作流里配置 Agent 策略新建一个工作流命名Dify_Mysql集成。在 Agent 节点里策略选ReActSupport MCP Tools。然后添加 MCP 服务配置把超时调大一点避免查大表时断连{mysql_mcp_server_pro:{url:http://10.0.15.153:3000/sse,timeout:600,sse_read_timeout:3000}}指令部分我写了一段中文提示把表结构说明塞进去让模型知道有哪些表、字段叫什么使用中文回复。 当用户提问中涉及学生、教师、成绩、班级、课程等实体时需要使用 MySQL MCP 进行数据查询和操作。 数据库包含以下表 - teachers教师表字段有 id, name, gender, subject, title, phone, office, wechat, isHeadTeacher - classes班级表字段有 id, className, grade, headTeacherId, classroom, studentCount, remark - courses课程表字段有 id, courseName, credit, teacherId, semester, type, prerequisite - students学生表字段有 id, name, gender, birthDate, enrollmentDate, classId, phone, email, emergencyContact, address, height, weight, healthStatus - scores成绩表字段有 id, studentId, courseId, score, examDate, usualScore, finalScore 表关系students.classId 关联 classes.idcourses.teacherId 关联 teachers.idscores.studentId 关联 students.idscores.courseId 关联 courses.idclasses.headTeacherId 关联 teachers.id。4.4 调用回显测试在工作流里输入测试问题比如「2023 级计算机 1 班有多少学生」。Agent 的推理过程应该是先调mysql_show_tables确认表存在再调mysql_query执行SELECT COUNT(*) FROM students WHERE classId 202301最后把结果用中文回显。我实测下来第一次调用可能会因为模型不确定表名而多绕一步但把表结构写进指令后基本一次命中。回显结果类似「2023 级计算机 1 班共有 35 名学生」。5. 本篇常见错排查跑这条链路报错基本集中在这几类我按出现频率排一下。连接被拒绝Connection refuseddify 里配的 URL 用了127.0.0.1但 dify 是容器部署容器里的127.0.0.1指向容器自己。改成宿主机实际 IP比如10.0.15.153。同时确认服务端HOST是0.0.0.0。SSE 连上但工具列表为空多半是 FastMCP 版本和 dify 插件不匹配。确认mcp包是最新版pip install -U mcp。另外检查服务端启动日志有没有报工具注册失败。SQL 被拦截ALLOWED_RISK_LEVELS只配了LOW但模型发了INSERT。开发环境可以放宽到LOW,MEDIUM生产环境建议保持LOW让模型只读。查询超时大表SELECT *没加LIMIT。在指令里明确要求模型加LIMIT或者服务端在mysql_query里对SELECT强制截断返回条数。中文乱码MySQL 连接没指定字符集。在get_conn里加charsetutf8mb4。dify 授权失败URL 末尾多了斜杠或者 JSON 格式不对。严格按{mysql_mcp_server_pro:{url:http://IP:3000/sse}}写别加多余空格。提示排查时先 curl 服务端再查 dify 插件授权最后看 Agent 推理日志。一层层来别一上来就怀疑模型。6. 后续怎么用从能查到好用链路跑通只是起点。实际用起来有几个地方值得再打磨。一是把表结构说明做成 dify 的知识库而不是硬塞在指令里这样表多了也不乱。二是给mysql_query加个结果条数上限防止模型手滑SELECT *把上下文撑爆。三是生产环境务必把ENV_TYPE设成productionALLOWED_RISK_LEVELS只留LOW让 MCP 服务只读不写。如果你后面要跑更重的编码或 Agent 任务模型通道那边可以看看 Coding Plan长期用更划算https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 。接入细节对照文档https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 。Key 在控制台建https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 。最后留个我踩过的坑FastMCP 的 SSE 端点在部分版本里路径是/sse但如果你改了HOST或PORTdify 里的 URL 一定要同步改别只改服务端忘了改 dify。这个错我犯了两次每次都要重新授权插件挺费时间的。
