Talk to AI via Phone Calls with Twilio and TEN Framework

# From: tenapp/ten_packages/extension/main_python/server.py
@self.app.post("/api/call")
async def create_call(request: Request):
    """Create a new outbound call"""
    body = await request.json()
    phone_number = body.get("phone_number")
    message = body.get("message", "Hello from Twilio!")

    # Create TwiML response with media stream
    twiml_response = VoiceResponse()

    # Configure WebSocket URL for real-time audio streaming
    ws_protocol = "wss" if self.config.twilio_use_wss else "ws"
    media_ws_url = f"{ws_protocol}://{self.config.twilio_public_server_url}/media"

    connect = twiml_response.connect()
    connect.stream(url=media_ws_url)  # This tells Twilio to connect to our WebSocket
    twiml_response.append(connect)

    # Create the call via Twilio API
    call = self.twilio_client.calls.create(
        to=phone_number,
        from_=self.config.twilio_from_number,
        twiml=str(twiml_response)
    )

# From: tenapp/ten_packages/extension/main_python/server.py
@self.app.websocket("/media")
async def websocket_endpoint(websocket: WebSocket):
    """WebSocket endpoint for Twilio media streaming"""
    # Accept the connection immediately
    await websocket.accept()

    # Send confirmation message
    await websocket.send_text(
        '{"type": "connected", "message": "WebSocket connection established"}'
    )

    # Store the WebSocket connection for this call
    call_sid = None

    while True:
        data = await websocket.receive_text()
        message = json.loads(data)

        if message.get("event") == "start":
            # Extract call SID and store WebSocket connection
            call_sid = message.get("start", {}).get("callSid", "")
            self.active_call_sessions[call_sid]["websocket"] = websocket

# From: tenapp/ten_packages/extension/main_python/server.py
if message.get("event") == "media":
    # Extract base64-encoded audio payload
    audio_payload = message.get("media", {}).get("payload", "")
    stream_sid = message.get("streamSid", "")

    if audio_payload and call_sid:
        # Forward audio to TEN framework for processing
        await self.extension_instance._forward_audio_to_ten(
            audio_payload, stream_sid
        )

# From: tenapp/ten_packages/extension/main_python/extension.py
async def _forward_audio_to_ten(self, audio_payload: str, call_sid: str):
    """Forward audio data to TEN framework and dump PCM audio"""
    # Decode base64 audio data from Twilio
    audio_data = base64.b64decode(audio_payload)

    # Send to ASR (speech-to-text) extension
    await _send_data(self.ten_env, "audio", "stt", {
        "audio_data": audio_data,
        "session_id": call_sid,
        "stream_id": 100
    })

# From: tenapp/ten_packages/extension/main_python/extension.py
async def send_audio_to_twilio(self, audio_data: bytes, call_sid: str):
    """Send audio data to Twilio via WebSocket"""
    # Get the WebSocket connection for this call
    websocket = self.server_instance.active_call_sessions[call_sid].get("websocket")

    # Downsample audio from 16000 Hz to 8000 Hz for Twilio
    source_rate = 16000  # TTS generated audio sample rate
    target_rate = 8000   # Twilio required sample rate
    downsampled_audio = audioop.ratecv(
        audio_data, 2, 1, source_rate, target_rate, None
    )[0]

    # Encode as base64 for Twilio
    encoded_audio = base64.b64encode(downsampled_audio).decode('utf-8')

    # Send to Twilio via WebSocket
    message = {
        "event": "media",
        "streamSid": call_sid,
        "media": {
            "payload": encoded_audio
        }
    }

    await websocket.send_text(json.dumps(message))