Twilio <- Pipecat -> Gemini Live

[tpcgold]

pipecat version

0.0.67

Python version

3.13

Operating System

macOS 15.5

Use Case Description

Call Bot - bidirectional LLM allowing interruptions incl. transcription + audio file generation

Current Approach

Combination of
https://github.com/pipecat-ai/pipecat/blob/main/examples/simple-chatbot/server/bot-gemini.py
https://github.com/pipecat-ai/pipecat/blob/main/examples/foundational/26-gemini-multimodal-live.py#L102

And twilio
https://github.com/pipecat-ai/pipecat/tree/main/examples/twilio-chatbot

Errors or Unexpected Behavior

I could get the audio response working if providing the user input inside the OpenAILLMContext initialization.
But it seems audio input is never working.

One problem could be the twilio x-mulaw 8khz -> gemini pcm 16khz. It looks like pipecat doesn't handle this

Additional Context

No response

[markbackman]

Here's an example file using Twilio + Gemini Live: https://github.com/pipecat-ai/pipecat/blob/main/examples/word-wrangler-gemini-live/server/bot_phone_twilio.py.

The example is fairly complicated as the use case includes two Gemini Live agents. Though, you can follow the setup in main() to see how it works.

README with more info too: https://github.com/pipecat-ai/pipecat/blob/main/examples/word-wrangler-gemini-live/README.md.

One problem could be the twilio x-mulaw 8khz -> gemini pcm 16khz. It looks like pipecat doesn't handle this

The TwilioFrameSerializer handles this conversion for you.

For audio input, just use the default sample rate, as you'll need 16khz, which Pipecat supports by default.

[tpcgold]

Hi @markbackman, thanks for the clarification, but it seems that the Word-Wrangler example also yields the same result as before.

Here is a minimal example where I get the audio from the LLM but cannot respond, as it's not reacting to the user's input:

# minimal_twilio_gemini.py
import json
import os
import sys

import uvicorn
from dotenv import load_dotenv
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from loguru import logger
from pipecat.audio.vad.silero import SileroVADAnalyzer
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineParams, PipelineTask
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
from pipecat.processors.filters.stt_mute_filter import (
    STTMuteConfig,
    STTMuteFilter,
    STTMuteStrategy,
)
from pipecat.serializers.twilio import TwilioFrameSerializer
from pipecat.services.gemini_multimodal_live.gemini import (
    GeminiMultimodalLiveLLMService,
)
from pipecat.transports.network.fastapi_websocket import (
    FastAPIWebsocketParams,
    FastAPIWebsocketTransport,
)
from starlette.responses import HTMLResponse

# Configure logger
logger.remove()
logger.add(sys.stderr, level="DEBUG")

# Load environment variables from .env file
load_dotenv(override=True)

# Environment variable checks
GOOGLE_API_KEY = os.getenv("GOOGLE_API_KEY")
TWILIO_ACCOUNT_SID = os.getenv("TWILIO_ACCOUNT_SID")
TWILIO_AUTH_TOKEN = os.getenv("TWILIO_AUTH_TOKEN")

if not GOOGLE_API_KEY:
    logger.error("GOOGLE_API_KEY environment variable not set.")
    sys.exit(1)
if not TWILIO_ACCOUNT_SID or not TWILIO_AUTH_TOKEN:
    logger.error(
        "TWILIO_ACCOUNT_SID or TWILIO_AUTH_TOKEN environment variable not set."
    )
    sys.exit(1)

# FastAPI app
app = FastAPI()


