最近在调 Dify 工作流时我又遇到一个几乎每个项目都跑不掉的需求Agent 要查订单、查库存、查用户信息而这些数据全在 MySQL 里。Dify 本身不做数据存储它的职责是编排 LLM 和工具调用所以真正要把“让大模型安全地查数据库”这件事落地核心思路是把 MySQL 变成 Dify 能调用的一个工具而不是让模型直接去写 SQL、直连数据库。这篇文章把我实际配置的完整过程写一遍包括 MySQL 连接准备、封装查询接口、在 Dify 工作流里接数据库节点、处理 SSL 错误和上下文超长这些问题。适合正在用 Dify 做企业级应用、又不想把数据库直接暴露给大模型的开发者参考。整个方案不挑部署方式Dify 社区版和企业版都能用。1. 整体思路与方案选型1.1 为什么 Dify 不能直接“连”MySQL很多人第一次接触 Dify会觉得它既然能做工作流编排那是不是能像 Navicat 一样直接配置一个数据库连接然后让大模型去查表这个想法很自然但实际走不通。Dify 的工作流节点本质上是“调工具、传参数、拿结果”它没有一个通用的 MySQL 客户端协议实现也不负责管理数据库连接生命周期。更重要的是如果把 MySQL 连接信息直接写进工作流LLM 生成的 SQL 一旦出现语法错误、死循环查询、超大批量扫描数据库很容易被打挂。企业环境里还有审计和权限要求数据库账号不能交给一个无法控制的模型去自由发挥。所以 Dify 访问 MySQL 的正确姿势一定是通过中间层要么是现成的插件工具要么是自己封装的查询 API。1.2 三条主流接入路线我在不同项目里分别试过三种方式各有适用场景插件工具方式Dify 1.x 的插件体系里可以安装社区或官方维护的 MySQL 连接器安装后在工具列表里直接填主机、端口、账号、密码然后选择内置动作比如执行 SQL、查表结构。优点是配置快适合内部工具、调试环境、查询逻辑固定的场景。缺点是插件能力受限于它暴露的参数比如动态 SQL、分页、结果集大小控制不一定能满足复杂需求。自建查询 API 方式自己用 FastAPI 或 Flask 写一个只读查询接口把 MySQL 访问封装在 API 里然后在 Dify 自定义工具中导入 OpenAPI Schema。这个方式最灵活可以在 API 层做权限校验、SQL 白名单、超时控制、结果集截断、审计日志甚至加一层 Redis 缓存。适合生产环境、多业务线复用、以及需要把数据库细节完全隐藏的场景。代码节点直连方式在 Dify 工作流里用“代码”节点通过 Python 的 pymysql 直接连 MySQL。这种方式我只用来做过本地功能验证不推荐上生产。原因很直接连接凭证会以代码变量形式出现在工作流里而且每次执行都会新建连接性能差也没有权限隔离。三种方式对比下来插件工具适合“快速打通”代码节点适合“临时验证”自建 API 才是长期可维护的选择。1.3 我为什么选“自建只读 API 自定义工具”我最终采用自建 API主要不是因为插件不好用而是因为一个现实问题业务方经常提出“帮我查一下某个表但不要给模型写权限”“查出来的订单金额字段要脱敏”“同一个接口要同时支持 MySQL 和 PostgreSQL”。如果直接依赖某个现成插件这些定制需求会变成插件参数里的一个正则表达式或一个备注字段根本没法维护。自建 API 之后Dify 那边只多了一个 HTTP 工具数据库地址、账号、密码、表结构全部留在 API 服务内部。Dify 工作流里能看到的只是一个POST /query接口和几个参数。数据链路变成用户问题 → Dify 工作流 → 查询 API → MySQL每一层都可以加控制点出了问题也容易定位。这个模式我在两个生产项目里都验证过稳。2. 环境准备与前置条件2.1 MySQL 侧先建一个只读账号无论你选哪种接入方式第一步都是给 Dify 建一个专用数据库账号权限控制在只读范围。原则是最小权限能只给某几个表就不要给整个库能只给 SELECT就不要给 INSERT、UPDATE、DELETE。我用 MySQL 8.0 举一个例子创建dify_readonly用户并授权CREATE USER dify_readonly% IDENTIFIED BY 这里填强密码; GRANT SELECT ON mydb.orders TO dify_readonly%; GRANT SELECT ON mydb.users TO dify_readonly%; GRANT SELECT ON mydb.order_items TO dify_readonly%;这里有两个细节容易被忽略。第一%表示允许从任意主机登录如果 Dify 服务器 IP 固定建议写成192.168.1.100避免账号被外部扫描到。第二MySQL 8.0 默认使用caching_sha2_password认证Python 的 pymysql 需要安装cryptography包才能正常握手否则连接时会直接报错。如果业务允许还可以给这条连接设置一个最大执行时间防止慢查询把数据库线程占满。MySQL 没有直接的会话级max_execution_time对 SELECT 全局限制但可以在 API 层通过SET SESSION MAX_EXECUTION_TIME 3000来约束下面会写具体实现。2.2 确认网络连通与 SSL 配置Dify 和 MySQL 之间首先要保证网络通。尤其是用 Docker 部署 Dify 时Dify 容器要能访问到宿主机或云数据库的地址。你可以先不进 Dify直接在部署 Dify 的服务器上执行telnet 192.168.1.10 3306如果端口不通先去查安全组和防火墙。端口通了再看 SSL。这里就是很多人会踩坑的地方云厂商的 MySQL 默认开启 SSL 或者require_secure_transport而你在 Dify 插件里只填了普通连接串结果就是连接被服务端拒绝或者 Dify 报“凭据验证失败”。我的处理方式是内网环境如果对传输加密没有强制要求就在 MySQL 服务端临时关闭require_secure_transport配置或者直接确认当前账号不需要 SSL如果是公网或等保环境强制要求 SSL那就把 CA 证书下载下来在 API 层配置ssl_ca参数不要试图在 Dify 插件的连接串里塞证书路径Dify 插件不一定支持这么细。SSL 问题我没少折腾后面单独写一节排查思路。2.3 准备 Dify 自定义工具入口Dify 里的入口位置是“工具 → 自定义 → 创建自定义工具”。进入后需要填一个 OpenAPI Schema 和服务器 URL。如果你的方案是自建 API这一步先不急着填等 API 写好、能本地跑通之后把/openapi.json的内容粘贴进去就行。如果你用的是现成 MySQL 插件入口不太一样要在“插件”页面安装然后在工具的凭据配置里填 MySQL 连接信息。注意填完凭据之后 Dify 会做一次连接校验任何一点配置不对都会弹出an error occurred during credentials validation。这个报错信息非常不具体排查顺序一般是网络通不通 → 端口对不对 → 账号密码对不对 → 账号授权是否生效 → 认证插件是否被客户端支持。下面第五章会展开讲。3. 用 FastAPI 封装一个 MySQL 查询接口3.1 项目结构和依赖我习惯用一个非常小的 FastAPI 服务来做这件事不需要引入复杂的 ORM因为查询本身是动态拼 SQLORM 反而碍事。项目结构大概长这样mysql-query-api/ ├── main.py ├── requirements.txt └── .envrequirements.txt内容如下fastapi uvicorn pymysql cryptography pydantic python-dotenv其中cryptography是为了支持 MySQL 8 的caching_sha2_password没有它光用 pymysql 连会报RuntimeError: cryptography package is required for sha256_password or caching_sha2_password auth methods。我第一次踩到这个坑时还以为是密码错了后来才发现少装了一个包。3.2 只读查询接口的代码实现接口设计上我只对外开放一个POST /query方法请求体包含sql和可选的params。在 API 内部做四件事检查 SQL 是否是只读查询、参数化执行、限制返回行数、关闭连接。import os import pymysql from fastapi import FastAPI, HTTPException from pydantic import BaseModel, Field from dotenv import load_dotenv load_dotenv() app FastAPI(titleMySQL Query API) DB_CONFIG { host: os.getenv(MYSQL_HOST, 127.0.0.1), port: int(os.getenv(MYSQL_PORT, 3306)), user: os.getenv(MYSQL_USER), password: os.getenv(MYSQL_PASSWORD), database: os.getenv(MYSQL_DB), charset: utf8mb4, cursorclass: pymysql.cursors.DictCursor, } class QueryRequest(BaseModel): sql: str Field(..., description只读 SQL必须以 SELECT 开头) params: dict | None Field(defaultNone, descriptionSQL 参数用于参数化查询) app.post(/query) def query(req: QueryRequest): sql req.sql.strip().rstrip(;).strip() if not sql.lower().startswith(select): raise HTTPException(status_code400, detail只允许 SELECT 查询) if ; in sql: raise HTTPException(status_code400, detail不允许多语句查询) conn pymysql.connect(**DB_CONFIG) try: with conn.cursor() as cursor: cursor.execute(SET SESSION MAX_EXECUTION_TIME 3000) cursor.execute(sql, req.params or ()) rows cursor.fetchmany(50) return {code: 0, rows: rows, count: len(rows)} except Exception as e: raise HTTPException(status_code500, detailstr(e)) finally: conn.close()这里有几个关键点。第一SET SESSION MAX_EXECUTION_TIME 3000的意思是这条会话里超过 3 秒的 SELECT 会被 MySQL 直接终止避免工作流里出现一条慢查询卡住整个 Agent。第二fetchmany(50)是应对上下文超长的重要手段不管查出来多少条先只取 50 行后面的要么分页要么让用户加条件。第三参数化查询能挡住大部分 SQL 注入配合 “只允许 SELECT” 和 “不允许分号” 两个判断安全性已经比让 LLM 自由拼 SQL 好太多。3.3 生成 OpenAPI Schema 并导入 DifyFastAPI 的一个好处是会自动生成/openapi.json。Dify 自定义工具支持直接粘贴这个 JSON它会自动解析出接口的路径、参数、请求体结构。如果不想手动粘贴也可以直接填服务器地址http://你的API地址然后 Dify 尝试拉取 Schema。生产环境我更推荐把 Schema 下载下来、贴身维护一份避免 API 升级时 Dify 侧自动同步出问题。导入 OpenAPI Schema 后Dify 工具页面上会出现一个query动作参数对应到QueryRequest的两个字段sql和params。我一般不建议把sql直接暴露给 LLM而是把参数做得更语义化比如接口改成SELECT * FROM orders WHERE id %(order_id)s让 LLM 只填order_id。这一步决定了你的数据库查询到底是“可控工具”还是“裸奔风险”强烈建议按语义化参数来设计。3.4 在 Dify 中完成自定义工具配置打开 Dify“工具 → 自定义 → 创建自定义工具”粘贴 Schema 后服务器 URL 填 API 的根地址比如http://192.168.1.100:8000。如果 API 需要鉴权在“API 密钥”里填一个自定义 Header比如X-API-Key我在 API 端也加了中间件校验这样可以防止内网其他服务随便调用。填完之后保存在工具列表里点“添加”就能在工作流节点里看到了。这里再强调一遍Dify 的凭据校验本质上是往这个 URL 发一个请求如果 API 服务没起、Schema 路径不对、端口不通都会报凭据验证失败。所以先curl通接口再进 Dify 配置能省很多排查时间。4. 在工作流节点里接入 MySQL 查询4.1 一个可复用的应用场景我用一个“查询订单状态”的场景串一遍流程。用户在小程序里问“订单 10086 现在到哪一步了”工作流设计如下开始节点 → LLM 节点从用户问题中提取订单号→ 自定义工具节点调用 MySQL 查询接口→ LLM 节点把查询结果整理成自然语言→ 结束节点。LLM 节点里做参数提取时输出变量名设为order_id引用的方式是{{#node_order_id_extract.order_id#}}。这里注意如果用户没说订单号不要强行去查而是让 LLM 先反问。我在实际项目里给 LLM 的 System Prompt 里写了硬性条件提取不到订单号就不允许调用工具节点。这个约束能避免很多无意义的数据库查询。4.2 自定义工具节点的完整参数配置在工具节点里选择刚才创建的query动作参数按照 Schema 填写。如果你的接口设计成语义化参数请求体大概长这样{ sql: SELECT * FROM orders WHERE id %(order_id)s LIMIT 1, params: { order_id: {{#node_order_id_extract.order_id#}} } }如果接口只接收原始 SQL那就需要 LLM 节点把 SQL 拼好再传进来。我强烈不建议这样做因为拼 SQL 一旦出错错误信息会带着表结构回到 LLM 上下文里等于变相泄露数据库元信息。语义化参数之后SQL 模板在 API 端写死LLM 只能决定参数值安全边界就清楚多了。配置完可以先用“运行”按钮测试传一个真实存在的订单号看返回结果是否正常。Dify 的工具节点执行日志会显示请求体和响应体遇到问题直接在日志里排查比盲猜快很多。4.3 处理结果集过大与上下文超长这是 Dify 工作流接入数据库后最常遇到的问题热搜里也频繁出现“dify工作流 上下文超长”。根源是数据库查询结果被原封不动塞给了 LLM比如查“最近三个月的订单”可能一次性返回几百行每个字段都算 token很快就把上下文窗口挤爆了。我的处理分三层。第一层API 侧已经把行数限制在 50 行内。第二层工作流里加一个“代码节点”或“变量聚合器”把查询结果压缩成精简的 Markdown 表格只保留关键字段。举个例子数据库返回的每行包含id, user_id, product_name, price, status, created_at, internal_remark, supplier_id但给 LLM 只需要id, product_name, price, status那就先在代码节点里过滤字段而不是让 LLM 自己挑。第三层如果 50 行还是太长可以在提示词里要求 LLM 只总结前 5 条或者提示用户缩小时间范围。这三层下来上下文超长基本能解决。“变量聚合器”在这里也很好用。它的典型场景是把多个分支的结果聚合到一个列表里比如先查用户信息再查该用户的订单列表两个工具节点返回后用变量聚合器把它们合并成一个结构化对象后面 LLM 节点只引用聚合后的变量避免工作流里到处传递大块 JSON。4.4 多个查询组合与循环处理实际业务很少只查一张表。比如用户问“我的账户余额和最近订单一起看一下”工作流里可以并行放两个工具节点一个查users一个查orders。Dify 工作流支持并行分支两个节点都依赖前面提取出的参数且互不依赖那就应该并行减少响应时间。如果查询结果是列表而你要对每一行再做一次判断可以用“迭代”节点。我试过一次“查待发货订单再逐单查询库存剩余”的场景外层查订单列表内层用迭代节点逐条查询库存。注意迭代节点里的变量引用要小心取的是当前迭代项的字段不是外层列表整体。这个细节我在第一次搭建时绕了很久最后看节点输出才发现引错了层级。5. 常见问题与排查实录5.1 凭据验证失败与 SSL 连接错误这里把最头疼的两个问题放一起说。an error occurred during credentials validation是 Dify 在保存工具凭据或测试工具时后端尝试连接 MySQL 失败后的通用报错。我遇到过的原因有四种MySQL 账号主机白名单限制了 Dify 所在 IP、端口不对、密码带了特殊字符没有正确转义、MySQL 8 的认证插件不被客户端支持。排查时先看 MySQL 侧日志比如SHOW PROCESSLIST里有没有来自 Dify 的连接记录有记录说明网络和账号是通的再看认证报错。mysql ssl 连接错误则是另一类。如果你确认 MySQL 开启了 SSL但你的 API 连接串里没带证书或者 Dify 插件的驱动不接受自签证书就会报 SSL 相关错误。内网测试环境我一般直接关闭require_secure_transport后重启 MySQL生产环境则在 pymysql 连接参数里加ssl_ca指向 CA 文件同时把ssl_disabled设为False。千万别在 Dify 插件的连接串里强行拼?ssl-ca/path很多插件根本不解析这种参数。5.2 中文乱码中文乱码不是复杂问题但出现频率极高。Dify 工具返回的数据里中文如果显示成???十有八九是 MySQL 连接字符集没设置成utf8mb4。在 pymysql 的connect()参数里一定要加charsetutf8mb4同时确认数据库表本身的字符集也是utf8mb4。只改连接不改表等于白改。5.3 权限不足与账号刷新MySQL 授权之后如果查询时提示SELECT command denied to user大多数情况是授权对象不对。比如我给了orders表的权限但 SQL 里带了mydb.orders而账号只授权了mydb.orders理论上没问题但如果你授权的是*.*之外的库恰好工作流里又查了别的表那就报错。还有一种情况是修改完授权后没生效执行一下FLUSH PRIVILEGES保险。我的习惯是授权语句尽量精确到表宁可多建几个账号也不给一个账号放开整个库。5.4 SQL 注入与多语句风险这不是 Dify 的问题而是接入数据库后必须面对的安全底线。LLM 生成的参数如果不做约束被恶意注入到 SQL 里后果很严重。前面的 API 实现里已经做了三层防御只允许SELECT、不允许分号、参数化查询。还不够的话可以在 API 端维护一个表名白名单任何 SQL 只要包含白名单之外的表名就直接拒绝。我在生产环境里加了一层正则校验虽然代码丑但心里踏实。5.5 常见问题速查表问题现象可能原因快速处理方式凭据验证失败网络不通、端口错误、账号权限不足先在 Dify 服务器 telnet 3306再检查 MySQL 账号 host 和权限SSL 连接错误MySQL 开启 SSL客户端未配置证书内网关闭require_secure_transport生产在 API 层配置ssl_ca中文乱码连接字符集不是 utf8mb4连接参数加charsetutf8mb4表字符集统一 utf8mb4权限不足授权不完整或未刷新精确到表授权执行FLUSH PRIVILEGES上下文超长查询结果集过大API 层fetchmany(50)代码节点过滤字段提示词限制总结条数连接超时MySQLwait_timeout短连接池耗尽API 使用连接池每次查询设置MAX_EXECUTION_TIME6. 一些我自己实践后的经验项目里把 Dify 接 MySQL 这个事从零到一跑通之后我最大的体会是不要把数据库查询设计成一个“万能接口”。Dify 工作流真正需要的是一个接一个的“固定动作”比如“按订单号查订单”“按用户 ID 查余额”每个动作对应一句固定 SQL。这样做的好处是工作流可解释、可测试、可审计配合 LLM 做参数提取后准确率远高于让模型自己写 SQL。另外一个小技巧在工具节点返回的结果里给 API 接口额外加一个summary字段比如“共命中 3 条耗时 0.02 秒”让 LLM 直接引用这个字段生成回复开头。这样用户感知到的响应更快也避免 LLM 对结果集数量做无意义的猜测。Dify 工作流的日志里能看到每个节点的耗时如果数据库查询超过 1 秒优先查 SQL 索引而不是去调 Dify 的超时参数。如果你只是想在内部系统里快速跑通一个 Demo用现成的 MySQL 插件就够了。但一旦涉及多部门复用、数据脱敏、审计要求我还是建议花半天时间把查询 API 搭起来。这套方案的价值不在代码本身而在于它把“数据库访问”变成了一个有边界的服务Dify 只是这个服务的调用方这个边界就是生产环境稳定性的最后一道防线。