Google ADK

1
import asyncio
2
import logging
3
import os
4
from typing import Optional
5
6
from dotenv import load_dotenv
7
from google.adk.agents import Agent
8
from google.adk.agents.live_request_queue import LiveRequestQueue
9
from google.adk.agents.run_config import RunConfig, StreamingMode
10
from google.adk.runners import Runner
11
from google.adk.sessions import InMemorySessionService
12
from google.genai import types
13
14
from agentduet import (
15
    BufferFullError,
16
    Call,
17
    CallAudioConfig,
18
    CallClosedError,
19
    IncomingCallNotification,
20
    SessionManager,
21
    SessionManagerConfig,
22
    new_session_id,
23
)
24
25
load_dotenv()
26
27
logging.basicConfig(
28
    level=logging.INFO,
29
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
30
)
31
logger = logging.getLogger(__name__)
32
33
MODEL = "gemini-3.1-flash-live-preview"
34
APP_NAME = "agentduet_adk_integration"
35
36
agent = Agent(
37
    name="agentduet_agent",
38
    model=MODEL,
39
    instruction="You are a helpful and friendly AI assistant talking over a phone call.",
40
)
41
42
session_service = InMemorySessionService()
43
runner = Runner(app_name=APP_NAME, agent=agent, session_service=session_service)
44
45
46
class ADKIntegration:
47
    def __init__(self, user_id: str, call: Call):
48
        self._call = call
49
        self._user_id = user_id
50
        self._session_id = call.id
51
        self._live_request_queue = LiveRequestQueue()
52
        self._send_to_adk_task: Optional[asyncio.Task] = None
53
        self._recv_from_adk_task: Optional[asyncio.Task] = None
54
        self._terminated = False
55
56
    async def _on_hangup(self, evt):
57
        logger.info("Call terminated - cancelling ADK bridge tasks")
58
        self._terminated = True
59
        if self._send_to_adk_task:
60
            self._send_to_adk_task.cancel()
61
        if self._recv_from_adk_task:
62
            self._recv_from_adk_task.cancel()
63
64
    async def run(self):
65
        self._call.on_hangup(self._on_hangup)
66
67
        session = await session_service.get_session(
68
            app_name=APP_NAME, user_id=self._user_id, session_id=self._session_id
69
        )
70
        if not session:
71
            await session_service.create_session(
72
                app_name=APP_NAME, user_id=self._user_id, session_id=self._session_id
73
            )
74
75
        self._send_to_adk_task = asyncio.create_task(self.stream_to_adk())
76
        self._recv_from_adk_task = asyncio.create_task(
77
            self.receive_from_adk(self._user_id, self._session_id)
78
        )
79
80
        try:
81
            await asyncio.gather(self._send_to_adk_task, self._recv_from_adk_task)
82
        except asyncio.CancelledError:
83
            logger.debug("ADK integration tasks cancelled")
84
85
    async def stream_to_adk(self):
86
        try:
87
            self._live_request_queue.send_content(
88
                types.Content(parts=[types.Part(text="Hi")])
89
            )
90
            async for audio_chunk in self._call.caller.audio_stream():
91
                if self._terminated:
92
                    break
93
                audio_blob = types.Blob(
94
                    data=audio_chunk,
95
                    mime_type="audio/pcm;rate=24000",
96
                )
97
                self._live_request_queue.send_realtime(audio_blob)
98
        except CallClosedError:
99
            logger.debug("Call closed; stopping stream to ADK")
100
        except Exception:
101
            if not self._terminated:
102
                logger.exception("Error in stream to ADK")
103
            raise
104
        finally:
105
            logger.debug("Stream to ADK completed")
106
107
    async def receive_from_adk(self, user_id: str, session_id: str):
108
        run_config = RunConfig(
109
            streaming_mode=StreamingMode.BIDI,
110
            response_modalities=["AUDIO"],
111
            input_audio_transcription=None,
112
            output_audio_transcription=None,
113
            realtime_input_config=types.RealtimeInputConfig(),
114
        )
115
116
        try:
117
            async for event in runner.run_live(
118
                user_id=user_id,
119
                session_id=session_id,
120
                live_request_queue=self._live_request_queue,
121
                run_config=run_config,
122
            ):
123
                if self._terminated:
124
                    break
125
126
                if event.interrupted:
127
                    await self._call.clear_send_audio_buffer()
128
129
                if event.content and event.content.parts:
130
                    part = event.content.parts[0]
131
                    if part.inline_data and part.inline_data.data:
132
                        try:
133
                            await self._call.send_audio(part.inline_data.data)
134
                        except BufferFullError:
135
                            logger.warning(
136
                                "Send buffer full - drop chunk or raise buffer_size"
137
                            )
138
                    event.content = None
139
        except CallClosedError:
140
            logger.debug("Call closed; stopping receive from ADK")
141
        except Exception:
142
            if not self._terminated:
143
                logger.exception("Error in receive from ADK")
144
            raise
145
        finally:
146
            logger.debug("Receive from ADK completed")
147
148
149
async def main():
150
    config = SessionManagerConfig.create(
151
        api_key=os.getenv("AGENTDUET_API_KEY"),
152
        connector_uuid=os.getenv("AGENTDUET_CONNECTOR_UUID"),
153
        call_audio=CallAudioConfig(sample_rate=24000),
154
    )
155
156
    async with SessionManager(config) as sm:
157
        logger.info("Connected. Waiting for calls...")
158
159
        @sm.on_incoming_call
160
        async def on_call(noti: IncomingCallNotification):
161
            session = await sm.open_session(new_session_id(), noti.subscriber)
162
            call = await session.process_call(noti)
163
            logger.info("Incoming call: %s", call.id)
164
            try:
165
                result = await call.answer()
166
                if not result:
167
                    logger.error(
168
                        "Answer failed %s: %s (%s)",
169
                        call.id,
170
                        result.error_message,
171
                        result.error_code,
172
                    )
173
                    return
174
                await ADKIntegration(user_id="default_user", call=call).run()
175
            except Exception:
176
                logger.exception("Error handling call with ADK")
177
                await call.close()
178
179
        await sm.run_forever()
180
181
182
if __name__ == "__main__":
183
    asyncio.run(main())