async def bot_logic(websocket: WebSocket):
    """
    Main logic for the Pipecat bot handling Twilio calls.
    """
    try:
        # 1. Read initial messages from Twilio to get call details
        # Twilio sends a 'connected' message first, then a 'start' message with call details
        # This part is crucial for Twilio Media Streams.
        start_data_iter = websocket.iter_text()

        # First message (ignore, it's a connect message)
        connect_msg = await start_data_iter.__anext__()
        logger.debug(f"Received connect message from Twilio: {connect_msg}")

        # Second message contains the call details
        start_msg_json = await start_data_iter.__anext__()
        logger.debug(f"Received start message from Twilio: {start_msg_json}")
        call_data = json.loads(start_msg_json)

        # Ensure this is the start event and extract SIDs
        if call_data.get("event") != "start":
            logger.error(
                f"Expected 'start' event from Twilio, got: {call_data.get('event')}"
            )
            await websocket.close()
            return

        stream_sid = call_data["start"]["streamSid"]
        call_sid = call_data["start"]["callSid"]
        logger.info(f"Call started. CallSid: {call_sid}, StreamSid: {stream_sid}")

        # 2. Initialize TwilioFrameSerializer for handling Twilio's specific format
        # This serializer handles converting Pipecat frames to/from Twilio's expected format (e.g., mulaw)
        # It can also automatically hang up the call using Twilio REST API if credentials are provided.
        twilio_serializer = TwilioFrameSerializer(
            stream_sid=stream_sid,
            call_sid=call_sid,
            account_sid=os.getenv("TWILIO_ACCOUNT_SID"),
            auth_token=os.getenv("TWILIO_AUTH_TOKEN"),
        )

        # 3. Initialize FastAPIWebsocketTransport
        # This transport handles communication over the WebSocket.
        # VAD (Voice Activity Detection) helps in identifying speech segments.
        transport = FastAPIWebsocketTransport(
            websocket=websocket,
            params=FastAPIWebsocketParams(
                audio_in_enabled=True,
                audio_out_enabled=True,
                add_wav_header=False,  # Twilio expects raw mulaw, not WAV
                vad_analyzer=SileroVADAnalyzer(),
                serializer=twilio_serializer,
            ),
        )

        # 4. Initialize Gemini Multimodal Live LLM Service
        # This service handles STT, LLM processing, and TTS using Google Gemini.
        # `transcribe_user_audio=True` enables integrated STT.
        llm = GeminiMultimodalLiveLLMService(
            api_key=GOOGLE_API_KEY,
            system_instruction="You are a friendly and helpful voice assistant. Keep your responses concise for voice.",
            transcribe_user_audio=True,
            # voice_id="en-US-Standard-A" # Optional: specify a Google TTS voice
        )

        # 5. Set up initial context for the conversation (bot speaks first)
        # The bot will say "Hello! How can I help you today?"
        initial_messages = [
            {
                "role": "user",
                "content": "Introduce yourself briefly and ask how you can help.",
            }
        ]
        context = OpenAILLMContext(initial_messages)
        context_aggregator = llm.create_context_aggregator(context)

        # 6. STT Mute Filter
        # This filter mutes the STT input until the bot has finished its first utterance.
        # Useful to prevent the bot from transcribing its own initial message.
        stt_mute_filter = STTMuteFilter(
            config=STTMuteConfig(
                strategies={STTMuteStrategy.MUTE_UNTIL_FIRST_BOT_COMPLETE}
            )
        )

        # 7. Define the pipeline
        # transport.input -> stt_mute_filter -> context_aggregator (user side) -> llm -> transport.output -> context_aggregator (assistant side)
        pipeline = Pipeline(
            [
                transport.input(),
                stt_mute_filter,
                context_aggregator.user(),  # Pushes user messages (including initial prompt) to LLM
                llm,  # Gemini LLM processes and generates response (including TTS)
                transport.output(),  # Sends audio back to Twilio
                context_aggregator.assistant(),  # Captures assistant responses for context
            ]
        )

        # 8. Create and run the pipeline task
        # `audio_out_sample_rate=8000` is crucial for Twilio (expects 8kHz audio).
        task = PipelineTask(
            pipeline,
            params=PipelineParams(
                audio_out_sample_rate=8000,  # Critical for Twilio
                allow_interruptions=True,  # Allow user to interrupt bot
                enable_metrics=True,
                enable_usage_metrics=True,  # Set to False if you don't need usage metrics
            ),
        )

        # Event handler for when the client (Twilio) connects
        @transport.event_handler("on_client_connected")
        async def on_client_connected(transport_instance, client_data):
            logger.info(f"Twilio client connected: {client_data}")
            # Send the initial context to the LLM to make it speak first
            await task.queue_frames([context_aggregator.user().get_context_frame()])

        # Event handler for when the client (Twilio) disconnects
        @transport.event_handler("on_client_disconnected")
        async def on_client_disconnected(transport_instance, client_data):
            logger.info(f"Twilio client disconnected: {client_data}")
            await task.cancel()  # Ensure the pipeline task is cancelled

        await PipelineRunner(handle_sigint=False).run(task)

    except WebSocketDisconnect:
        logger.info("WebSocket disconnected by Twilio.")
    except json.JSONDecodeError as e:
        logger.error(f"Failed to decode JSON from Twilio: {e}")
        if (
            websocket.client_state != websocket.client_state.DISCONNECTED
        ):  # Check if websocket is still open
            await websocket.close(code=1003)  # 1003: "cannot accept data"
    except Exception as e:
        logger.exception(f"Error in bot_logic: {e}")
        if websocket.client_state != websocket.client_state.DISCONNECTED:
            await websocket.close(code=1011)  # 1011: "internal error"
    finally:
        logger.info("Bot logic finished.")


