> 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. # Voice Cloning This guide shows how to integrate the AgentDuet SDK with the [`voice-cloning-tts`](https://github.com/AgentDuet/voice-cloning-tts) streaming server to run real-time voice-cloned AI phone agents. ## What this integration provides A real-time voice cloning architecture that allows your phone agent to speak in any custom cloned voice generated from a short reference audio sample. ### Architecture Overview ``` Caller Audio ──▶ AgentDuet Call ──▶ Gemini Live (Listening & Reasoning) │ ▼ Transcript deltas SentenceBuffer (Complete sentences) │ ▼ Text chunks Voice Cloning TTS Server (Pocket-TTS) ws://localhost:8000/tts/{voice_id}/24000 │ ▼ Cloned PCM audio chunks call.send_audio() ──▶ Caller ``` - **Speech-to-Text & Reasoning:** Gemini Live acts as the agent's ears and brain, processing caller audio in real time. - **Voice-Cloning Mouth:** The self-hosted [`voice-cloning-tts`](https://github.com/AgentDuet/voice-cloning-tts) server synthesizes Gemini's transcript in your cloned voice profile. - **Low-Latency Streaming:** Synthesized 24 kHz int16 PCM audio streams chunk-by-chunk over a persistent WebSocket directly into `call.send_audio()`. - **Zero-Latency Barge-in:** When the caller interrupts, Gemini emits an interruption signal. The agent immediately drops in-flight audio, flushes pending sentences, and triggers `await call.clear_send_audio_buffer()` to fall silent instantly. --- ## Prerequisites 1. **Hugging Face Token:** The underlying voice model (`kyutai/pocket-tts`) is gated on Hugging Face: - Create an account at [huggingface.co/join](https://huggingface.co/join). - Generate a Read access token at [huggingface.co/settings/tokens](https://huggingface.co/settings/tokens). - Visit [huggingface.co/kyutai/pocket-tts](https://huggingface.co/kyutai/pocket-tts) and click **Agree and access repository**. 2. **Docker:** Installed and running on your machine. 3. **AgentDuet Account:** API key and connector UUID from [agentduet.com](https://agentduet.com). 4. **Google AI Studio Key:** Gemini API key from [aistudio.google.com](https://aistudio.google.com/apikey). --- ## Step 1: Start the Voice Cloning Server Clone the repository and run the server using Docker: ```bash git clone https://github.com/AgentDuet/voice-cloning-tts.git cd voice-cloning-tts # Build the local Docker image docker build -t tts-server:local . # Run the container with your Hugging Face token docker run --rm -it \ -p 8000:8000 \ -e HF_TOKEN= \ -v "$(pwd)/voices:/app/voices" \ -v "$(pwd)/hf-cache:/root/.cache/huggingface" \ --name tts-server \ tts-server:local ``` Verify the server is running: ```bash curl http://localhost:8000/health # Returns: {"status":"ok","workers":4} ``` --- ## Step 2: Register a voice sample Upload a short audio sample (5 to 10 seconds WAV file of the target speaker) to create a reusable voice profile: ```bash curl -X POST http://localhost:8000/tts/upload \ -F "voice_file=@sample.wav" \ -F "voice_name=executive_persona" ``` The response returns a `voice_id`: ```json { "voice_id": "v_a81f4b2e9c", "voice_name": "executive_persona", "status": "ready" } ``` Save this `voice_id` for your agent configuration. --- ## Step 3: Set up the phone agent project Create a dedicated folder and install dependencies: ```bash mkdir voice-cloned-agent && cd voice-cloned-agent python3 -m venv .venv && source .venv/bin/activate pip install agentduet "google-genai>=1.57.0" websockets python-dotenv ``` --- ## Step 4: Configure `.env` Create a `.env` file with your credentials: ```bash cat > .env << 'EOF' AGENTDUET_API_KEY=your-agentduet-api-key AGENTDUET_CONNECTOR_UUID=your-connector-uuid GEMINI_API_KEY=your-gemini-api-key TTS_VOICE_ID=v_a81f4b2e9c TTS_SERVER_URL=ws://localhost:8000 EOF ``` --- ## Step 5: Write `phone_agent.py` Create `phone_agent.py` next to `.env`: ```python 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()) ``` --- ## Step 6: Test the cloned voice call 1. Start your phone agent: ```bash python phone_agent.py ``` 2. Call your AgentDuet phone number or WhatsApp line. 3. Speak to the agent. The agent listens via Gemini Live and answers back in real time speaking in your custom cloned voice profile. ## Related - [voice-cloning-tts on GitHub](https://github.com/AgentDuet/voice-cloning-tts) - [Gemini Live](/integrations/gemini-live) - [Amazon Nova Sonic](/integrations/amazon-nova-sonic) ## Next Step Explore integration with Google ADK.