Grok Voice

1
from __future__ import annotations
2
3
import asyncio
4
import base64
5
import json
6
import logging
7
import os
8
from typing import Any, Optional
9
10
import websockets
11
from dotenv import load_dotenv
12
from websockets.asyncio.client import ClientConnection
13
from websockets.exceptions import ConnectionClosed, InvalidStatus
14
15
from agentduet import (
16
    BufferFullError,
17
    Call,
18
    CallAudioConfig,
19
    CallClosedError,
20
    IncomingCallNotification,
21
    SessionManager,
22
    SessionManagerConfig,
23
    new_session_id,
24
)
25
26
load_dotenv()
27
28
logging.basicConfig(
29
    level=logging.INFO,
30
    format="%(asctime)s %(name)s %(levelname)s %(message)s",
31
)
32
logger = logging.getLogger(__name__)
33
34
AGENTDUET_API_KEY = os.environ["AGENTDUET_API_KEY"]
35
AGENTDUET_CONNECTOR_UUID = os.environ["AGENTDUET_CONNECTOR_UUID"]
36
XAI_API_KEY = os.environ["XAI_API_KEY"]
37
38
GROK_MODEL = "grok-voice-think-fast-1.0"
39
GROK_REALTIME_URL = f"wss://api.x.ai/v1/realtime?model={GROK_MODEL}"
40
GROK_SAMPLE_RATE = 24000
41
GROK_VOICE = os.environ.get("GROK_VOICE", "leo").lower()
42
AGENT_NAME = "Grok"
43
44
SYSTEM_PROMPT = (
45
    f"Your name is {AGENT_NAME}. You are a witty, helpful voice assistant on a phone call. "
46
    "Greet the caller briefly, keep answers short and conversational, and ask clarifying "
47
    "questions when needed. When the caller wants to hang up or says goodbye, say a brief "
48
    "goodbye and call the hang_up tool."
49
)
50
51
HANG_UP_TOOL = {
52
    "type": "function",
53
    "name": "hang_up",
54
    "description": (
55
        "End the phone call. Use when the caller asks to hang up, "
56
        "end the call, or says goodbye and wants to leave."
57
    ),
58
    "parameters": {"type": "object", "properties": {}, "required": []},
59
}
60
61
62
class GrokLiveIntegration:
63
    """Bidirectional audio bridge: AgentDuet Call ↔ xAI Grok Realtime."""
64
65
    def __init__(self, call: Call, grok_ws: ClientConnection):
66
        self._call = call
67
        self._grok_ws = grok_ws
68
        self._send_to_grok_task: Optional[asyncio.Task] = None
69
        self._recv_from_grok_task: Optional[asyncio.Task] = None
70
        self._playback_task: Optional[asyncio.Task] = None
71
        self._terminated = False
72
73
        # Playback must not block the Grok event loop, or speech_started
74
        # arrives too late for clear_send_audio_buffer() to help.
75
        self._playback_gen = 0
76
        self._audio_queue: asyncio.Queue[tuple[int, bytes] | None] = asyncio.Queue()
77
        self._active_response_id: Optional[str] = None
78
        self._cancelled_response_ids: set[str] = set()
79
80
    async def _on_hangup(self, _evt: Any) -> None:
81
        logger.info("Call %s hung up", self._call.id)
82
        self._terminated = True
83
        try:
84
            await self._grok_ws.close()
85
        except Exception:
86
            logger.exception("Error closing Grok WebSocket")
87
88
        for task in (
89
            self._send_to_grok_task,
90
            self._recv_from_grok_task,
91
            self._playback_task,
92
        ):
93
            if task and not task.done():
94
                task.cancel()
95
                try:
96
                    await task
97
                except asyncio.CancelledError:
98
                    pass
99
100
    async def run(self) -> None:
101
        self._call.on_hangup(self._on_hangup)
102
        await self._configure_session()
103
        await self._greet_caller()
104
105
        self._playback_task = asyncio.create_task(self._playback_worker())
106
        self._send_to_grok_task = asyncio.create_task(self._stream_to_grok())
107
        self._recv_from_grok_task = asyncio.create_task(self._receive_from_grok())
108
109
        await asyncio.gather(
110
            self._send_to_grok_task,
111
            self._recv_from_grok_task,
112
            return_exceptions=True,
113
        )
