当前位置：首页 > news >正文

别再手动写接口了！用Flask+Ngrok快速给MySQL做个API，Dify直接调用

news 2026/5/28 16:28:15

用Flask+Ngrok为MySQL构建轻量级API网关：Dify集成实战指南

当我们需要在Dify中操作MySQL数据库时，直接暴露数据库连接信息不仅存在安全隐患，还会让工作流变得难以维护。这里介绍一种更优雅的解决方案：通过Flask构建API中间层，再借助Ngrok实现内网穿透，最终在Dify中通过简单的HTTP调用完成数据库操作。这种方法特别适合需要快速搭建原型或中小型项目的数据集成场景。

1. 为什么需要API中间层？

直接连接数据库虽然简单，但在企业环境中往往面临诸多限制：

安全风险：数据库凭证直接暴露在前端或低代码平台中
网络限制：生产环境数据库通常不允许外部直接访问
维护困难：连接参数变更时需要修改多处代码

相比之下，API中间层提供了：

统一入口：所有数据库操作通过标准化接口完成
权限集中：只需在API服务端维护数据库凭证
灵活扩展：可轻松添加缓存、日志、限流等中间件功能

# 安全对比示例 直接连接风险 = { "凭证暴露": "高", "网络要求": "需开放3306端口", "维护成本": "每个调用点单独修改" } API网关优势 = { "凭证暴露": "仅服务端可见", "网络要求": "只需HTTP访问", "维护成本": "单点修改全局生效" }

2. 快速搭建Flask数据库API服务

2.1 基础环境准备

首先确保已安装必要的Python包：

pip install flask mysql-connector-python

2.2 核心API实现

以下是一个完整的Flask应用示例，提供了安全的数据库操作接口：

from flask import Flask, request, jsonify import mysql.connector from functools import wraps app = Flask(__name__) # 数据库配置（实际使用时应从环境变量读取） DB_CONFIG = { 'host': '127.0.0.1', 'user': 'api_user', 'password': 'secure_password', 'database': 'business_data', 'pool_size': 5 # 连接池大小 } def db_connection_handler(f): @wraps(f) def wrapper(*args, **kwargs): conn = None try: conn = mysql.connector.connect(**DB_CONFIG) return f(conn, *args, **kwargs) except Exception as e: app.logger.error(f"Database error: {str(e)}") return jsonify({"error": "Database operation failed"}), 500 finally: if conn and conn.is_connected(): conn.close() return wrapper @app.route('/query', methods=['POST']) @db_connection_handler def query_data(conn): """安全执行查询操作""" data = request.json if not data or 'sql' not in data: return jsonify({"error": "SQL query is required"}), 400 cursor = conn.cursor(dictionary=True) try: cursor.execute(data['sql']) return jsonify({"data": cursor.fetchall()}) finally: cursor.close() @app.route('/execute', methods=['POST']) @db_connection_handler def execute_sql(conn): """执行写入操作""" data = request.json if not data or 'sql' not in data: return jsonify({"error": "SQL statement is required"}), 400 cursor = conn.cursor() try: cursor.execute(data['sql']) conn.commit() return jsonify({"affected_rows": cursor.rowcount}) except Exception as e: conn.rollback() raise e finally: cursor.close() if __name__ == '__main__': app.run(host='0.0.0.0', port=5000, threaded=True)

关键安全措施：

使用连接装饰器确保资源释放
查询和写入操作分离
自动事务回滚机制
字典格式游标便于结果处理

3. 使用Ngrok实现内网穿透

3.1 Ngrok基本配置

注册Ngrok账号并获取authtoken
下载对应平台的客户端
启动隧道服务：

ngrok http 5000 --region=us --subdomain=yourprefix

提示：商业项目中建议使用付费计划以获得固定域名和更高稳定性

3.2 高级配置选项

在ngrok.yml配置文件中可以设置更专业的参数：

authtoken: your_auth_token region: us tunnels: mysql-api: addr: 5000 proto: http host_header: "localhost:5000" bind_tls: true

启动指定配置：

ngrok start --config=ngrok.yml mysql-api

4. Dify中的集成实践

4.1 工作流配置

在Dify中创建新的工作流，添加"代码执行"节点：

import requests from typing import Dict API_ENDPOINT = "https://yourprefix.ngrok-free.app" AUTH_TOKEN = "your_shared_secret" # 建议使用Dify的密钥管理 def execute_sql(sql: str) -> Dict: headers = {"Authorization": f"Bearer {AUTH_TOKEN}"} try: response = requests.post( f"{API_ENDPOINT}/execute", json={"sql": sql}, headers=headers, timeout=10 ) response.raise_for_status() return response.json() except requests.exceptions.RequestException as e: return {"error": str(e)}

4.2 安全增强措施

接口认证：在Flask端添加JWT验证
SQL白名单：限制可执行的操作类型
请求限流：防止API被滥用

# Flask中的JWT验证示例 from flask_jwt_extended import JWTManager, jwt_required app.config['JWT_SECRET_KEY'] = 'super-secret' jwt = JWTManager(app) @app.route('/protected/query', methods=['POST']) @jwt_required() def protected_query(): # 受保护的查询接口 pass

5. 性能优化与生产建议

5.1 连接池管理

使用DBUtils实现高效的连接池：

from dbutils.pooled_db import PooledDB pool = PooledDB( creator=mysql.connector, maxconnections=10, **DB_CONFIG ) @app.route('/pool/query') def pool_query(): conn = pool.connection() # 使用连接...

5.2 监控与日志

添加Prometheus监控指标：

from prometheus_client import Counter, generate_latest REQUEST_COUNT = Counter('api_requests_total', 'Total API requests') @app.route('/metrics') def metrics(): return generate_latest() @app.after_request def after_request(response): REQUEST_COUNT.inc() return response