Twilio对接Ngrok遇ERR_NGROK_3200错误及通话中断问题求助

问题概述 #

调用/make-call接口触发Internal Server Error，排查发现Ngrok出现ERR_NGROK_3200错误。通话可成功发起，但接通后立即断开并提示Sorry, an application error has occurred。

代码实现 #

import os
import json
import base64
import asyncio
import websockets
import uvicorn
from fastapi import FastAPI, WebSocket, Request
from fastapi.responses import JSONResponse
from fastapi.responses import HTMLResponse
from fastapi.websockets import WebSocketDisconnect
from twilio.rest import Client
from twilio.twiml.voice_response import VoiceResponse, Connect, Say, Stream
from google import genai
from dotenv import load_dotenv
import logging


logger = logging.getLogger(__name__)
logging.basicConfig(
    filename='errors.log',
    encoding='utf-8',
    level=logging.INFO,
    format='%(asctime)s %(message)s'
)

load_dotenv()

# Initialize Google Gemini API
MODEL = os.getenv("MODEL")
genai_client = genai.Client(http_options={"api_version": "v1alpha"})


# Twilio credentials and config
TWILIO_ACCOUNT_SID = os.getenv("TWILIO_ACCOUNT_SID")
TWILIO_AUTH_TOKEN = os.getenv("TWILIO_AUTH_TOKEN")
TWILIO_PHONE_NUMBER = os.getenv("TWILIO_PHONE_NUMBER")
NGROK_URL = os.getenv("NGROK_URL")
PORT = os.getenv("PORT", 5050)

client = Client(TWILIO_ACCOUNT_SID, TWILIO_AUTH_TOKEN)

# FastAPI app
app = FastAPI()


@app.get("/", response_class=HTMLResponse)
async def index_page():
    logging.error("Inside '/'.")
    return {"message": "Twilio Media Stream Server is running!"}


@app.post("/make-call")
async def make_call(request: Request):
    logging.error("Inside in /make_call.")
    """
    Initiates a VoIP call.
    Handles both JSON and form-encoded requests.
    """
    try:
        logging.error("Inside try block 01 of /make-call.")
        # Parsing JSON payload
        data = await request.json()
        to_number = data.get("to")
        logging.error(f'Data:{data}')
    except Exception as e:
        logging.error(f'Error at line 68:{e}')
        logging.error("Inside except block 01 of /make-call.")
        # Fallback to form data if JSON isn't provided
        form_data = await request.form()
        to_number = form_data.get("to")

    if not to_number:
        return {"error": "Recipient number is required"}, 400

    try:
        logging.error("Inside try block 02 of /make-call.")
        call = client.calls.create(
            url=f"{NGROK_URL}/outgoing-call",
            to=to_number,
            from_=TWILIO_PHONE_NUMBER
        )

        return {"message": "Call initiated",
                "call_sid": call.sid}, 200
    except Exception as e:
        logging.error(f"Error in /make_call: {e}")
        return {"error": str(e)}, 500


@app.api_route("/outgoing-call", methods=["GET", "POST"])
async def handle_outgoing_call(request: Request):
    logging.error("Inside in /outgoing-call.")
    """
    Handle outgoing call and return TwiML response to connect to Media Stream.
    """
    response = VoiceResponse()
    response.say("Please wait while we connect your call to the AI voice assistant...")
    response.pause(length=1)
    connect = Connect()
    connect.stream(url=f"wss://{request.url.hostname}/media-stream")
    response.append(connect)
    logging.error("Exiting /outgoing-call.")
    return HTMLResponse(content=str(response), media_type="application/xml")


@app.websocket("/media-stream")
async def handle_media_stream(websocket: WebSocket):
    logging.error("Inside /media_stream.")
    """
    Handle WebSocket connections between Twilio and Gemini API.
    """
    print("Client connected")
    await websocket.accept()
    try:
        async for message in websocket.iter_text():
            data = json.loads(message)
            logging.info(f'Data:{data}')
            if "event" in data and data["event"] == "media":
                print(f"Received audio payload: {data['media']['payload']}")
            # Process this data with Gemini if needed
    except WebSocketDisconnect:
        print("Client disconnected.")

    # Connect to Gemini session
    async with genai_client.aio.live.connect(model=MODEL) as gemini_ws:
        logging.error("Inside connect to gemini session")
        async def receive_from_twilio():
            logging.error("Inside receive_from_twilio.")
            """
            Receive audio data from Twilio and send it to Gemini API.
            """
            try:
                async for message in websocket.iter_text():
                    data = json.loads(message)
                    if data["event"] == "media":
                        audio = data["media"]["payload"]
                        await gemini_ws.send(input=base64.b64decode(audio), end_of_turn=False)
            except WebSocketDisconnect:
                print("Twilio disconnected")

        async def send_to_twilio():
            logging.error("Inside send_to_twilio.")
            """
            Receive audio or text data from Gemini and send back to Twilio.
            """
            async for chunk in gemini_ws.receive():
                if chunk.get("text"):
                    ai_text = chunk["text"]
                    print(f"Gemini Response: {ai_text}")
                    # Send text-to-speech via Twilio's media stream
                    await websocket.send_json({"text": ai_text})

                if chunk.get("audio"):
                    ai_audio = base64.b64encode(chunk["audio"]).decode("utf-8")
                    await websocket.send_json({"event": "media", "media": {"payload": ai_audio}})

        logging.error("Exiting /media-stream")

        await asyncio.gather(receive_from_twilio(), send_to_twilio())