114
115
        if self._playback_task and not self._playback_task.done():
116
            self._audio_queue.put_nowait(None)
117
            try:
118
                await self._playback_task
119
            except asyncio.CancelledError:
120
                pass
121
122
    async def _configure_session(self) -> None:
123
        await self._grok_ws.send(
124
            json.dumps(
125
                {
126
                    "type": "session.update",
127
                    "session": {
128
                        "voice": GROK_VOICE,
129
                        "instructions": SYSTEM_PROMPT,
130
                        "reasoning": {"effort": "none"},
131
                        "turn_detection": {
132
                            "type": "server_vad",
133
                            "threshold": 0.5,
134
                            "silence_duration_ms": 300,
135
                            "prefix_padding_ms": 200,
136
                        },
137
                        "tools": [HANG_UP_TOOL],
138
                        "audio": {
139
                            "input": {
140
                                "format": {
141
                                    "type": "audio/pcm",
142
                                    "rate": GROK_SAMPLE_RATE,
143
                                },
144
                            },
145
                            "output": {
146
                                "format": {
147
                                    "type": "audio/pcm",
148
                                    "rate": GROK_SAMPLE_RATE,
149
                                },
150
                            },
151
                        },
152
                    },
153
                }
154
            )
155
        )
156
157
    async def _greet_caller(self) -> None:
158
        await self._grok_ws.send(
159
            json.dumps(
160
                {
161
                    "type": "response.create",
162
                    "response": {
163
                        "instructions": (
164
                            f"Greet the caller warmly. Introduce yourself as {AGENT_NAME} "
165
                            "and ask how you can help today."
166
                        ),
167
                    },
168
                }
169
            )
170
        )
171
172
    async def _stream_to_grok(self) -> None:
173
        try:
174
            async for chunk in self._call.caller.audio_stream():
175
                if self._terminated:
176
                    break
177
                await self._grok_ws.send(
178
                    json.dumps(
179
                        {
180
                            "type": "input_audio_buffer.append",
181
                            "audio": base64.b64encode(chunk).decode("ascii"),
182
                        }
183
                    )
184
                )
185
        except (CallClosedError, ConnectionClosed):
186
            pass
187
        except asyncio.CancelledError:
188
            raise
189
        except Exception:
190
            logger.exception("Error streaming caller audio to Grok")
191
            raise
192
193
    async def _playback_worker(self) -> None:
194
        while True:
195
            item = await self._audio_queue.get()
196
            if item is None:
197
                break
198
            gen, audio = item
199
            if gen != self._playback_gen:
200
                continue
201
            try:
202
                await self._call.send_audio(audio)
203
            except BufferFullError:
204
                logger.warning("Outgoing buffer full - dropping chunk")
205
            except CallClosedError:
206
                break
207
            except asyncio.CancelledError:
208
                raise
209
210
    async def _flush_playback(self) -> None:
211
        self._playback_gen += 1
212
        self._active_response_id = None
213
        while not self._audio_queue.empty():
214
            try:
215
                self._audio_queue.get_nowait()
216
            except asyncio.QueueEmpty:
217
                break
218
        try:
219
            await self._call.clear_send_audio_buffer()
220
        except CallClosedError:
221
            return
222
223
    async def _agent_hang_up(self) -> None:
224
        logger.info("Agent hanging up call %s", self._call.id)
225
        try:
226
            for _ in range(40):  # up to ~4s for goodbye audio to drain
227
                if await self._call.get_send_audio_buffer_size() == 0:
228
                    break
229
                await asyncio.sleep(0.1)
230
        except CallClosedError:
231
            return
232
        result = await self._call.close()
233
        if not result:
234
            logger.error(
235
                "Hang up failed for %s: %s (%s)",
236
                self._call.id,
237
                result.error_message,
238
                result.error_code,
239
            )
240
241
    async def _receive_from_grok(self) -> None:
242
        try:
243
            async for raw_event in self._grok_ws:
244
                if self._terminated:
245
                    break
246
247
                event = json.loads(raw_event)
248
                etype = event.get("type")
249
250
                if etype == "response.created":
251
                    self._active_response_id = event.get("response", {}).get("id")
252
253
                elif etype == "input_audio_buffer.speech_started":
254
                    try:
255
                        buf_size = await self._call.get_send_audio_buffer_size()
