> For clean Markdown of any page, append .md to the page URL. > For a complete documentation index, see https://docs.agentduet.com/llms.txt. > For AI client integration (Claude Code, Cursor, etc.), connect to the MCP server at https://docs.agentduet.com/_mcp/server. # Gemini Live This guide shows how to integrate the AgentDuet SDK with the [Google GenAI SDK](https://ai.google.dev/gemini-api/docs/live) for bidirectional real-time audio streaming with Gemini's native audio model. ## What this integration provides A phone agent that answers incoming calls, streams caller audio to Gemini Live at 24 kHz PCM, plays Gemini audio back on the call, clears the outbound buffer on barge-in, and tears down the Live session on hangup. ## Prerequisites - Python **3.12+** - Get an API key and connector UUID at [agentduet.com](https://agentduet.com) - Gemini API key ([Google AI Studio](https://aistudio.google.com/apikey)) ```bash pip install "agentduet==1.0.0" google-genai python-dotenv ``` ## Step 1: Create a project folder ```bash mkdir agentduet-gemini-live-bridge && cd $_ python3.12 -m venv .venv && source .venv/bin/activate pip install "agentduet==1.0.0" google-genai python-dotenv ``` ## Step 2: Configure `.env` ```bash cat > .env << 'EOF' AGENTDUET_API_KEY=your-connector-api-key AGENTDUET_CONNECTOR_UUID=your-connector-uuid GEMINI_API_KEY=your-gemini-key EOF ``` ## Step 3: Write `gemini_live_bridge.py` Create the file next to `.env`: ```python import asyncio import logging import os from typing import Optional from dotenv import load_dotenv from google import genai from google.genai import errors as genai_errors from google.genai import types from google.genai.live import AsyncSession from websockets import ConnectionClosed from agentduet import ( BufferFullError, Call, CallAudioConfig, CallClosedError, IncomingCallNotification, SessionManager, SessionManagerConfig, new_session_id, ) load_dotenv() logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", ) logger = logging.getLogger(__name__) genai_client = genai.Client(vertexai=False, api_key=os.getenv("GEMINI_API_KEY")) MODEL = "models/gemini-3.1-flash-live-preview" CONFIG = types.LiveConnectConfig( response_modalities=[types.Modality.AUDIO], speech_config=types.SpeechConfig( voice_config=types.VoiceConfig( prebuilt_voice_config=types.PrebuiltVoiceConfig(voice_name="Zephyr") ) ), system_instruction="You are a helpful and friendly AI assistant.", ) class GeminiLiveIntegration: def __init__(self, call: Call, gemini_session: AsyncSession): self._call = call self._gemini_session = gemini_session self._send_to_gemini_task: Optional[asyncio.Task] = None self._recv_from_gemini_task: Optional[asyncio.Task] = None self._terminated = False async def _on_hangup(self, evt): logger.info("Call terminated - closing Gemini session") self._terminated = True try: await self._gemini_session.close() except Exception: logger.exception("Error closing Gemini session") for task in (self._send_to_gemini_task, self._recv_from_gemini_task): if task: task.cancel() try: await task except (asyncio.CancelledError, genai_errors.APIError): pass async def run(self): self._call.on_hangup(self._on_hangup) self._send_to_gemini_task = asyncio.create_task(self.stream_to_gemini()) self._recv_from_gemini_task = asyncio.create_task( self.receive_audio_from_gemini() ) await asyncio.gather( self._send_to_gemini_task, self._recv_from_gemini_task, return_exceptions=True, ) async def stream_to_gemini(self): try: await self._gemini_session.send_realtime_input(text="Hello, how are you?") async for audio_chunk in self._call.caller.audio_stream(): await self._gemini_session.send_realtime_input( audio=types.Blob( data=audio_chunk, mime_type="audio/pcm;rate=24000" ) ) except ConnectionClosed: pass except CallClosedError: logger.debug("Call closed; stopping stream to Gemini") except asyncio.CancelledError: raise except Exception: logger.exception("Error in stream to Gemini") raise async def receive_audio_from_gemini(self): try: while True: async for response in self._gemini_session.receive(): if server_content := response.server_content: if server_content.interrupted: # Public barge-in API - clears queued outbound PCM await self._call.clear_send_audio_buffer() logger.debug("Gemini interrupted - cleared send buffer") break if model_turn := server_content.model_turn: for part in model_turn.parts: if part.inline_data and isinstance( part.inline_data.data, bytes ): try: await self._call.send_audio( part.inline_data.data ) except BufferFullError: logger.warning( "Send buffer full - drop chunk or " "raise CallAudioConfig.buffer_size" ) if part.text is not None: logger.debug("Text: %s", part.text) except (ConnectionClosed, genai_errors.APIError): logger.debug("Gemini session closed") except asyncio.CancelledError: raise except CallClosedError: logger.debug("Call closed; stopping stream from Gemini") except Exception: logger.exception("Error in stream from Gemini") raise async def main(): config = SessionManagerConfig.create( api_key=os.getenv("AGENTDUET_API_KEY"), connector_uuid=os.getenv("AGENTDUET_CONNECTOR_UUID"), call_audio=CallAudioConfig( sample_rate=24000, buffer_size=1024 * 1024, ), ) async with SessionManager(config) as sm: logger.info("SessionManager started %s", sm.id) @sm.on_incoming_call async def on_call(noti: IncomingCallNotification): session = await sm.open_session(new_session_id(), noti.subscriber) call = await session.process_call(noti) logger.info("Incoming call=%s caller=%s", call.id, call.caller) try: async with genai_client.aio.live.connect( model=MODEL, config=CONFIG ) as gemini_session: result = await call.answer() if not result: logger.error( "Answer failed %s: %s (%s)", call.id, result.error_message, result.error_code, ) return await GeminiLiveIntegration(call, gemini_session).run() except Exception: logger.exception("Error in Gemini Live bridge") await call.close() raise await sm.run_forever() if __name__ == "__main__": asyncio.run(main()) ``` ## Step 4: Run the bridge ```bash python gemini_live_bridge.py ``` Keep the process running. Confirm logs show the SessionManager connected and waiting for calls. ## Step 5: Call your number Dial the connector phone number. Verify two-way audio, then interrupt mid-reply to confirm barge-in clears playback. Hang up and confirm the process remains ready for the next call. ## How the pieces fit `SessionManager` receives the inbound call → you `open_session` + `process_call` → `answer()` → two tasks move PCM between AgentDuet and the model. Hangup closes the model session and cancels those tasks. Telephony stays AgentDuet's job; the model never sees SIP. ## Notes - **Sample rate:** Gemini Live expects 24 kHz mono PCM. Keep `CallAudioConfig.sample_rate=24000`. - **Interrupt:** Always use `await call.clear_send_audio_buffer()` when `server_content.interrupted` is set. Private methods like `_interrupt` are not part of the public API. - **`receive()` is turn-scoped:** Re-enter the `async for` loop after each turn (or after an interrupt break) so the agent stays live for the whole call. - **`BufferFullError`:** Raise `buffer_size` or drop chunks under load; do not block the Gemini receive loop indefinitely. - **Hangup order:** Close Gemini first in `on_hangup`, then cancel tasks so `receive()` exits cleanly. ## Related - [Google ADK](/integrations/google-adk) - [Amazon Nova Sonic](/integrations/amazon-nova-sonic) - [OpenAI Realtime](/integrations/open-ai-realtime) - [Audio Streaming](/concepts/audio-streaming)