[Python] Google Gemini Live API never sends a response to audio queries sent via Twilio

I'm making an application that relays data from a Twilio websocket to Gemini's live API, and viceversa. This is part of a vocal chatbot I'm building that lets you talk to Gemini via a phonecall.

The issue is that despite me sending the vocal data from Twilio to Gemini, converting Twilio's 8khz Mulaw output to Gemini's 16khz PCM input, I never get a response from Gemini.

This is the code I've written for now. It's heavily based on the sample you can find here, but I got rid of the text and video components and modified the audio component so that the audio comes from Twilio instead of the computer's mic.

import asyncio
import base64
import io
import logging
import os
import sys
import traceback
from uuid import uuid4

import dotenv
import pyaudio

import uvicorn
from fastapi import FastAPI, Request, Response, HTTPException
from google import genai
from google.genai.live import AsyncSession
from pydantic import BaseModel
from pydub import AudioSegment
from starlette.websockets import WebSocket
from twilio.base.exceptions import TwilioException

from twilio.rest import Client as TwilioClient
from twilio.twiml.voice_response import VoiceResponse


dotenv.load_dotenv()

if sys.version_info < (3, 11, 0):
import taskgroup, exceptiongroup

asyncio.TaskGroup = taskgroup.TaskGroup
asyncio.ExceptionGroup = exceptiongroup.ExceptionGroup

FORMAT = pyaudio.paInt16
CHANNELS = 1
SEND_SAMPLE_RATE = 16000
RECEIVE_SAMPLE_RATE = 24000
CHUNK_SIZE = 1024

TWILIO_SAMPLE_RATE = 8000

MODEL = "gemini-2.5-flash-preview-native-audio-dialog"

DEFAULT_MODE = "camera"

client = genai.Client(http_options={"api_version": "v1beta"})

CONFIG = {"response_modalities": ["AUDIO"]}

pya = pyaudio.PyAudio()

app = FastAPI()

# Twilio configuration
TWILIO_ACCOUNT_SID = os.getenv("TWILIO_ACCOUNT_SID")
TWILIO_AUTH_TOKEN = os.getenv("TWILIO_AUTH_TOKEN")
TWILIO_PHONE_NUMBER = os.getenv("TWILIO_PHONE_NUMBER")
BASE_URL = os.getenv('BASE_URL')

# Initialize Twilio client
twilio_client = TwilioClient(TWILIO_ACCOUNT_SID, TWILIO_AUTH_TOKEN)


def mulaw_decode(mulaw_bytes: bytes) -> bytes:
byte_stream = io.BytesIO(mulaw_bytes)
return AudioSegment.from_file(
byte_stream,
format='raw',
codec='mulaw',
sample_width=1,
frame_rate=TWILIO_SAMPLE_RATE,
channels=CHANNELS,
).set_frame_rate(SEND_SAMPLE_RATE).raw_data

def mulaw_encode(pcm_bytes: bytes) -> bytes:
byte_stream = io.BytesIO(pcm_bytes)
return AudioSegment.from_file(
byte_stream,
format='raw',
codec='pcm',
sample_width=2,
frame_rate=RECEIVE_SAMPLE_RATE,
channels=CHANNELS,
).set_sample_width(1).set_frame_rate(TWILIO_SAMPLE_RATE).raw_data