@app.post("/")
async def start_call():
    print("POST TwiML")
    return HTMLResponse(
        content=open("templates/streams.xml").read(), media_type="application/xml"
    )


@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    """
    FastAPI WebSocket endpoint for Twilio Media Streams.
    Twilio will connect to this endpoint.
    """
    await websocket.accept()
    logger.info("Twilio WebSocket connection accepted.")
    await bot_logic(websocket)
    logger.info("Twilio WebSocket connection closed.")


if __name__ == "__main__":
    host = os.getenv("FASTAPI_HOST", "0.0.0.0")
    port = int(
        os.getenv("FASTAPI_PORT", "7001")
    )  # Using a different port than default 8000

    logger.info(f"Starting Uvicorn server on {host}:{port}")
    # Ensure you have:
    # 1. GOOGLE_API_KEY set in your environment or .env file.
    # 2. TWILIO_ACCOUNT_SID and TWILIO_AUTH_TOKEN set for Twilio integration.
    # 3. Ngrok or similar to expose this local endpoint to Twilio if running locally.
    #    Example ngrok command: `ngrok http 7001`
    #    Then configure your Twilio phone number's voice URL to use the WebSocket URL:
    #    `wss://<your-ngrok-subdomain>.ngrok.io/twilio`

    uvicorn.run(app, host=host, port=port)

[tpcgold]

Hmm seems to be a strange issue.
this code works at least on one type of connection (mobile):

# minimal_twilio_gemini.py
import json
import os
import sys

import uvicorn
from dotenv import load_dotenv
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from fastapi.responses import HTMLResponse  # Added for serving XML
from loguru import logger
from pipecat.audio.vad.silero import SileroVADAnalyzer
from pipecat.frames.frames import Frame
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineParams, PipelineTask
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
from pipecat.processors.filters.stt_mute_filter import (
    STTMuteConfig,
    STTMuteFilter,
    STTMuteStrategy,
)
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.serializers.twilio import TwilioFrameSerializer
from pipecat.services.gemini_multimodal_live.gemini import (
    GeminiMultimodalLiveLLMService,
)
from pipecat.services.gemini_multimodal_live.gemini import (
    GeminiMultimodalLiveLLMService,
    GeminiMultimodalModalities,
    Language,
    InputParams,
    GeminiVADParams,
    events,
)
from pipecat.transports.network.fastapi_websocket import (
    FastAPIWebsocketParams,
    FastAPIWebsocketTransport,
)

# Configure logger
logger.remove()
logger.add(sys.stderr, level="DEBUG")  # Ensure DEBUG level for verbose logs

# Load environment variables from .env file
load_dotenv(override=True)


# --- Helper FramePrinter class for debugging (Corrected) ---
class FramePrinter(FrameProcessor):
    def __init__(self, prefix: str = ""):
        super().__init__()
        self._prefix = prefix
        logger.debug(f"Initialized FramePrinter: {self._prefix}")

    async def process_frame(self, frame: Frame, direction: FrameDirection):
        # It's crucial to call the superclass's process_frame.
        # This handles essential base class logic and state.
        await super().process_frame(frame, direction)

        # Now, do our custom logging
        logger.debug(f"{self._prefix} Frame: {frame}, Direction: {direction}")

        # And then push the frame to the next processor in the pipeline
        await self.push_frame(frame, direction)


# --- End Helper FramePrinter class ---

# Environment variable checks
GOOGLE_API_KEY = os.getenv("GOOGLE_API_KEY")
TWILIO_ACCOUNT_SID = os.getenv("TWILIO_ACCOUNT_SID")
TWILIO_AUTH_TOKEN = os.getenv("TWILIO_AUTH_TOKEN")

if not GOOGLE_API_KEY:
    logger.error("GOOGLE_API_KEY environment variable not set.")
    sys.exit(1)
if not TWILIO_ACCOUNT_SID or not TWILIO_AUTH_TOKEN:
    logger.error(
        "TWILIO_ACCOUNT_SID or TWILIO_AUTH_TOKEN environment variable not set."
    )
    sys.exit(1)

# FastAPI app
app = FastAPI()


