From 81f47686613cd6fa3e51ea7e5e41d91d74dbec65 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Tue, 22 Apr 2025 12:43:56 -0400 Subject: [PATCH] Automatically hangup Twilio calls --- CHANGELOG.md | 5 +++ src/pipecat/serializers/twilio.py | 36 +++++++++++++++++-- .../transports/network/fastapi_websocket.py | 1 + .../transports/network/websocket_server.py | 5 ++- 4 files changed, 44 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 9ae41fe67..3458f23e7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Added automatic hangup logic to the Twilio serializer. This feature hangs up + the Twilio call when an `EndFrame` is received. It is enabled by default and + is configurable via the `auto_hang_up` `InputParam`. To use this feature, set + up the following env vars: `TWILIO_ACCOUNT_SID` and `TWILIO_AUTH_TOKEN`. + - Added `SmartTurnMetricsData`, which contains end-of-turn prediction metrics, to the `MetricsFrame`. Using `MetricsFrame`, you can now retrieve prediction confidence scores and processing time metrics from the smart turn analyzers. diff --git a/src/pipecat/serializers/twilio.py b/src/pipecat/serializers/twilio.py index e08920132..009cb748b 100644 --- a/src/pipecat/serializers/twilio.py +++ b/src/pipecat/serializers/twilio.py @@ -4,15 +4,19 @@ # SPDX-License-Identifier: BSD 2-Clause License # +import asyncio import base64 import json +import os from typing import Optional +from loguru import logger from pydantic import BaseModel from pipecat.audio.utils import create_default_resampler, pcm_to_ulaw, ulaw_to_pcm from pipecat.frames.frames import ( AudioRawFrame, + EndFrame, Frame, InputAudioRawFrame, InputDTMFFrame, @@ -29,9 +33,11 @@ class TwilioFrameSerializer(FrameSerializer): class InputParams(BaseModel): twilio_sample_rate: int = 8000 # Default Twilio rate (8kHz) sample_rate: Optional[int] = None # Pipeline input rate + auto_hang_up: bool = True # Whether to automatically hang up on EndFrame/CancelFrame - def __init__(self, stream_sid: str, params: InputParams = InputParams()): + def __init__(self, stream_sid: str, call_sid: str, params: InputParams = InputParams()): self._stream_sid = stream_sid + self._call_sid = call_sid self._params = params self._twilio_sample_rate = self._params.twilio_sample_rate @@ -47,7 +53,9 @@ class TwilioFrameSerializer(FrameSerializer): self._sample_rate = self._params.sample_rate or frame.audio_in_sample_rate async def serialize(self, frame: Frame) -> str | bytes | None: - if isinstance(frame, StartInterruptionFrame): + if self._params.auto_hang_up and isinstance(frame, (EndFrame)): + asyncio.create_task(self._hang_up_call()) + elif isinstance(frame, StartInterruptionFrame): answer = {"event": "clear", "streamSid": self._stream_sid} return json.dumps(answer) elif isinstance(frame, AudioRawFrame): @@ -68,6 +76,30 @@ class TwilioFrameSerializer(FrameSerializer): elif isinstance(frame, (TransportMessageFrame, TransportMessageUrgentFrame)): return json.dumps(frame.message) + async def _hang_up_call(self): + """Hang up the Twilio call using the REST API.""" + try: + from twilio.rest import Client + + account_sid = os.getenv("TWILIO_ACCOUNT_SID") + auth_token = os.getenv("TWILIO_AUTH_TOKEN") + + if not account_sid or not auth_token: + logger.warning( + "Missing Twilio credentials (TWILIO_ACCOUNT_SID, TWILIO_AUTH_TOKEN), cannot hang up call" + ) + return + + # Create Twilio client and hang up the call using the correct CallSid + client = Client(account_sid, auth_token) + client.calls(self._call_sid).update(status="completed") + + logger.info(f"Successfully terminated Twilio call {self._call_sid}") + except ImportError: + logger.warning("Twilio Python SDK not installed. Install with: pip install twilio") + except Exception as e: + logger.exception(f"Failed to hang up Twilio call: {e}") + async def deserialize(self, data: str | bytes) -> Frame | None: message = json.loads(data) diff --git a/src/pipecat/transports/network/fastapi_websocket.py b/src/pipecat/transports/network/fastapi_websocket.py index 2114d357d..8148f8007 100644 --- a/src/pipecat/transports/network/fastapi_websocket.py +++ b/src/pipecat/transports/network/fastapi_websocket.py @@ -207,6 +207,7 @@ class FastAPIWebsocketOutputTransport(BaseOutputTransport): async def stop(self, frame: EndFrame): await super().stop(frame) + await self._write_frame(frame) await self._client.disconnect() async def cancel(self, frame: CancelFrame): diff --git a/src/pipecat/transports/network/websocket_server.py b/src/pipecat/transports/network/websocket_server.py index 8947cbab3..ca0f0e3a9 100644 --- a/src/pipecat/transports/network/websocket_server.py +++ b/src/pipecat/transports/network/websocket_server.py @@ -157,7 +157,8 @@ class WebsocketServerInputTransport(BaseInputTransport): self, websocket: websockets.WebSocketServerProtocol, session_timeout: int ): """Wait for session_timeout seconds, if the websocket is still open, - trigger timeout event.""" + trigger timeout event. + """ try: await asyncio.sleep(session_timeout) if not websocket.closed: @@ -205,6 +206,8 @@ class WebsocketServerOutputTransport(BaseOutputTransport): if isinstance(frame, StartInterruptionFrame): await self._write_frame(frame) self._next_send_time = 0 + elif isinstance(frame, EndFrame): + await self._write_frame(frame) async def send_message(self, frame: TransportMessageFrame | TransportMessageUrgentFrame): await self._write_frame(frame)