> 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. # Google ADK This guide shows how to integrate the AgentDuet SDK with the [Google Agent Development Kit (ADK)](https://google.github.io/adk-docs/) for complex AI agent behavior, including multi-agent orchestration, long-term memory, and structured tools. ## What this integration provides A phone agent that answers calls, feeds caller audio into ADK’s `LiveRequestQueue`, plays ADK audio back on the call, clears the outbound buffer when ADK reports `interrupted`, and cancels bridge tasks 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)) - [Google ADK](https://google.github.io/adk-docs/) ```bash pip install "agentduet==1.0.0" google-genai google-adk python-dotenv ``` ## Step 1: Create a project folder ```bash mkdir agentduet-adk-bidi-bridge && cd $_ python3.12 -m venv .venv && source .venv/bin/activate pip install "agentduet==1.0.0" google-genai google-adk 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 # ADK / genai often also read GOOGLE_API_KEY - set either EOF ``` ## Step 3: Write `adk_bidi_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.adk.agents import Agent from google.adk.agents.live_request_queue import LiveRequestQueue from google.adk.agents.run_config import RunConfig, StreamingMode from google.adk.runners import Runner from google.adk.sessions import InMemorySessionService from google.genai import types 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__) MODEL = "gemini-3.1-flash-live-preview" APP_NAME = "agentduet_adk_integration" agent = Agent( name="agentduet_agent", model=MODEL, instruction="You are a helpful and friendly AI assistant talking over a phone call.", ) session_service = InMemorySessionService() runner = Runner(app_name=APP_NAME, agent=agent, session_service=session_service) class ADKIntegration: def __init__(self, user_id: str, call: Call): self._call = call self._user_id = user_id self._session_id = call.id self._live_request_queue = LiveRequestQueue() self._send_to_adk_task: Optional[asyncio.Task] = None self._recv_from_adk_task: Optional[asyncio.Task] = None self._terminated = False async def _on_hangup(self, evt): logger.info("Call terminated - cancelling ADK bridge tasks") self._terminated = True if self._send_to_adk_task: self._send_to_adk_task.cancel() if self._recv_from_adk_task: self._recv_from_adk_task.cancel() async def run(self): self._call.on_hangup(self._on_hangup) session = await session_service.get_session( app_name=APP_NAME, user_id=self._user_id, session_id=self._session_id ) if not session: await session_service.create_session( app_name=APP_NAME, user_id=self._user_id, session_id=self._session_id ) self._send_to_adk_task = asyncio.create_task(self.stream_to_adk()) self._recv_from_adk_task = asyncio.create_task( self.receive_from_adk(self._user_id, self._session_id) ) try: await asyncio.gather(self._send_to_adk_task, self._recv_from_adk_task) except asyncio.CancelledError: logger.debug("ADK integration tasks cancelled") async def stream_to_adk(self): try: self._live_request_queue.send_content( types.Content(parts=[types.Part(text="Hi")]) ) async for audio_chunk in self._call.caller.audio_stream(): if self._terminated: break audio_blob = types.Blob( data=audio_chunk, mime_type="audio/pcm;rate=24000", ) self._live_request_queue.send_realtime(audio_blob) except CallClosedError: logger.debug("Call closed; stopping stream to ADK") except Exception: if not self._terminated: logger.exception("Error in stream to ADK") raise finally: logger.debug("Stream to ADK completed") async def receive_from_adk(self, user_id: str, session_id: str): run_config = RunConfig( streaming_mode=StreamingMode.BIDI, response_modalities=["AUDIO"], input_audio_transcription=None, output_audio_transcription=None, realtime_input_config=types.RealtimeInputConfig(), ) try: async for event in runner.run_live( user_id=user_id, session_id=session_id, live_request_queue=self._live_request_queue, run_config=run_config, ): if self._terminated: break if event.interrupted: await self._call.clear_send_audio_buffer() if event.content and event.content.parts: part = event.content.parts[0] if part.inline_data and part.inline_data.data: try: await self._call.send_audio(part.inline_data.data) except BufferFullError: logger.warning( "Send buffer full - drop chunk or raise buffer_size" ) event.content = None except CallClosedError: logger.debug("Call closed; stopping receive from ADK") except Exception: if not self._terminated: logger.exception("Error in receive from ADK") raise finally: logger.debug("Receive from ADK completed") 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), ) async with SessionManager(config) as sm: logger.info("Connected. Waiting for 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("Incoming call: %s", call.id) try: result = await call.answer() if not result: logger.error( "Answer failed %s: %s (%s)", call.id, result.error_message, result.error_code, ) return await ADKIntegration(user_id="default_user", call=call).run() except Exception: logger.exception("Error handling call with ADK") await call.close() await sm.run_forever() if __name__ == "__main__": asyncio.run(main()) ``` ## Step 4: Run the bridge ```bash python adk_bidi_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:** ADK/Gemini Live path expects 24 kHz; set `mime_type="audio/pcm;rate=24000"`. - **Interrupt:** On `event.interrupted`, call `await call.clear_send_audio_buffer()` - not private interrupt helpers. - **Session IDs:** Using `call.id` as the ADK session id keeps one ADK session per phone call. - **`event.content = None`:** Dropping processed content helps avoid retaining large audio blobs in memory. - **Hangup:** ADK owns the model connection via `Runner`; cancel your tasks and let `run_live` exit. ## Related - [Gemini Live](/integrations/gemini-live) - [Audio Streaming](/concepts/audio-streaming)