Automatically hangup Twilio calls

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.

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)
 

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):

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)