Production-Grade Voice AI Agent with Gemini Live API & Google ADK [2026]

pip install google-genai>=1.10.0 google-adk>=0.4.0 \
    fastapi uvicorn websockets aiohttp python-dotenv

# .env
GOOGLE_CLOUD_PROJECT=your-project-id
GOOGLE_CLOUD_LOCATION=us-central1
GOOGLE_APPLICATION_CREDENTIALS=/path/to/service-account.json
 
# Or API key auth (dev only)
GEMINI_API_KEY=your-api-key

# agent.py
import asyncio
from datetime import datetime
from google.adk.agents import LiveAgent
from google.adk.tools import function_tool
from google.genai.types import LiveConnectConfig, SpeechConfig, VoiceConfig
 
@function_tool
async def get_current_weather(city: str) -> dict:
    """Retrieve current weather data for a given city.
 
    Args:
        city: City name to look up (e.g., "Tokyo", "New York")
 
    Returns:
        Dict with temperature, condition, and humidity
    """
    mock_weather = {
        "Tokyo": {"temp": 18, "condition": "Sunny", "humidity": 55},
        "New York": {"temp": 12, "condition": "Cloudy", "humidity": 70},
    }
    data = mock_weather.get(city, {"temp": 15, "condition": "Unknown", "humidity": 50})
    return {
        "city": city,
        "temperature": data["temp"],
        "condition": data["condition"],
        "humidity": data["humidity"],
        "timestamp": datetime.now().isoformat(),
    }
 
@function_tool
async def search_knowledge_base(query: str, category: str = "general") -> dict:
    """Search the internal knowledge base and return relevant results.
 
    Args:
        query: The search query string
        category: Search scope (general / product / support)
 
    Returns:
        List of matching documents with relevance scores
    """
    return {
        "query": query,
        "category": category,
        "results": [
            {"title": f"Document about '{query}'", "relevance": 0.92},
            {"title": f"{category} category FAQ", "relevance": 0.78},
        ],
    }
 
@function_tool
async def create_support_ticket(
    issue_summary: str,
    priority: str = "medium",
    user_id: str = ""
) -> dict:
    """Create a support ticket for the user's issue.
 
    Args:
        issue_summary: Brief description of the issue
        priority: Ticket priority (low / medium / high / urgent)
        user_id: Optional user identifier
 
    Returns:
        Created ticket details including ID and ETA
    """
    import random
    ticket_id = f"TKT-{random.randint(10000, 99999)}"
    return {
        "ticket_id": ticket_id,
        "summary": issue_summary,
        "priority": priority,
        "status": "created",
        "estimated_response": "Within 2 hours" if priority in ["high", "urgent"] else "Within 1 business day",
    }
 
def create_voice_agent() -> LiveAgent:
    """Instantiate a production-ready voice AI agent."""
    return LiveAgent(
        model="gemini-live-2.5-flash-native-audio",
        system_instruction="""You are a polite and capable customer support AI assistant.
 
Your responsibilities:
- Answer customer questions clearly via voice
- Use weather, knowledge base, and ticket creation tools as needed
- Keep responses concise to maintain natural conversation flow
- Always confirm before taking irreversible actions
 
Constraints:
- Never handle sensitive data like passwords or credit card numbers
- Don't make definitive claims about unverifiable facts""",
        tools=[get_current_weather, search_knowledge_base, create_support_ticket],
        config=LiveConnectConfig(
            response_modalities=["AUDIO"],
            speech_config=SpeechConfig(
                voice_config=VoiceConfig(
                    prebuilt_voice_config={"voice_name": "Aoede"}
                )
            ),
        ),
    )

# server.py
import asyncio
import logging
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from fastapi.middleware.cors import CORSMiddleware
from agent import create_voice_agent
 
logger = logging.getLogger(__name__)
app = FastAPI(title="Voice Agent API")
 
app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"])
 
