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

Python Flask集成Twilio媒体流:无法记录WebSocket消息

问题描述

正在开发Python Flask应用,集成Twilio实现通话媒体流功能,通过WebSocket处理Twilio的通话流消息。WebSocket已成功连接到localhost:8080,但无法记录传入的WebSocket消息(如connected/start/media/stop事件)。基于Twilio的JS教程转换为Python代码后,预期日志未出现。

代码如下:

import asyncio
import websockets
import os
from flask import Flask, request, Response

app = Flask(__name__)
app.config['SECRET_KEY'] = os.urandom(24)

@app.route("/", methods=["GET", "POST"])
def root():
    # ... TwiML响应处理 ...

async def handle_websocket(websocket, path):
    print("New Connection Initiated")
    try:
        async for message in websocket:
            print(f"Received message: {message}")
            # 在此添加WebSocket消息处理逻辑
            # 可使用'message'变量解析来自Twilio的JSON消息
            msg = json.loads(message)
            event = msg.get('event')
            if event == 'connected':
                print('A new call has connected.')
            elif event == 'start':
                print(f"Starting Media Stream {msg.get('streamSid')}")
            elif event == 'media':
                print("Receiving Audio...")
            elif event == 'stop':
                print('Call Has Ended')
            # 目前先将消息回传给客户端
            await websocket.send(message)
    except websockets.exceptions.ConnectionClosedOK:
        print("WebSocket connection closed.")
    except Exception as e:
        print(f"Error: {e}")

if __name__ == "__main__":
    start_server = websockets.serve(handle_websocket, "0.0.0.0", 8080)
    print("Listening at Port 8080")
    asyncio.get_event_loop().run_until_complete(start_server)
    asyncio.get_event_loop().run_forever()

运行时仅打印New Connection Initiated,未收到Twilio的预期消息事件日志。

排查与解决方案
  • 缺失json模块导入:代码中使用了json.loads()但未导入json库,会触发异常并被捕获,导致后续消息处理逻辑无法执行。添加import json到代码顶部即可修复。
  • 验证TwiML配置正确性:确保/路由返回的TwiML中,<Stream>标签的url属性指向公网可访问的WebSocket地址(Twilio无法直接访问localhost)。例如使用内网穿透工具生成公网地址,将url设为wss://xxx.ngrok.io/ws(需对应你的WebSocket路径配置)。
  • 先打印原始消息排查解析问题:暂时注释json.loads()及后续事件判断代码,仅保留print(f"Received message: {message}"),确认是否真的收到Twilio的消息。如果原始消息未打印,说明Twilio未成功推送消息到你的WebSocket服务,重点检查TwiML的Stream配置和公网可达性。
  • 细化异常日志:在通用异常捕获中打印更详细的错误信息,比如print(f"Error type: {type(e).__name__}, Message: {str(e)}"),便于定位解析或消息处理中的具体问题。
  • 检查websockets库版本:确保使用的websockets库版本支持Twilio的WebSocket协议,建议安装最新稳定版:pip install --upgrade websockets。

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

相关产品推荐
方舟 Agent Plan

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

最近更新时间:2026.07.14 04:53:26