gemini_twilio.py
Three jobs: read Twilio's stream, convert audio both ways, and relay Gemini's replies.
from google import genai
from quart import websocket
import json, base64, audioop
class GeminiTwilio:
def __init__(self):
self.client = genai.Client(
vertexai=True, project='YOUR_GCP_PROJECT', location='us-central1')
self.model_id = "gemini-2.0-flash-exp"
self.config = {"response_modalities": ["AUDIO"]}
self.stream_sid = None
self.rate_state = None
async def twilio_audio_stream(self):
"""Yield caller audio as PCM. Twilio sends JSON events."""
while True:
data = json.loads(await websocket.receive())
if data['event'] == 'start':
self.stream_sid = data['start']['streamSid']
elif data['event'] == 'media':
mulaw = base64.b64decode(data['media']['payload'])
yield audioop.ulaw2lin(mulaw, 2)
elif data['event'] == 'stop':
break
def to_mulaw(self, pcm: bytes) -> str:
"""24 kHz PCM from Gemini to 8 kHz base64 mu-law for Twilio."""
pcm, self.rate_state = audioop.ratecv(pcm, 2, 1, 24000, 8000, self.rate_state)
return base64.b64encode(audioop.lin2ulaw(pcm, 2)).decode('utf-8')
async def gemini_websocket(self):
async with self.client.aio.live.connect(
model=self.model_id, config=self.config) as session:
try:
async for response in session.start_stream(
stream=self.twilio_audio_stream(), mime_type='audio/pcm'):
if data := response.data:
await websocket.send(json.dumps({
"event": "media",
"streamSid": self.stream_sid,
"media": {"payload": self.to_mulaw(data)},
}))
finally:
await websocket.close(1000)
To use a Gemini API key instead of Vertex AI, create the client with an API key. Add system instructions or tools to config to give the agent a job.