Twilio Media Streams无法连接WebSocket，调用/voice端点后立即断开

针对你遇到的通话立即断开、Twilio无WebSocket连接尝试记录的问题，按以下步骤排查解决：

1. 检查TwiML响应的完整性与正确性 #

Twilio执行完TwiML指令后，若没有后续保持通话的动作会直接挂断，这是最常见的原因：

• 必须确保/voice端点返回的TwiML包含**根节点**，且嵌套在标签内
• 必须添加、等保持通话的动词，否则Twilio启动流后会立即终止通话
• 正确示例：

<Response>
  <Start>
    <Stream url="wss://my-ngrok-subdomain.ngrok.io/stream" />
  </Start>
  <Say>请开始说话，音频正在处理中</Say>
</Response>

2. 验证Flask端点的Content-Type设置 #

Twilio要求TwiML响应的Content-Type必须为application/xml，Flask默认返回text/html会导致解析失败：

• 在/voice路由中显式设置响应头：

from flask import make_response

@app.route('/voice', methods=['POST'])
def voice():
    twiml = """
    <Response>
      <Start>
        <Stream url="wss://my-ngrok-subdomain.ngrok.io/stream" />
      </Start>
      <Say>正在连接音频服务...</Say>
    </Response>
    """
    return make_response(twiml, 200, {'Content-Type': 'application/xml'})

3. 确认Twilio号码的Webhook配置 #

• 检查Twilio控制台中目标号码的语音Webhook是否正确指向你的/voice端点（POST方法），注意ngrok域名每次重启会变化，需同步更新
• 点击Webhook设置中的「Validate」按钮，确认端点能正常返回格式正确的TwiML

4. 排查WebSocket路由配置 #

确保Flask-Sockets的路由设置正确，没有路径或初始化错误：

from flask_sockets import Sockets

app = Flask(__name__)
sockets = Sockets(app)

@sockets.route('/stream')
def stream(ws):
    handle_audio(ws)
    return ''

5. 检查Twilio调试工具与日志 #

• 打开Twilio控制台的Debugger，查看是否有隐藏的错误（如TwiML解析失败、Webhook超时等），这类错误不会出现在普通通话日志中
• 确认WebSocket端点支持TLS 1.2+（ngrok默认支持，自建服务器需手动配置）

代码优化建议（避免意外断开） #

给你的WebSocket处理函数添加异常捕获，防止因消息格式错误导致连接中断：

def handle_audio(ws):
    logging.info('Handling audio')
    while not ws.closed:
        message = ws.receive()
        if message is None:
            logging.info("No message received...")
            continue

        try:
            data = json.loads(message)
        except json.JSONDecodeError:
            logging.error(f"Failed to parse message: {message}")
            continue

        event = data.get('event')
        if not event:
            logging.warning(f"Message missing event field: {message}")
            continue

        if event == "connected":
            logging.info("Connected Message received: {}".format(message))
        elif event == "start":
            logging.info("Start Message received: {}".format(message))
        elif event == "media":
            try:
                payload = data['media']['payload']
                logging.info("Payload is: {}".format(payload))
                audio_data = base64.b64decode(payload)
                logging.info("That's {} bytes".format(len(audio_data)))

                audio_chunks = whisper_handler.split_on_silence(audio_data)
                logging.info('Split audio data into %d chunks', len(audio_chunks))

                transcriptions = [whisper_handler.transcribe_audio(chunk) for chunk in audio_chunks]
                logging.info('Transcribed audio chunks: %s', transcriptions)
            except KeyError as e:
                logging.error(f"Missing key in media message: {e}")
        elif event == "closed":
            logging.info("Closed Message received: {}".format(message))
            break

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