Flask+uWSGI+Nginx环境下，Webhook触发后如何向EventSource发送响应？

我最近做了一个Flask应用，专门对接外部电话系统，实现了来电触发前端实时推送的功能，核心是用线程事件（threading.Event）在不同路由间传递触发信号，具体实现逻辑和代码如下：

核心逻辑拆解 #

• /webhook 路由：负责接收外部电话应用发来的POST请求（来电时自动触发），收到请求后通过threading.Event.set()发送触发信号
• /eventstream 路由：生成保持长连接的EventSource响应，监听线程事件，一旦收到信号就向前端推送事件内容
• /eventsource 路由：用来渲染带有EventSource连接的网页，建立前端和后端的长连接通道

关键代码片段 #

from flask import Flask, render_template, request
import threading

app = Flask(__name__)
# 假设telfa是你定义的蓝图，这里直接用app做示例
telfa = app

# 线程事件对象，用来跨路由传递来电触发信号
telfa_called = threading.Event()
telfa_called.clear()
# 临时存储来电相关信息的变量
call = ""

@telfa.route('/webhook', methods=['GET', 'POST'])
def webhook():
    global call
    print('THE CALL IS H...')
    # 解析POST请求中的来电数据（根据外部电话应用的格式调整）
    if request.method == 'POST':
        call = request.json.get('call_details', 'New incoming call detected')
    # 触发事件，通知EventStream推送内容
    telfa_called.set()
    return "Call received successfully", 200

@telfa.route('/eventstream')
def eventstream():
    def event_generator():
        while True:
            # 阻塞等待来电触发信号
            telfa_called.wait()
            # 按照EventSource协议格式推送数据
            yield f"data: {call}\n\n"
            # 重置事件，等待下一次来电触发
            telfa_called.clear()
    # 返回符合EventSource要求的响应类型
    return app.response_class(event_generator(), mimetype='text/event-stream')

@telfa.route('/eventsource')
def eventsource():
    # 渲染带有EventSource连接的前端页面
    return render_template('call_notify_page.html')

额外说明 #

• threading.Event 是线程安全的信号机制，适配Flask默认的多线程运行环境，能安全地在路由间传递触发信号
• EventSource协议要求响应MIME类型必须为text/event-stream，且数据格式要严格遵循data: 内容\n\n的规范
• 示例中用全局变量call存储来电信息，生产环境建议改用线程本地存储或轻量消息队列，避免多线程数据冲突

内容的提问来源于stack exchange，提问作者Fide