dify案例：工作流执行sql获取执行结果

摘要：看了很多dify的资料和视频，但是动起手来之后，才发现有很多知识自己理解的不是很透彻，所以需要实现很多的案例来加深对知识的理解。第一个案例就是对话框输入sql，然后通过两种工作流节点：代码执行和http请求，测试dify的工作流节点+python使用flash

背景

看了很多dify的资料和视频，但是动起手来之后，才发现有很多知识自己理解的不是很透彻，所以需要实现很多的案例来加深对知识的理解。第一个案例就是对话框输入sql，然后通过两种工作流节点：代码执行和http请求，测试dify的工作流节点+python使用flash搭建基本服务器的过程。

最终效果

python搭建后台服务

基本代码如下，强烈建议大家使用ai工具编写，cursor或者trae，即使python很多工具不会，也能快速搭建服务，新建server.py：

from Flask import Flask, request, jsonifyfrom flask_sqlalchemy import SQLAlchemyfrom SQLalchemy import text# 创建Flask应用app = Flask(__name__)# 数据库配置app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql+pymysql://root:123456@localhost:3306/ry'# 初始化数据库db = SQLAlchemy(app)# 自定义接口@app.route('/api/execute-sql', methods=['POST'])def execute_sql:try:data = request.get_jsonsql = data.get('sql')if not sql:return jsonify({'success': False,'message': 'SQL语句不能为空'}), 400# 执行SQL查询result = db.session.execute(text(sql))# 获取结果if result.returns_rows:columns = result.keysrows = [dict(zip(columns, row)) for row in result.fetchall]return jsonify({'success': True,'data': rows,'message': '查询执行成功'else:return jsonify({'success': True,'message': 'SQL执行成功'except Exception as e:return jsonify({'success': False,'message': f'执行SQL时发生错误: {str(e)}'}), 500if __name__ == '__main__':# 启动服务app.run(debug=True, host='0.0.0.0', port=5000)

启动服务：python server.py，出现如下信息，启动成功，端口是5000

启动服务成功

代码执行节点

最后查询的节点，一定要使用result封装，否则会报错result没有返回值，我在这里踩了个坑

官网对输出格式做了限定，所以一定要按照固定的格式返回

执行代码：就是发起http请求，然后返回结果，可以参考下，我的代码是经过测试通过的，可以直接使用。注意这里的host.docker.internal地址，不要写localhost，因为dify是在docker容器里面启动的，而docker容器无法直接使用localhost访问宿主机。

def main(arg1: str) -> str:import requests# 设置超时时间为 10 秒response = requests.post(f'http://host.docker.internal:5000/api/execute-sql',json={'sql': arg1},timeout=10# 这里必须要使用result返回result = response.jsonreturn {"result":result}http节点配置

这里需要注意的就是json参数的拼装，dify不会直接给变量加上双引号，需要我们显示手动加上