For clean Markdown of any page, append .md to the page URL. For a complete documentation index, see https://docs.telcoflow.com/integrations/llms.txt. For full documentation content, see https://docs.telcoflow.com/integrations/llms-full.txt.

# Google GenAI SDK (Gemini Live)

This guide shows how to integrate the Telcoflow SDK with the [Google GenAI SDK](https://github.com/google-gemini/generative-ai-python) for bidirectional real-time audio streaming with Gemini's native audio model.

## Overview

The integration bridges two real-time streams:
- **Caller audio -> Gemini**: Incoming phone audio is forwarded to a Gemini Live session
- **Gemini audio -> Caller**: Gemini's voice responses are sent back to the caller via `send_audio()`

Interruption handling is built in: when Gemini detects the user is speaking over the model, the outgoing audio buffer is cleared instantly.

## Full Example

```python
import asyncio
import os
from google import genai
from google.genai import types
from telcoflow_sdk import TelcoflowClient, TelcoflowClientConfig, ActiveCall
import telcoflow_sdk.events as events

gemini_client = genai.Client(api_key=os.getenv("GEMINI_API_KEY"))
MODEL = "gemini-2.5-flash-native-audio-preview-12-2025"

async def start_gemini_session(call: ActiveCall):
    await call.answer()

    async with gemini_client.aio.live.connect(model=MODEL) as session:
        async def stream_to_gemini():
            async for chunk in call.audio_stream():
                await session.send_realtime_input(
                    audio=types.Blob(
                        data=chunk, mime_type="audio/pcm;rate=24000"
                    )
                )

        async def receive_from_gemini():
            async for response in session.receive():
                if content := response.server_content:
                    if content.interrupted:
                        await call.clear_send_audio_buffer()
                    elif content.model_turn:
                        for part in content.model_turn.parts:
                            if part.inline_data:
                                await call.send_audio(part.inline_data.data)

        await asyncio.gather(stream_to_gemini(), receive_from_gemini())

async def main():
    config = TelcoflowClientConfig.sandbox(
        api_key=os.getenv("WSS_API_KEY"),
        connector_uuid=os.getenv("WSS_CONNECTOR_UUID"),
        sample_rate=24000,
    )

    async with TelcoflowClient(config) as client:
        @client.on(events.INCOMING_CALL)
        async def on_call(call: ActiveCall):
            await start_gemini_session(call)

        await client.run_forever()

if __name__ == "__main__":
    asyncio.run(main())
```

## How It Works

### Stream to Gemini

The `stream_to_gemini()` coroutine reads audio chunks from `call.audio_stream()` and forwards them to the Gemini Live session using `send_realtime_input()`. The audio is wrapped in a `types.Blob` with the PCM MIME type.

### Receive from Gemini

The `receive_from_gemini()` coroutine listens for Gemini responses:

- **Interruption**: When `content.interrupted` is `True`, the caller has started speaking over the model. `clear_send_audio_buffer()` is called to immediately stop any queued audio.
- **Model audio**: When `content.model_turn` contains `inline_data`, the raw audio bytes are sent to the caller via `send_audio()`.

### Concurrency

Both coroutines run concurrently via `asyncio.gather()`. This allows the system to simultaneously listen to the caller and send AI responses without blocking.

## Environment Variables

| Variable | Description |
|---|---|
| `GEMINI_API_KEY` | Google API key with Gemini API access |
| `WSS_API_KEY` | Telcoflow API key |
| `WSS_CONNECTOR_UUID` | Telcoflow connector UUID |

## Audio Format

Both Telcoflow and Gemini native audio use **PCM 16-bit linear, 24kHz, mono** (`audio/pcm;rate=24000`). No transcoding is needed.

## Next Steps

- [Google ADK Integration](/integrations/google-adk-agent-development-kit) - For structured multi-agent orchestration
- [Audio Streaming](/concepts/audio-streaming) - Buffer management and interruption handling
- [Use Cases](/use-cases/database-lookup) - Apply this integration to real-world scenarios