256
                    except CallClosedError:
257
                        buf_size = 0
258
                    should_flush = (
259
                        self._active_response_id is not None
260
                        or buf_size > 0
261
                        or self._audio_queue.qsize() > 0
262
                    )
263
                    if should_flush:
264
                        if self._active_response_id is not None:
265
                            self._cancelled_response_ids.add(self._active_response_id)
266
                        logger.info("Caller interrupted - stopping playback")
267
                        await self._flush_playback()
268
269
                elif etype == "response.done":
270
                    response_id = event.get("response", {}).get("id")
271
                    if response_id:
272
                        self._cancelled_response_ids.discard(response_id)
273
                    if response_id == self._active_response_id:
274
                        self._active_response_id = None
275
276
                elif etype in ("response.output_audio.delta", "response.audio.delta"):
277
                    response_id = event.get("response_id")
278
                    if response_id and response_id in self._cancelled_response_ids:
279
                        continue
280
                    if (
281
                        response_id
282
                        and self._active_response_id
283
                        and response_id != self._active_response_id
284
                    ):
285
                        continue
286
                    audio = base64.b64decode(event["delta"])
287
                    self._audio_queue.put_nowait((self._playback_gen, audio))
288
289
                elif etype == "response.function_call_arguments.done":
290
                    if event.get("name") != "hang_up":
291
                        logger.warning("Unknown tool: %s", event.get("name"))
292
                        continue
293
                    await self._grok_ws.send(
294
                        json.dumps(
295
                            {
296
                                "type": "conversation.item.create",
297
                                "item": {
298
                                    "type": "function_call_output",
299
                                    "call_id": event["call_id"],
300
                                    "output": json.dumps({"status": "hanging_up"}),
301
                                },
302
                            }
303
                        )
304
                    )
305
                    await self._agent_hang_up()
306
                    return
307
308
                elif etype == "error":
309
                    logger.error("Grok error event: %s", event)
310
311
        except (CallClosedError, ConnectionClosed):
312
            pass
313
        except asyncio.CancelledError:
314
            raise
315
        except Exception:
316
            logger.exception("Error receiving from Grok")
317
            raise
318
        finally:
319
            self._audio_queue.put_nowait(None)
320
321
322
async def bridge_call_to_grok(call: Call) -> None:
323
    try:
324
        async with websockets.connect(
325
            GROK_REALTIME_URL,
326
            additional_headers={"Authorization": f"Bearer {XAI_API_KEY}"},
327
            open_timeout=15,
328
        ) as grok_ws:
329
            result = await call.answer()
330
            if not result:
331
                logger.error(
332
                    "Answer failed for call %s: %s (%s)",
333
                    call.id,
334
                    result.error_message,
335
                    result.error_code,
336
                )
337
                return
338
            await GrokLiveIntegration(call, grok_ws).run()
339
    except InvalidStatus as e:
340
        body = getattr(e.response, "body", b"") or b""
341
        detail = body.decode("utf-8", errors="replace") if body else str(e)
342
        logger.error(
343
            "Grok WebSocket rejected (HTTP %s). Check XAI_API_KEY - %s",
344
            e.response.status_code,
345
            detail,
346
        )
347
    except Exception:
348
        logger.exception("Failed during Grok bridge for call %s", call.id)
349
350
351
async def main() -> None:
352
    config = SessionManagerConfig.create(
353
        api_key=AGENTDUET_API_KEY,
354
        connector_uuid=AGENTDUET_CONNECTOR_UUID,
355
        call_audio=CallAudioConfig(
356
            sample_rate=GROK_SAMPLE_RATE,
357
            buffer_size=1024 * 1024,
358
        ),
359
    )
360
361
    async with SessionManager(config) as sm:
362
        logger.info("Connected. Waiting for calls...")
363
364
        @sm.on_incoming_call
365
        async def on_call(noti: IncomingCallNotification) -> None:
366
            logger.info("Incoming call %s from %s", noti.call_id, noti.participant)
367
            session = await sm.open_session(new_session_id(), noti.subscriber)
368
            call = await session.process_call(noti)
369
            try:
370
                await bridge_call_to_grok(call)
371
            finally:
372
                await call.close()
373
374
        await sm.run_forever()
375
376
377
if __name__ == "__main__":
378
    asyncio.run(main())