logging.basicConfig(
level=os.getenv("LOGGING_LEVEL", "INFO"),
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

class OutboundCallRequest(BaseModel):
phone_number: str

class AudioLoop:
def __init__(self):

self.audio_in_queue = asyncio.Queue()
self.out_queue = asyncio.Queue(maxsize=5)

self.session: AsyncSession | None = None

self.send_text_task = None
self.receive_audio_task = None
self.play_audio_task = None
self.websocket: WebSocket | None = None
self.stream_sid: str | None = None

async def send_realtime(self):
while True:
msg = await self.out_queue.get()
mulaw_bytes = base64.b64decode(msg['data']['payload'])
logging.debug("Got Mulaw: " + str(mulaw_bytes))
pcm_bytes = mulaw_decode(mulaw_bytes)
logging.debug("Sending PCM: " + str(pcm_bytes))

await self.session.send_realtime_input(audio=genai.types.Blob(data=buffer, mime_type="audio/pcm"))


async def listen_audio(self):
while True:
data = await self.websocket.receive_json()
if data['event'] == 'start':
self.stream_sid = data['streamSid']
elif data['event'] == 'media':
await self.out_queue.put({"data": data['media']})

async def receive_audio(self):
"Background task to reads from the websocket and write pcm chunks to the output queue"
while True:
turn = self.session.receive()
async for response in turn:
if data := response.data:
self.audio_in_queue.put_nowait(data)
continue
if text := response.text:
print(text, end="")

# If you interrupt the model, it sends a turn_complete.
# For interruptions to work, we need to stop playback.
# So empty out the audio queue because it may have loaded
# much more audio than has played yet.
while not self.audio_in_queue.empty():
self.audio_in_queue.get_nowait()

async def send_audio_twilio(self):
while True:
pcm_bytes = await self.audio_in_queue.get()
mulaw_bytes = mulaw_encode(pcm_bytes)

if self.stream_sid is None:
logging.error("stream_sid is None, this probably means that you're trying to send data to Twilio before the 'start' event from the WebSocket")

logging.debug("Sending mulaw data to Twilio stream with sid " + self.stream_sid)
await self.websocket.send_json({
"event": "media",
"streamSid": self.stream_sid,
"media": {
"payload": str(base64.b64encode(mulaw_bytes))
}
})


async def run(self):
if self.websocket is None:
logger.error("Websocket not initialized, stopping.")
return

try:
async with (
client.aio.live.connect(model=MODEL, config=CONFIG) as session,
asyncio.TaskGroup() as tg,
):
self.session = session

# send_text_task = tg.create_task(self.send_text())
tg.create_task(self.listen_audio())
tg.create_task(self.send_realtime())
tg.create_task(self.receive_audio())
await tg.create_task(self.send_audio_twilio())

# await send_text_task
raise asyncio.CancelledError("User requested exit")

except asyncio.CancelledError:
pass
except ExceptionGroup as EG:
await self.websocket.close()
traceback.print_exception(EG)

async def init_websocket(self, websocket: WebSocket):
await websocket.accept()
self.websocket = websocket


active_conversations = dict[str, AudioLoop]()

@app.post("/webhook/voice")
async def handle_incoming_call(request: Request):
"""Handle incoming Twilio voice calls"""
form_data = await request.form()
call_sid = form_data.get("CallSid")

logger.info(f"Received incoming call: {call_sid}")

# Create conversation ID
conversation_id = str(uuid4())
active_conversations[conversation_id] = AudioLoop()

logger.info(f"Created conversation: {conversation_id}")
logger.info(f"Active conversations after creation: {list(active_conversations.keys())}")

# Create TwiML response
response = VoiceResponse()

# Transition directly to WebSocket for real-time audio streaming
connect = response.connect()

# Build WebSocket URL - ensure it's accessible from Twilio
base_url = os.getenv('BASE_URL', f"{request.base_url.scheme}://{request.base_url.hostname}:{request.base_url.port}")
websocket_url = f"{base_url.replace('http://', 'ws://').replace('https://', 'wss://')}/ws/{conversation_id}"

connect.stream(url=websocket_url)

logger.info(f"Generated TwiML for conversation: {conversation_id} with WebSocket URL: {websocket_url}")

return Response(content=str(response), media_type="application/xml")

@app.post("/webhook/status")
async def handle_call_status(request: Request):
"""Handle Twilio call status updates"""
form_data = await request.form()
call_sid = form_data.get("CallSid")
call_status = form_data.get("CallStatus")

logger.info(f"Call {call_sid} status: {call_status}")

if call_status in ["completed", "failed", "busy", "no-answer"]:
# Clean up any associated conversations
# You might want to implement conversation cleanup logic here
pass

return {"status": "received"}


@app.post("/api/start-call")
async def start_outbound_call_json(request: OutboundCallRequest):
"""Start an outbound call using Twilio (JSON body version)"""
try:
# Validate phone number format
if not request.phone_number or not request.phone_number.startswith('+'):
raise HTTPException(status_code=400, detail="Phone number must be in E.164 format (e.g., +1234567890)")

# Create the call
call = twilio_client.calls.create(
url=f"{os.getenv('BASE_URL', 'http://localhost:8000')}/webhook/voice",
to=request.phone_number,
from_=TWILIO_PHONE_NUMBER,
status_callback=f"{os.getenv('BASE_URL', 'http://localhost:8000')}/webhook/status",
status_callback_event=['initiated', 'ringing', 'answered', 'completed']
)

logger.info(f"Outbound call initiated: {call.sid} to {request.phone_number}")
return {
"call_sid": call.sid,
"status": "initiated",
"to": request.phone_number,
"from": TWILIO_PHONE_NUMBER,
"message": "Call is being initiated"
}
except TwilioException as e:
logger.error(f"Failed to initiate call to {request.phone_number}: {e}")
raise HTTPException(status_code=400, detail=str(e))
except Exception as e:
logger.error(f"Unexpected error initiating call: {e}")
raise HTTPException(status_code=500, detail="Internal server error")


@app.websocket("/ws/{conversation_id}")
async def websocket_endpoint(websocket: WebSocket, conversation_id: str):
if conversation_id not in active_conversations:
await websocket.close(code=404, reason="Conversation not found")
return

conversation = active_conversations[conversation_id]
await conversation.init_websocket(websocket)

await conversation.run()


if __name__ == "__main__":
uvicorn.run(
"ai_caller_twilio:app",
host="0.0.0.0",
port=8000,
reload=True
)



In this case, the async for response in turn: in the receive_audio function inside of AudioLoop never yields anything. No errors, or anything interesting, gets printed to the console, even if putting the logging level to DEBUG. Judging from these messages the audio chunks seem to be sent correctly:

2025-08-22 22:13:51,130 - websockets.client - DEBUG - > TEXT '{"realtime_input": {"audio": {"data": "_v7-_v7-...e_type": "audio/pcm"}}}' [1773 bytes]


To use the application you need to set the following env variables: GEMINI_API_KEY, TWILIO_ACCOUNT_SID, TWILIO_AUTH_TOKEN, TWILIO_PHONE_NUMBER, OUTGOING_TWILIO_PHONE_NUMBER, BASE_URL (the last one is the application's public URL, I get it from ngrok but you can use any method that exposes the application to the internet so that Twilio can call the webhooks). You can initiate a call using the POST /api/start-call REST endpoint.

I tried debugging the code using Python's debugger, but can't really find a reason why the code doesn't work. I also tried to make a "buffer" for the data that's sent to Gemini, but to no avail.

Continue reading...