class VoiceAgentSession:
    """Manages a single voice agent session with retry logic."""
 
    MAX_RETRIES = 3
    RETRY_DELAY = 1.5  # seconds
 
    def __init__(self, session_id: str):
        self.session_id = session_id
        self.agent = create_voice_agent()
        self._live_session = None
        self._retry_count = 0
 
    async def _connect_live_session(self):
        """Connect to Gemini Live API with exponential backoff."""
        while self._retry_count < self.MAX_RETRIES:
            try:
                self._live_session = await self.agent.connect()
                self._retry_count = 0
                logger.info(f"[{self.session_id}] Live API connected")
                return
            except Exception as e:
                self._retry_count += 1
                wait = self.RETRY_DELAY * (2 ** (self._retry_count - 1))
                logger.warning(f"[{self.session_id}] Retry {self._retry_count}/{self.MAX_RETRIES} in {wait:.1f}s: {e}")
                await asyncio.sleep(wait)
        raise RuntimeError(f"[{self.session_id}] Max retries exceeded")
 
    async def handle(self, client_ws: WebSocket):
        """Bridge between the client WebSocket and Gemini Live API."""
        await client_ws.accept()
        await self._connect_live_session()
 
        send_task = asyncio.create_task(self._forward_client_to_gemini(client_ws))
        recv_task = asyncio.create_task(self._forward_gemini_to_client(client_ws))
 
        try:
            done, pending = await asyncio.wait(
                [send_task, recv_task], return_when=asyncio.FIRST_EXCEPTION
            )
            for task in pending:
                task.cancel()
        except WebSocketDisconnect:
            logger.info(f"[{self.session_id}] Client disconnected")
        finally:
            await self._cleanup()
 
    async def _forward_client_to_gemini(self, client_ws: WebSocket):
        async for message in client_ws.iter_bytes():
            if self._live_session:
                await self._live_session.send_audio(message)
 
    async def _forward_gemini_to_client(self, client_ws: WebSocket):
        async for event in self._live_session.receive():
            if event.type == "audio":
                await client_ws.send_bytes(event.data)
            elif event.type == "text":
                await client_ws.send_json({"type": "transcript", "text": event.text})
 
    async def _cleanup(self):
        if self._live_session:
            try:
                await self._live_session.close()
            except Exception:
                pass
 
@app.websocket("/ws/{session_id}")
async def websocket_endpoint(websocket: WebSocket, session_id: str):
    session = VoiceAgentSession(session_id)
    await session.handle(websocket)
 
@app.get("/health")
async def health_check():
    return {"status": "ok", "model": "gemini-live-2.5-flash-native-audio"}

# parallel_tools.py
import asyncio
from typing import Any
 
class ParallelToolExecutor:
    """Orchestrates parallel execution of multiple Function Calls."""
 
    def __init__(self, timeout_seconds: float = 5.0):
        self.timeout = timeout_seconds
        self._registry: dict[str, callable] = {}
 
    def register(self, name: str, fn: callable):
        self._registry[name] = fn
 
    async def execute_parallel(self, tool_calls: list[dict]) -> list[dict]:
        """Execute multiple tool calls concurrently.
 
        Args:
            tool_calls: [{"name": "tool_name", "args": {...}}, ...]
 
        Returns:
            Results list with status for each call
        """
        tasks = [
            (call["name"], asyncio.create_task(
                self._run(self._registry[call["name"]], call["args"])
            ))
            for call in tool_calls
            if call["name"] in self._registry
        ]
        results = []
        for name, task in tasks:
            try:
                results.append({"tool": name, "result": await task, "status": "success"})
            except asyncio.TimeoutError:
                results.append({"tool": name, "result": {"error": "Timed out"}, "status": "timeout"})
            except Exception as e:
                results.append({"tool": name, "result": {"error": str(e)}, "status": "error"})
        return results
 
    async def _run(self, fn: callable, args: dict) -> Any:
        return await asyncio.wait_for(fn(**args), timeout=self.timeout)
 
# Usage
executor = ParallelToolExecutor(timeout_seconds=4.0)
executor.register("get_current_weather", get_current_weather)
executor.register("search_knowledge_base", search_knowledge_base)
 
results = await executor.execute_parallel([
    {"name": "get_current_weather", "args": {"city": "Tokyo"}},
    {"name": "search_knowledge_base", "args": {"query": "return policy", "category": "support"}},
])

[
  {"tool": "get_current_weather", "result": {"city": "Tokyo", "temperature": 18, "condition": "Sunny"}, "status": "success"},
  {"tool": "search_knowledge_base", "result": {"query": "return policy", "results": [...]}, "status": "success"}
]

async def keepalive_ping(websocket):
    while True:
        await asyncio.sleep(30)
        try:
            await websocket.ping()
        except Exception:
            break

gcloud projects add-iam-policy-binding YOUR_PROJECT_ID \
  --member="serviceAccount:YOUR-SA@YOUR_PROJECT_ID.iam.gserviceaccount.com" \
  --role="roles/aiplatform.user"

FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
ENV PORT=8080
CMD ["uvicorn", "server:app", "--host", "0.0.0.0", "--port", "8080"]

gcloud builds submit --tag gcr.io/YOUR_PROJECT_ID/voice-agent .
 
gcloud run deploy voice-agent \
  --image gcr.io/YOUR_PROJECT_ID/voice-agent \
  --platform managed \
  --region us-central1 \
  --timeout 3600 \
  --concurrency 80 \
  --min-instances 1 \
  --max-instances 20 \
  --set-env-vars GOOGLE_CLOUD_PROJECT=YOUR_PROJECT_ID \
  --allow-unauthenticated