You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

Python Flask接收Meraki Webhook时如何实时更新所有客户端

解决方案:在Flask-SocketIO中全局发送消息给所有客户端

核心思路

Flask-SocketIO的socketio实例支持在任意代码位置调用emit方法,只要指定broadcast=True参数,就能将消息发送给所有已连接的客户端,无需局限在@socketio.on装饰的回调函数内。

完整实现代码

后端(Flask服务器)

from flask import Flask, request, render_template
from flask_socketio import SocketIO

app = Flask(__name__)
app.config['SECRET_KEY'] = 'your-secret-key-here'
# 初始化SocketIO实例,推荐使用eventlet作为异步引擎
socketio = SocketIO(app, cors_allowed_origins="*")

# 存储历史告警(可选,用于页面加载时展示已有内容)
alerts_history = []

def send_to_all_clients(data):
    """全局可用的消息推送函数"""
    socketio.emit('new_alert', data, broadcast=True)

@app.route('/endpoint', methods=['POST'])
def webhook_endpoint():
    """接收Meraki Webhook告警的端点"""
    if request.is_json:
        alert_data = request.get_json()
        alerts_history.append(alert_data)
        # 触发实时推送
        send_to_all_clients(alert_data)
        return 'Alert received', 200
    return 'Invalid request', 400

@app.route('/view')
def view_alerts():
    """展示告警的页面"""
    return render_template('alerts_view.html', alerts=alerts_history)

if __name__ == '__main__':
    # 必须用socketio.run启动服务器,才能处理WebSocket连接
    socketio.run(app, debug=True, host='0.0.0.0')

前端模板(templates/alerts_view.html)

<!DOCTYPE html>
<html>
<head>
    <title>Meraki Alerts</title>
</head>
<body>
    <h1>实时告警列表</h1>
    <ul id="alerts-list">
        {% for alert in alerts %}
            <li>{{ alert['alertType'] }} - {{ alert['occurredAt'] }}</li>
        {% endfor %}
    </ul>

    <!-- 引入SocketIO客户端库 -->
    <script src="https://cdnjs.cloudflare.com/ajax/libs/socket.io/4.0.1/socket.io.js"></script>
    <script>
        // 连接到SocketIO服务器
        const socket = io();

        // 监听服务器推送的new_alert事件
        socket.on('new_alert', function(data) {
            const alertsList = document.getElementById('alerts-list');
            const newItem = document.createElement('li');
            newItem.textContent = `${data.alertType} - ${data.occurredAt}`;
            alertsList.appendChild(newItem);
        });
    </script>
</body>
</html>

关键说明

  • send_to_all_clients函数直接通过全局的socketio实例调用emit,broadcast=True参数确保消息发送给所有已连接客户端。
  • Webhook端点接收到Meraki告警后,先存入历史列表(可选),再调用推送函数触发实时更新。
  • 前端通过SocketIO客户端连接服务器,监听new_alert事件,收到消息后动态更新DOM。
  • 启动服务器必须使用socketio.run(app),而非传统的app.run(),否则无法处理WebSocket连接。

注意事项

  • 安装依赖:pip install flask flask-socketio eventlet(eventlet是推荐的异步引擎,也可替换为gevent)。
  • 生产环境部署时,需搭配支持异步的服务器(如Gunicorn + eventlet worker)。

内容的提问来源于stack exchange,提问作者Ruben Driezen

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.07.02 23:31:24