adk-python-live

ADK Live and Voice Agents

Real-time voice and video agents built on ADK's **Gemini Live API Toolkit** (`Runner.run_live` + `LiveRequestQueue`). Use this for low-latency spoken conversation and interruption ("barge-in") — not for turn-based request/response agents (use a normal `Agent` with `run_async` for those).

**Official docs:** [ADK Live Docs](https://adk.dev/live/index.md) · [Configuration](https://adk.dev/live/configuration/index.md) · [Tools](https://adk.dev/live/tools/index.md) · [Get started](https://adk.dev/live/get-started/index.md)

---

1. When to use Live

| Use `run_live` (Live) | Use `run_async` (normal) | |-----------------------|--------------------------| | Voice/phone assistants, live captioning | Chatbots, tools, batch/RAG | | User can interrupt mid-response | One turn completes before the next | | Continuous audio/video input | Discrete text/multimodal messages | | Latency-critical spoken UX | Latency-tolerant |

> **Live agents can't be served over A2A, and can't be published to Gemini > Enterprise.** Both are request/response with no Live transport. This has > consequences for a scaffolded project beyond the model line — see > "Converting a scaffolded project to Live" next.

2. Converting a scaffolded project to Live

There is no Live scaffold template. Scaffold `adk` normally, then make these edits **together** — they are all consequences of one decision, and stopping after the first leaves a project whose `uv run pytest` fails.

| # | File | Edit | |---|------|------| | 1 | `app/agent.py` | Point `MODEL` at a Live model and, on Vertex, add `client_kwargs={"location": …}` to the `Gemini(...)` the scaffold already emits (see "Models"). No `.env` or Terraform change. | | 2 | `app/fast_api_app.py` | Delete the `attach_a2a_routes` call and its imports; delete `app/app_utils/a2a.py` and the `a2a-sdk` dependency. `/run_live` comes from `get_fast_api_app` and is untouched. | | 3 | `tests/integration/test_agent.py` | Rewrite onto `run_live` (below). | | 4 | `tests/integration/test_server_e2e.py` | Repoint the readiness probe, drop the A2A and SSE tests, add a `/run_live` test (below). | | 5 | — | Drop `agents-cli publish gemini-enterprise` from the plan; Gemini Enterprise has no Live transport to register against. |

Steps 3 and 4 are not optional cleanup. Both scaffolded integration tests drive the **non-live** path, and a Live model doesn't degrade to text — it rejects it, so `runner.run()` and `/run_sse` fail outright. Rewrite the tests; don't delete them.

`tests/integration/test_agent.py`

Same shape as the scaffolded test, but driven by `run_live` and asserting on the transcript rather than on text parts:

import asyncio

from google.adk.agents.live_request_queue import LiveRequestQueue
from google.adk.agents.run_config import RunConfig
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types

from app.agent import root_agent

APP_NAME = "test"
USER_ID = "test_user"


def test_agent_live_stream() -> None:
    """The agent answers over run_live and returns audio plus a transcript."""

    async def _turn() -> tuple[str, int]:
        session_service = InMemorySessionService()
        # The session must exist before run_live, or it raises "Session not found".
        session = await session_service.create_session(
            app_name=APP_NAME, user_id=USER_ID
        )
        runner = Runner(
            agent=root_agent, session_service=session_service, app_name=APP_NAME
        )
        queue = LiveRequestQueue()
        queue.send_content(
            types.Content(role="user", parts=[types.Part(text="Why is the sky blue?")])
        )

        transcript, audio_chunks = "", 0
        try:
            async for event in runner.run_live(
                user_id=USER_ID,
                session_id=session.id,
                live_request_queue=queue,
                # Transcription is on by default — see "RunConfig".
                run_config=RunConfig(response_modalities=["AUDIO"]),
            ):
                # Consume the finished aggregate; partials would duplicate text.
                if (
                    event.output_transcription
                    and event.output_transcription.finished
                    and event.output_transcription.text
                ):
                    transcript += event.output_transcription.text
                if event.content and event.content.parts:
                    audio_chunks += sum(
                        1 for part in event.content.parts if part.inline_data
                    )
                if event.turn_complete:
                    break
        finally:
            queue.close()
        return transcript, audio_chunks

    transcript, audio_chunks = asyncio.run(_turn())
    assert transcript.strip(), "Expected a non-empty output transcript"
    assert audio_chunks > 0, "Expected the agent to return audio"

`tests/integration/test_server_e2e.py`

Three edits, and the first is the one that's easy to miss:

• **Repoint the readiness probe.** `wait_for_server()` polls `AGENT_CARD_URL`,

which step 2 deleted — use `f"{BASE_URL}/list-apps"` instead. Miss this and the probe raises `NameError`, which the surrounding `except RequestException` does **not** catch, so every test in the file errors before the server is even contacted. (`/list-apps` is a sound readiness signal: uvicorn serves no request until the lifespan startup completes.)

• **Delete `test_adk_run_sse`, `test_a2a_chat_stream`, and `test_agent_card`**,

and add the `/run_live` test below. Those three are the whole file, so the new test replaces them rather than joining them.

• **On the `agent_runtime` target only**, also delete

`test_reasoning_engine_stream` — `async_stream_query` is the non-live path too.

import asyncio
import json
import uuid

import requests
from websockets.asyncio.client import connect

APP_NAME = "