使用Dify訪問mysql數(shù)據(jù)庫詳細代碼示例

更新時間：2025年03月15日 11:49:55 作者：wxl781227

這篇文章主要介紹了使用Dify訪問mysql數(shù)據(jù)庫的相關資料,并詳細講解了如何在本地搭建數(shù)據(jù)庫訪問服務,使用ngrok暴露到公網(wǎng),并創(chuàng)建知識庫、數(shù)據(jù)庫訪問工作流和智能體,需要的朋友可以參考下

1、在本地搭建數(shù)據(jù)庫訪問的服務，并使用ngrok暴露到公網(wǎng)。

#sql_tools.py

from flask import Flask, request, jsonify
import mysql.connector

# 數(shù)據(jù)庫連接配置
config = {
    'user': 'your_username',
    'password': 'your_password',
    'host': 'localhost',
    'database': 'your_database',
    'raise_on_warnings': True
}

# 初始化Flask應用
app = Flask(__name__)

# 連接數(shù)據(jù)庫
def connect_to_database():
    try:
        conn = mysql.connector.connect(**config)
        print("Connected to MySQL database")
        return conn
    except mysql.connector.Error as err:
        print(f"Error: {err}")
        return None

# 執(zhí)行SQL查詢
def execute_query(conn, sql):
    cursor = conn.cursor()
    try:
        cursor.execute(sql)
        if sql.strip().lower().startswith("select"):
            # 如果是查詢操作，返回結(jié)果
            result = cursor.fetchall()
            return result
        else:
            # 如果是插入、更新、刪除操作，提交事務并返回受影響的行數(shù)
            conn.commit()
            return cursor.rowcount
    except mysql.connector.Error as err:
        print(f"Error executing SQL: {err}")
        return None
    finally:
        cursor.close()

# HTTP接口：執(zhí)行SQL
@app.route('/execute', methods=['POST'])
def execute_sql():
    # 獲取請求中的SQL語句
    data = request.json
    if not data or 'sql' not in data:
        return jsonify({"error": "SQL statement is required"}), 400

    sql = data['sql']
    conn = connect_to_database()
    if not conn:
        return jsonify({"error": "Failed to connect to database"}), 500

    # 執(zhí)行SQL
    result = execute_query(conn, sql)
    conn.close()

    if result is None:
        return jsonify({"error": "Failed to execute SQL"}), 500

    # 返回結(jié)果
    return jsonify({"result": result})

# 啟動Flask應用
if __name__ == '__main__':
    app.run(host='0.0.0.0', port=3000)

2、創(chuàng)建知識庫，導入表結(jié)構(gòu)描述。

3、創(chuàng)建數(shù)據(jù)庫訪問工作流。

代碼執(zhí)行：

import requests
def main(sql: str) -> dict:
    # 定義API的URL
    url = "https://xxx.ngrok-free.app/execute"

    # 構(gòu)造請求體
    payload = {
        "sql": sql
    }

    # 發(fā)送POST請求
    try:
        response = requests.post(url, json=payload)
    
        # 檢查響應狀態(tài)碼
        if response.status_code == 200:
            # 解析響應數(shù)據(jù)
            result = response.json()
            return {
                "result": f"{result}"
            }
        else:
            return {
                "result": f"請求失敗，狀態(tài)碼：{response.status_code},{response.json()}"
            }
    except requests.exceptions.RequestException as e:
        return {
            "result": f"請求異常：{e}"
        }

4、創(chuàng)建數(shù)據(jù)庫智能體