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())
