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