import asyncio
import json
import logging
import os
import re
from typing import Optional
import websockets
from websockets import ConnectionClosed
from dotenv import load_dotenv
from google import genai
from google.genai import types
from google.genai import errors as genai_errors
from google.genai.live import AsyncSession
from agentduet import (
SessionManager,
SessionManagerConfig,
Call,
CallAudioConfig,
IncomingCallNotification,
BufferFullError,
CallClosedError,
new_session_id,
)
load_dotenv()
logging.basicConfig(
level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)
logging.getLogger("agentduet").setLevel(logging.INFO)
# Call and TTS synthesis match at 24 kHz mono PCM
SAMPLE_RATE = 24000
genai_client = genai.Client(vertexai=False, api_key=os.getenv("GEMINI_API_KEY"))
MODEL = "models/gemini-2.5-flash"
CONFIG = types.LiveConnectConfig(
response_modalities=[types.Modality.AUDIO],
output_audio_transcription=types.AudioTranscriptionConfig(),
system_instruction=(
"You are a professional voice assistant on a telephone call. "
"Keep your replies concise and conversational."
),
)
class SentenceBuffer:
"""Accumulates streamed transcript deltas and emits complete sentences."""
_BOUNDARY = re.compile(r".*?[.!?\n]+", re.S)
def __init__(self) -> None:
self._buf = ""
def add(self, text: str) -> list[str]:
self._buf += text
sentences: list[str] = []
while True:
match = self._BOUNDARY.match(self._buf)
if not match:
break
sentence = match.group().strip()
self._buf = self._buf[match.end():]
if sentence:
sentences.append(sentence)
return sentences
def flush(self) -> Optional[str]:
trailing = self._buf.strip()
self._buf = ""
return trailing or None
class TtsSpeaker:
"""Streams sentences to the voice cloning server and plays audio to the call."""
def __init__(self, call: Call, voice_id: str, server_url: str) -> None:
self._call = call
self._uri = f"{server_url.rstrip('/')}/tts/{voice_id}/{SAMPLE_RATE}"
self._queue: asyncio.Queue[str] = asyncio.Queue()
self._ws: Optional[websockets.WebSocketClientProtocol] = None
self._drop = False
self._consumer_task: Optional[asyncio.Task] = None
async def __aenter__(self) -> "TtsSpeaker":
try:
self._ws = await websockets.connect(self._uri)
logger.info("Connected to Voice Cloning TTS server at %s", self._uri)
except Exception:
logger.exception("Could not connect to Voice Cloning TTS server at %s", self._uri)
self._ws = None
self._consumer_task = asyncio.create_task(self._consume())
return self
async def __aexit__(self, *_exc) -> None:
await self.close()
def enqueue(self, sentence: str) -> None:
self._queue.put_nowait(sentence)
def interrupt(self) -> None:
"""Barge-in: drop queued sentences and discard remaining in-flight audio."""
while not self._queue.empty():
try:
self._queue.get_nowait()
except asyncio.QueueEmpty:
break
self._drop = True
async def _consume(self) -> None:
try:
while True:
sentence = await self._queue.get()
try:
await self._speak(sentence)
except ConnectionClosed:
logger.warning("TTS socket closed; agent will stay silent")
self._ws = None
except Exception:
logger.exception("Error synthesizing sentence")
except asyncio.CancelledError:
raise
async def _speak(self, sentence: str) -> None:
if self._ws is None:
return
self._drop = False
await self._ws.send(json.dumps({"text": sentence}))
async for message in self._ws:
if isinstance(message, bytes):
if self._drop:
continue
try:
await self._call.send_audio(message)
except BufferFullError:
logger.warning("Call buffer full; dropping audio chunk")
else:
event = json.loads(message)
kind = event.get("event")
if kind == "done":
break
if kind == "error":
logger.error("TTS server error: %s", event.get("message"))
break
async def close(self) -> None:
if self._consumer_task:
self._consumer_task.cancel()
try:
await self._consumer_task
except asyncio.CancelledError:
pass
if self._ws:
await self._ws.close()
self._ws = None
class PhoneAgent:
"""Bridges the AgentDuet call and Gemini Live through the cloned voice TTS."""
def __init__(
self,
call: Call,
gemini_session: AsyncSession,
voice_id: str,
tts_server_url: str,
) -> None:
self._call = call
self._gemini_session = gemini_session
self._voice_id = voice_id
self._tts_server_url = tts_server_url
self._sentences = SentenceBuffer()
self._speaker: Optional[TtsSpeaker] = None
self._send_task: Optional[asyncio.Task] = None
self._recv_task: Optional[asyncio.Task] = None
async def _on_hangup(self, _evt) -> None:
logger.info("Call terminated, cleaning up resources")
try:
await self._gemini_session.close()
except Exception:
pass
for task in (self._send_task, self._recv_task):
if task:
task.cancel()
try:
await task
except (asyncio.CancelledError, genai_errors.APIError):
pass
if self._speaker:
await self._speaker.close()
async def run(self) -> None:
self._call.on_hangup(self._on_hangup)
async with TtsSpeaker(self._call, self._voice_id, self._tts_server_url) as speaker:
self._speaker = speaker
self._send_task = asyncio.create_task(self._stream_to_gemini())
self._recv_task = asyncio.create_task(self._receive_from_gemini())
await asyncio.gather(
self._send_task, self._recv_task, return_exceptions=True
)
async def _stream_to_gemini(self) -> None:
"""Stream caller audio to Gemini Live."""
try:
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=f"audio/pcm;rate={SAMPLE_RATE}"
)
)
except (ConnectionClosed, CallClosedError):
pass
except Exception:
logger.exception("Error streaming caller audio to Gemini")
raise
async def _receive_from_gemini(self) -> None:
"""Receive Gemini's transcript and synthesize speech in the cloned voice."""
try:
while True:
async for response in self._gemini_session.receive():
server_content = response.server_content
if not server_content:
continue
# Handle caller interruption
if server_content.interrupted:
logger.info("Caller interrupted - halting agent speech")
if self._speaker:
self._speaker.interrupt()
await self._call.clear_send_audio_buffer()
self._sentences.flush()
continue
transcription = server_content.output_transcription
if transcription and transcription.text:
for sentence in self._sentences.add(transcription.text):
if self._speaker:
self._speaker.enqueue(sentence)
if server_content.turn_complete:
trailing = self._sentences.flush()
if trailing and self._speaker:
self._speaker.enqueue(trailing)
except (ConnectionClosed, genai_errors.APIError):
logger.info("Gemini live session closed")
except asyncio.CancelledError:
raise
except CallClosedError:
logger.info("Call closed, stopping receiver")
except Exception:
logger.exception("Error receiving transcript from Gemini")
raise
async def main() -> None:
voice_id = os.getenv("TTS_VOICE_ID")
if not voice_id:
raise SystemExit(
"TTS_VOICE_ID is required. Register a voice via /tts/upload first."
)
tts_server_url = os.getenv("TTS_SERVER_URL", "ws://localhost:8000")
config = SessionManagerConfig.create(
api_key=os.getenv("AGENTDUET_API_KEY"),
connector_uuid=os.getenv("AGENTDUET_CONNECTOR_UUID"),
call_audio=CallAudioConfig(
sample_rate=SAMPLE_RATE,
buffer_size=1024 * 1024,
),
)
async with SessionManager(config) as sm:
logger.info("SessionManager started and listening for incoming calls")
@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("Answering call from %s", 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("Failed to answer call %s", call.id)
return
agent = PhoneAgent(call, gemini_session, voice_id, tts_server_url)
await agent.run()
except Exception:
logger.exception("Error in phone agent")
await call.close()
raise
await sm.run_forever()
if __name__ == "__main__":
asyncio.run(main())