if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=PORT)

运行步骤 #

• 终端1：uvicorn <file-name>:app --port 5000（启动本地服务器）
• 终端2：ngrok http 5000（获取公网URL并配置到.env文件及Twilio webhook）
• 终端3：curl -X POST http://127.0.0.1:5000/make-call -d "to=+xxxxxxxxxxxx"（发起通话）

已尝试的方法 #

• 查阅Ngrok ERR_NGROK_3200官方文档
• 使用GPT排查错误
• 观看相关YouTube视频

修复方案 #

1. 解决ERR_NGROK_3200错误 #

该错误是Ngrok免费版的流量限制或隧道滥用检测导致：

• 重启Ngrok隧道，生成新的公网URL，更新.env中的NGROK_URL（仅保留域名，如abc123.ngrok.io，不要带http://前缀）
• 避免短时间内频繁发起测试通话，减少隧道负载

2. 修复WebSocket逻辑错误 #

当前/media-stream函数存在时序问题：先接收Twilio消息再连接Gemini，导致Twilio超时断开。修改后的代码如下：

@app.websocket("/media-stream")
async def handle_media_stream(websocket: WebSocket):
    logging.error("Inside /media_stream.")
    print("Client connected")
    await websocket.accept()

    try:
        async with genai_client.aio.live.connect(model=MODEL) as gemini_ws:
            logging.error("Connected to Gemini session")
            
            async def receive_from_twilio():
                logging.error("Starting receive_from_twilio")
                try:
                    async for message in websocket.iter_text():
                        data = json.loads(message)
                        logging.info(f'Received data: {data}')
                        if data["event"] == "media":
                            audio = data["media"]["payload"]
                            await gemini_ws.send(input=base64.b64decode(audio), end_of_turn=False)
                        elif data["event"] == "start":
                            # 发送初始消息触发Gemini响应
                            await gemini_ws.send(input="Hello, how can I assist you?", end_of_turn=False)
                except WebSocketDisconnect:
                    print("Twilio disconnected")
                    await gemini_ws.close()

            async def send_to_twilio():
                logging.error("Starting send_to_twilio")
                try:
                    async for chunk in gemini_ws.receive():
                        if chunk.get("text"):
                            ai_text = chunk["text"]
                            print(f"Gemini Response: {ai_text}")
                            # 用Twilio指定格式触发语音合成
                            await websocket.send_json({"event": "speak", "text": ai_text})

                        if chunk.get("audio"):
                            ai_audio = base64.b64encode(chunk["audio"]).decode("utf-8")
                            await websocket.send_json({"event": "media", "media": {"payload": ai_audio}})
                except Exception as e:
                    logging.error(f"Error sending to Twilio: {e}")

            await asyncio.gather(receive_from_twilio(), send_to_twilio())
    except Exception as e:
        logging.error(f"Error in media stream: {e}")
        await websocket.close()

• 将Gemini连接逻辑移至WebSocket接受后立即执行，避免Twilio等待超时
• 修复Twilio消息格式：使用{"event": "speak", "text": "..."}触发语音合成，原格式不被Twilio识别
• 增加start事件处理，主动发送初始消息触发Gemini响应

3. 统一端口配置 #

代码中默认端口为5050，但启动命令用的是5000，需保持一致：

• 修改.env中的PORT=5000，或启动命令改为uvicorn <file-name>:app --port 5050

4. 验证Twilio配置 #

• 确认Twilio语音Webhook指向{NGROK_URL}/outgoing-call，请求方法设置为POST
• 在Twilio控制台查看通话日志，检查/outgoing-call返回的TwiML是否有效

5. 日志排查 #

查看errors.log文件，重点检查：

• Gemini API密钥是否正确配置（需在.env中添加GOOGLE_API_KEY）
• Twilio账号SID、Auth Token、号码是否匹配控制台信息

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