Overview
TwilioFrameSerializer
enables integration with Twilio’s Media Streams WebSocket protocol, allowing your Pipecat application to handle phone calls via Twilio’s voice services.
Features
- Bidirectional audio conversion between Pipecat and Twilio
- DTMF (touch-tone) event handling
- Automatic call termination via Twilio’s REST API
- μ-law audio encoding/decoding
Installation
The TwilioFrameSerializer
does not require any additional dependencies beyond the core Pipecat library.
Configuration
Constructor Parameters
The Twilio Media Stream SID
call_sid
Optional[str]
default:"None"
The associated Twilio Call SID (required for auto hang-up)
account_sid
Optional[str]
default:"None"
Twilio account SID (required for auto hang-up)
auth_token
Optional[str]
default:"None"
Twilio auth token (required for auto hang-up)
params
InputParams
default:"InputParams()"
Sample rate used by Twilio (typically 8kHz)
Optional override for pipeline input sample rate
Whether to automatically terminate call on EndFrame
Basic Usage
from pipecat.serializers.twilio import TwilioFrameSerializer
from pipecat.transports.network.fastapi_websocket import (
FastAPIWebsocketTransport,
FastAPIWebsocketParams
)
# Extract required values from Twilio WebSocket connection
stream_sid = call_data["start"]["streamSid"]
call_sid = call_data["start"]["callSid"]
# Create serializer
serializer = TwilioFrameSerializer(
stream_sid=stream_sid,
call_sid=call_sid,
account_sid="your_twilio_account_sid",
auth_token="your_twilio_auth_token"
)
# Use with FastAPIWebsocketTransport
transport = FastAPIWebsocketTransport(
websocket=websocket,
params=FastAPIWebsocketParams(
audio_in_enabled=True,
audio_out_enabled=True,
vad_analyzer=SileroVADAnalyzer(),
serializer=serializer,
)
)
Hang-up Functionality
When auto_hang_up
is enabled, the serializer will automatically hang up the Twilio call when an EndFrame
or CancelFrame
is processed, using Twilio’s REST API:
# Properly configured with hang-up support
serializer = TwilioFrameSerializer(
stream_sid=stream_sid,
call_sid=call_sid, # Required for auto hang-up
account_sid=os.getenv("TWILIO_ACCOUNT_SID"), # Required for auto hang-up
auth_token=os.getenv("TWILIO_AUTH_TOKEN"), # Required for auto hang-up
)
Server Code Example
Here’s a complete example of handling a Twilio WebSocket connection:
from fastapi import FastAPI, WebSocket
from pipecat.serializers.twilio import TwilioFrameSerializer
import json
import os
app = FastAPI()
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
await websocket.accept()
# Read initial messages from Twilio
start_data = websocket.iter_text()
await start_data.__anext__() # Skip first message
# Parse the second message to get call details
call_data = json.loads(await start_data.__anext__())
# Extract Twilio-specific IDs
stream_sid = call_data["start"]["streamSid"]
call_sid = call_data["start"]["callSid"]
# Create serializer with authentication for auto hang-up
serializer = TwilioFrameSerializer(
stream_sid=stream_sid,
call_sid=call_sid,
account_sid=os.getenv("TWILIO_ACCOUNT_SID"),
auth_token=os.getenv("TWILIO_AUTH_TOKEN"),
)
# Continue with transport and pipeline setup...