@app.post("/")
async def start_call():
    logger.info("Received POST request to / - Serving TwiML")
    try:
        templates_dir = os.path.join(os.path.dirname(__file__), "templates")
        xml_file_path = os.path.join(templates_dir, "streams.xml")

        if not os.path.exists(xml_file_path):
            logger.error(f"TwiML file not found at {xml_file_path}")
            return HTMLResponse(
                content="<Response><Say>Error: TwiML configuration file not found.</Say></Response>",
                media_type="application/xml",
                status_code=500,
            )

        with open(xml_file_path, "r") as f:
            twiml_content = f.read()
        return HTMLResponse(content=twiml_content, media_type="application/xml")
    except Exception as e:
        logger.exception("Error serving TwiML")
        return HTMLResponse(
            content=f"<Response><Say>Error serving TwiML: {e}</Say></Response>",
            media_type="application/xml",
            status_code=500,
        )


async def bot_logic(websocket: WebSocket):
    """
    Main logic for the Pipecat bot handling Twilio calls.
    """
    try:
        start_data_iter = websocket.iter_text()
        connect_msg = await start_data_iter.__anext__()
        logger.debug(f"Received connect message from Twilio: {connect_msg}")

        start_msg_json = await start_data_iter.__anext__()
        logger.debug(f"Received start message from Twilio: {start_msg_json}")
        call_data = json.loads(start_msg_json)

        if call_data.get("event") != "start":
            logger.error(
                f"Expected 'start' event from Twilio, got: {call_data.get('event')}"
            )
            await websocket.close()
            return

        stream_sid = call_data["start"]["streamSid"]
        logger.info(
            f"Call started. CallSid from event: {call_data['start']['callSid']}, StreamSid: {stream_sid}"
        )

        # Initialize TwilioFrameSerializer
        # Parameters 'call_sid' and 'auto_hang_up' were removed as they caused TypeErrors with pipecat-ai 0.0.67
        twilio_serializer = TwilioFrameSerializer(
            stream_sid=stream_sid,
            account_sid=TWILIO_ACCOUNT_SID,
            auth_token=TWILIO_AUTH_TOKEN,
        )
        logger.info("TwilioFrameSerializer initialized.")

        transport = FastAPIWebsocketTransport(
            websocket=websocket,
            params=FastAPIWebsocketParams(
                audio_in_enabled=True,
                audio_out_enabled=True,
                add_wav_header=False,
                vad_analyzer=None,  # SileroVADAnalyzer(),
                serializer=twilio_serializer,
            ),
        )

        llm_params = InputParams(
            temperature=0.7,  # Set model input params
            # language=Language.EN_US,  # Set language (30+ languages supported)
            modalities=GeminiMultimodalModalities.AUDIO,  # Response modality
            vad=GeminiVADParams(
                start_sensitivity=events.StartSensitivity.HIGH,  # Detect speech quickly
                end_sensitivity=events.EndSensitivity.LOW,  # Allow longer pauses
                prefix_padding_ms=300,  # Keep 300ms before speech
                silence_duration_ms=1000,  # End turn after 1s silence
            ),
        )

        llm = GeminiMultimodalLiveLLMService(
            api_key=GOOGLE_API_KEY,
            voice_id="Aoede",
            system_instruction="You are a friendly and helpful voice assistant. Keep your responses concise for voice.",
            transcribe_user_audio=True,  # This enables Gemini to do STT
            transcribe_model_audio=True,  # Auf False gesetzt
            params=llm_params,
        )

        initial_messages = [
            {
                "role": "user",
                "content": "Introduce yourself briefly and ask how you can help.",
            }
        ]
        context = OpenAILLMContext(initial_messages)
        context_aggregator = llm.create_context_aggregator(context)

        stt_mute_filter = STTMuteFilter(
            config=STTMuteConfig(
                strategies={STTMuteStrategy.MUTE_UNTIL_FIRST_BOT_COMPLETE}
            )
        )

        pipeline = Pipeline(
            [
                transport.input(),
                FramePrinter(
                    "[AUDIO IN] After Transport Input"
                ),  # Log frames coming from Twilio
                stt_mute_filter,
                FramePrinter(
                    "[AUDIO IN] After STT Mute Filter"
                ),  # Log frames after mute filter
                context_aggregator.user(),  # This will receive InputAudioRawFrame if transcribe_user_audio=True on LLM
                FramePrinter(
                    "[LLM IN] After Context Aggregator User / Before LLM"
                ),  # Log frames going to LLM
                llm,
                FramePrinter(
                    "[LLM OUT] After LLM / Before TTS"
                ),  # Log frames (like LLMTextFrame) from LLM
                transport.output(),
                FramePrinter(
                    "[ASSISTANT CONTEXT] After Transport Output"
                ),  # Log frames going to assistant context
                context_aggregator.assistant(),
            ]
        )
        logger.info("Pipeline created with FramePrinters.")

        task = PipelineTask(
            pipeline,
            params=PipelineParams(
                audio_out_sample_rate=8000,
                allow_interruptions=True,
                enable_metrics=True,
                enable_usage_metrics=True,
            ),
        )

        @transport.event_handler("on_client_connected")
        async def on_client_connected(transport_instance, client_data):
            logger.info(f"Twilio client connected to WebSocket: {client_data}")
            # Kick off the conversation by sending the initial prompt to the LLM
            await task.queue_frames([context_aggregator.user().get_context_frame()])
            logger.info("Initial prompt sent to LLM.")

        @transport.event_handler("on_client_disconnected")
        async def on_client_disconnected(transport_instance, client_data):
            logger.info(f"Twilio client disconnected from WebSocket: {client_data}")
            await task.cancel()

        await PipelineRunner(handle_sigint=False).run(task)

    except WebSocketDisconnect:
        logger.info("WebSocket disconnected by Twilio.")
    except json.JSONDecodeError as e:
        logger.error(f"Failed to decode JSON from Twilio: {e}")
        if websocket.client_state != websocket.client_state.DISCONNECTED:
            await websocket.close(code=1003)
    except Exception as e:
        logger.exception(f"Error in bot_logic: {e}")
        if websocket.client_state != websocket.client_state.DISCONNECTED:
            await websocket.close(code=1011)
    finally:
        logger.info("Bot logic finished.")


