​
Overview

ExotelFrameSerializer enables integration with Exotel’s WebSocket media streaming protocol, allowing your Pipecat application to handle phone calls via Exotel’s voice services.

​
Features

  • Bidirectional audio conversion between Pipecat and Exotel
  • DTMF (touch-tone) event handling

​
Installation

The ExotelFrameSerializer does not require any additional dependencies beyond the core Pipecat library.

​
Configuration

​
Constructor Parameters

​
stream_id
str
required
The Stream ID for Exotel
​
call_sid
Optional[str]
default:"None"
The associated Exotel Call SID.
​
params
InputParams
default:"InputParams()"
Configuration parameters

​
InputParams Configuration

​
exotel_sample_rate
int
default:"8000"
Sample rate used by Exotel (typically 8kHz)
​
sample_rate
int | None
default:"None"
Optional override for pipeline input sample rate

​
Basic Usage

from pipecat.serializers.exotel import ExotelFrameSerializer
from pipecat.transports.network.fastapi_websocket import (
    FastAPIWebsocketTransport,
    FastAPIWebsocketParams
)

# Extract required values from Exotel WebSocket connection
stream_id = call_data["stream_id"]
call_sid = call_data["start"]["call_sid"]

# Create serializer
serializer = ExotelFrameSerializer(
    stream_id=stream_id,
    call_sid,
)

# Use with FastAPIWebsocketTransport
transport = FastAPIWebsocketTransport(
    websocket=websocket,
    params=FastAPIWebsocketParams(
        audio_in_enabled=True,
        audio_out_enabled=True,
        vad_analyzer=SileroVADAnalyzer(),
        serializer=serializer,
    )
)

​
Server Code Example

Here’s a complete example of handling a Exotel WebSocket connection: 
from fastapi import FastAPI, WebSocket
from pipecat.serializers.exotel import ExotelFrameSerializer
import json
import os

app = FastAPI()

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()

    # Read initial messages from Exotel
    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 Exotel-specific IDs and encoding
    stream_id = call_data["stream_id"]
    call_sid = call_data["start"]["call_sid"]

    # Create serializer with API key for auto hang-up
    serializer = ExotelFrameSerializer(
        stream_id=stream_id,
        call_sid=call_sid,
    )

    # Continue with transport and pipeline setup...