一、创建一个工作流模版

工作流开始设置一个sql的参数字段

然后添加一个执行代码的模块

然后画线关联起来

然后点代码执行设置开始时候输入的sql参数

然后把python代码丢进去,（自己本地写好一个连接数据库调用的对外接口）

python
import json
import requests

def main(sql: str) -> dict:
    """
    调用执行SQL的接口，适配Dify的函数调用规范
    :param sql: 传入的SQL语句字符串
    :return: 包含执行结果的字典
    """
    # 注意：host.docker.internal是Docker容器访问宿主机的地址，若Dify部署在Docker中，需用此地址访问宿主机的Flask服务
    url = "http://192.168.31.22:5000/execute_sql"
    # 数据库连接信息
    connection_info = {
        "host": "localhost",
        "user": "root",
        "password": "123456",
        "database": "database"
    }
    # 构造请求体
    payload = {
        "sql": sql,
        "connection_info": connection_info
    }
    headers = {
        "Content-Type": "application/json"
    }
    try:
        # 发送POST请求
        response = requests.post(url, json=payload, headers=headers)
        if response.status_code == 200:
            try:
                # 解析响应结果
                return {"result": str(response.json()["result"])}
            except Exception as e:
                return {"result": f"解析响应 JSON 失败: {str(e)}"}
        else:
            return {"result": f"请求失败，状态码: {response.status_code}"}
    except Exception as e:
        return {"result": str(e)}

然后添加一个结束节点，然后关联起来，然后写入执行代码的输出参数即可

二、测试使用

发布后测试使用，输入一个sql语句进行查询 然后成功返回对应的结果