@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    logger.info("Twilio WebSocket connection accepted at /ws.")
    await bot_logic(websocket)
    logger.info("Twilio WebSocket connection closed at /ws.")


if __name__ == "__main__":
    host = os.getenv("FASTAPI_HOST", "0.0.0.0")
    port = int(os.getenv("FASTAPI_PORT", "7001"))

    logger.info(f"Starting Uvicorn server on {host}:{port}")
    logger.info(
        "Ensure your Twilio phone number is configured to POST to http://<your-ngrok-url>/"
    )
    logger.info(
        "And your templates/streams.xml file points its <Stream> URL to wss://<your-ngrok-url>/ws"
    )

    templates_dir = os.path.join(os.path.dirname(__file__), "templates")
    if not os.path.exists(templates_dir):
        os.makedirs(templates_dir)
    xml_file_path = os.path.join(templates_dir, "streams.xml")
    if not os.path.exists(xml_file_path):
        logger.warning(
            f"{xml_file_path} not found. Creating a placeholder. Please update it with your ngrok URL."
        )
        with open(xml_file_path, "w") as f:
            f.write('<?xml version="1.0" encoding="UTF-8"?>\n')
            f.write("<Response>\n")
            f.write("    <Connect>\n")
            # Make sure this URL is correct and uses /ws
            f.write(
                '        <Stream url="wss://YOUR_NGROK_SUBDOMAIN_HERE.ngrok-free.app/ws"></Stream>\n'
            )
            f.write("    </Connect>\n")
            f.write("    <Say>Connecting to assistant.</Say>\n")
            f.write('    <Pause length="40"/>\n')
            f.write("</Response>\n")

    uvicorn.run(app, host=host, port=port)

• Calling via MS Team -> twilio -> pipcat -> gemini ❌
• Calling via Mobile Phone -> twilio -> pipcat -> gemini ✅
• Calling MS Teams <-> Mobile Phone ✅

so just in case the call happens via MS teams it does not work 🤔
seems there must be some kind of difference in the raw audio stream

[markbackman]

Ok, I'm glad that works for the two cases, as that matches my expectation.

Regarding:

Calling via MS Team -> twilio -> pipcat -> gemini

I'm not familiar with that interface, so it's probably worth figuring out what is different about the MediaStreams in this case. Please let us know what you find out.

[tpcgold]

Hi,
I'll report back as soon as we have more information.
It's just a landline connection via Teams through an SBC and SIP trunk, or possibly a PSTN gateway. (Telephony really is a hell of a rabbit hole.)
Theoretically, it shouldn't make a difference — but it seems like it does.

[tpcgold]

Hi there, I was able to resolve the issue. I got a statement from Twilio: they made a change on 13th May, but they had to roll it back. Now it also works via MS Teams again.