Google GenAI SDK (Gemini Live)

1
import asyncio
2
import os
3
from google import genai
4
from google.genai import types
5
from telcoflow_sdk import TelcoflowClient, TelcoflowClientConfig, ActiveCall
6
import telcoflow_sdk.events as events
7
8
gemini_client = genai.Client(api_key=os.getenv("GEMINI_API_KEY"))
9
MODEL = "gemini-2.5-flash-native-audio-preview-12-2025"
10
11
async def start_gemini_session(call: ActiveCall):
12
    await call.answer()
13
14
    async with gemini_client.aio.live.connect(model=MODEL) as session:
15
        async def stream_to_gemini():
16
            async for chunk in call.audio_stream():
17
                await session.send_realtime_input(
18
                    audio=types.Blob(
19
                        data=chunk, mime_type="audio/pcm;rate=24000"
20
                    )
21
                )
22
23
        async def receive_from_gemini():
24
            async for response in session.receive():
25
                if content := response.server_content:
26
                    if content.interrupted:
27
                        await call.clear_send_audio_buffer()
28
                    elif content.model_turn:
29
                        for part in content.model_turn.parts:
30
                            if part.inline_data:
31
                                await call.send_audio(part.inline_data.data)
32
33
        await asyncio.gather(stream_to_gemini(), receive_from_gemini())
34
35
async def main():
36
    config = TelcoflowClientConfig.sandbox(
37
        api_key=os.getenv("WSS_API_KEY"),
38
        connector_uuid=os.getenv("WSS_CONNECTOR_UUID"),
39
        sample_rate=24000,
40
    )
41
42
    async with TelcoflowClient(config) as client:
43
        @client.on(events.INCOMING_CALL)
44
        async def on_call(call: ActiveCall):
45
            await start_gemini_session(call)
46
47
        await client.run_forever()
48
49
if __name__ == "__main__":
50
    asyncio.run(main())