How to Build a Voice Agent That Never Goes Silent with Google ADK: Streaming and Non-Blocking Tools

Read time: ~13 minutes.

Key facts:

  • In a live voice session, a tool that takes 13 seconds produces 13 seconds of dead air. Google’s ADK team measured that switching one blocking tool to a streaming tool cut time-to-first-words from 14.4 s to 1.2 s on gemini-3.8-live in their sales-report benchmark (Google Developers Blog, October 19, 2026).
  • ADK gives you two non-blocking patterns: streaming tools (an async def that yields instead of returns) and background tools (a normal tool with response_scheduling or behavior = NON_BLOCKING).
  • Result delivery has three modes: WHEN_IDLE (wait for a natural pause), INTERRUPT (deliver now), SILENT (add to context, don’t announce).
  • Version gates from the ADK changelog: response_scheduling arrived in 2.4.0, per-yield scheduling in 2.8.0, the behavior field in 2.10.0. Google’s post lists google-adk >= 2.10.0 as the requirement. The current PyPI release is 2.12.0 (October 13, 2026), Python 3.11+.
  • Model support matters: ADK’s docs say Gemini 3.1 Flash Live does not support response_scheduling, and the Gemini API docs say asynchronous function calling is not yet supported on 3.1 Flash Live.

What we tested: we installed google-adk==2.12.0 (with google-genai 2.31.0) in a fresh uv virtualenv on Linux with Python 3.13, built the agent below, confirmed that ADK marks both the streaming tool and the WHEN_IDLE tool as NON_BLOCKING in the Live request, and confirmed adk web starts and lists the agent. We did not hold a paid voice session, so every latency number here is Google’s, not ours.

If you’ve built a voice agent on the Gemini Live API, you’ve probably hit this: the user asks for something that needs a database query or a RAG lookup, the model calls your tool, and then nothing happens until the tool returns. In a text chat that’s a spinner. In a voice call it sounds like the line dropped. Prompting the model to “say you’re working on it” doesn’t fix it, because the model’s turn is blocked waiting for the function response.

Google’s Agent Development Kit (ADK) has a built-in fix, and on October 19, 2026 Google published a walkthrough of it. This tutorial turns that into a project you can run: a small sales-analyst voice agent with one streaming tool, one background tool, a cancel tool, and the configuration and limits you need to know before you ship it. If you’re new to the site, the 7minai homepage has more step-by-step builder guides, and our Gemini 3.8 Live release notes cover pricing and audio formats for the model family used here.

1. Why a normal tool goes silent

In ADK, a standard tool is an async def that does its work and returns once. Google’s example:

from typing import Any

async def query_sales_database(region: str) -> dict[str, Any]:
    """Runs the quarterly sales report query for a region.

    Args:
      region: Sales region to report on, e.g. "EMEA" or "APAC".
    Returns:
      A dict containing report metrics upon completion.
    """
    report = await run_expensive_query(region)       # e.g., 13 seconds of work
    return {"status": "success", "report": report}   # Single return at the end

During a bidirectional Live session (run_live()), ADK’s loop awaits that return before the model can produce another turn. If the query takes 13 seconds, the user hears 13 seconds of silence. (Google Developers Blog)

ADK’s live tools documentation frames the choice like this:

Your situationUseHow
Tool returns in under a secondBlocking (the default)A normal return tool
Long wait with nothing to narrateNon-blocking toolSet response_scheduling (or behavior) on the tool
Long wait worth narratingStreaming toolyield progress from an async generator

Google’s rule of thumb from the post: if the wait has a story, stream it; if the wait is just a wait, run it in the background and let the result land when idle.

None of this makes the backend faster. A 13-second query still takes 13 seconds. What changes is that the agent keeps talking.

2. Prerequisites

  • Python 3.11 or newer. That’s the requires_python on the google-adk PyPI package.
  • google-adk 2.10.0 or newer. Google’s post lists google-adk >= 2.10.0. We used 2.12.0, the latest release on PyPI as of October 20, 2026.
  • A Live-capable Gemini model and credentials for one of two backends:
    • Google AI Studio: an API key (GOOGLE_API_KEY). Best for prototyping.
    • Gemini Enterprise Agent Platform: a Google Cloud project with the Agent Platform API enabled and gcloud auth login done.
  • A browser with a microphone for testing in adk web.

ADK’s supported-models page explains that a standard Gemini model won’t hold a bidirectional connection; you need a Live model. We cover which one to pick in step 4, because it decides whether non-blocking tools work at all.

3. Install ADK and create the project

ADK’s Python streaming quickstart uses a plain venv and pip:

python3 -m venv .venv
source .venv/bin/activate          # Windows PowerShell: .venv\Scripts\Activate.ps1
pip install google-adk

To pin the version we tested:

pip install "google-adk==2.12.0"
adk --version                      # adk, version 2.12.0

If you use uv, uv venv followed by uv pip install "google-adk==2.12.0" works the same way; that’s how we installed it.

Now create this layout (the quickstart’s structure, with our agent name):

adk-voice/
└── app/
    ├── .env
    └── sales_agent/
        ├── __init__.py
        └── agent.py

sales_agent/__init__.py contains one line:

from . import agent

adk web discovers agents by folder, and each agent.py must define a root_agent.

4. Pick the model and backend (read this before writing tools)

This is the step most likely to bite you, because non-blocking tool support is per model.

What ADK’s docs say (checked October 20, 2026). The supported-models page lists two Live models:

ModelAI Studio IDAgent Platform IDresponse_scheduling
Gemini 2.5 Flash Livegemini-2.5-flash-native-audio-preview-12-2025 (Preview)gemini-live-2.5-flash-native-audio (GA)Supported
Gemini 3.1 Flash Livegemini-3.1-flash-live-preview (Preview)Not availableNot supported — function calling is synchronous, so the model stays silent until you return

gemini-live-2.5-flash-native-audio is ADK’s LlmAgent.DEFAULT_LIVE_MODEL; we confirmed that value in the installed 2.12.0 package.

What Google’s newer docs add. The Gemini API models page lists gemini-3.8-live and gemini-3.8-live-extended-thinking, and describes Gemini 3.1 Flash Live as a “legacy Live API preview model” with a recommendation to update to Gemini 3.8 Live. Google’s ADK post ran all its benchmarks on gemini-3.8-live and lists the requirement as “gemini-3.8-live on Gemini Enterprise.” The Gemini API Live tools page separately notes that asynchronous function calling “is not yet supported in Gemini 3.1 Flash Live.”

Practical takeaway:

  • Use gemini-3.8-live if your backend offers it; that’s the model Google’s numbers come from.
  • Otherwise use Gemini 2.5 Flash Live (the ID depends on backend).
  • Do not use Gemini 3.1 Flash Live for this tutorial. Your streaming and background tools will behave like blocking tools.

ADK’s docs recommend reading the model ID from an environment variable instead of hard-coding it, because the same model has different IDs on each backend and Live models get deprecated. We follow that pattern.

The .env file

For Google AI Studio (from the quickstart, plus the model variable):

GOOGLE_GENAI_USE_ENTERPRISE=FALSE
GOOGLE_API_KEY=PASTE_YOUR_ACTUAL_API_KEY_HERE
DEMO_AGENT_MODEL=gemini-3.8-live

For Agent Platform:

GOOGLE_GENAI_USE_ENTERPRISE=TRUE
GOOGLE_CLOUD_PROJECT=PASTE_YOUR_ACTUAL_PROJECT_ID
GOOGLE_CLOUD_LOCATION=us-central1
DEMO_AGENT_MODEL=gemini-3.8-live

If gemini-3.8-live isn’t available to your project, set DEMO_AGENT_MODEL to the Gemini 2.5 Flash Live ID for your backend from the table above. Switching backends is only the GOOGLE_GENAI_USE_ENTERPRISE flag; no code changes.

Agent Platform gotcha: Live models are not available at GOOGLE_CLOUD_LOCATION=global. Use a regional endpoint such as us-central1, us-east1 or asia-northeast1. (ADK supported models)

5. Pattern 1: a streaming tool that narrates progress

When the work has meaningful stages, make the tool an async generator. Changing return to yield is the whole conversion. ADK’s docs: “ADK treats any async-generator tool as non-blocking automatically.” (ADK live tools)

Start sales_agent/agent.py with this tool, adapted from Google’s example:

import asyncio
import os
from typing import AsyncGenerator

from google.adk.agents import Agent
from google.adk.tools import FunctionTool
from google.genai import types


async def query_sales_database(region: str) -> AsyncGenerator[str, None]:
    """Runs the quarterly sales report. Call this ONCE; it pushes updates automatically.

    Args:
      region: Sales region to report on, e.g. "EMEA" or "APAC".
    """
    stages = [
        ("Connecting to the sales warehouse...", 1.5),
        ("Scanning order records...", 3.5),
        ("Aggregating revenue by product line...", 3.5),
        ("Formatting the report...", 1.5),
    ]
    try:
        for message, delay in stages:
            yield f"PROGRESS: {message}"
            await asyncio.sleep(delay)
        # Demo data. Replace with your real query.
        yield f"DONE: {region} Q2 revenue reached $4.81M, up 12.4% year over year."
    except asyncio.CancelledError:
        # stop_streaming lands here: close connections, then re-raise.
        raise

The requirements for any streaming tool, per ADK: it must be an async function typed to return AsyncGenerator[T, None], where T is the type you yield.

What happens at runtime, according to Google’s post:

  1. Immediate acknowledgement. When the model calls the tool, ADK immediately sends an asynchronous status event, so the agent can speak right away (“Looking into the EMEA figures now…”).
  2. Partial updates. Each yield sends a partial function response. The model paraphrases each one, though narration can trail the tool by several seconds.
  3. Parallel conversation. If the user asks something unrelated mid-query, the agent can answer without cancelling the background task.

The $4.81M figure is demo data from Google’s example. Replace the asyncio.sleep stages with your real work.

Make it single-flight

Google warns that because streaming tools push updates across several turns, models “may occasionally attempt to call the tool again if prompted repeatedly.” Two defences from the post: keep “Call this ONCE” in the docstring (the model reads it), and add an in-memory latch or idempotency key in the tool itself for production. A simple version is a module-level set of regions currently running that the tool checks before starting a second query.

6. Pattern 2: a background tool that just waits

Some waits have nothing worth narrating: an export, a batch job, a media generation call. Progress chatter would just interrupt. You have two options.

Option A: yield once

The fastest setup is an async generator that yields only at the end:

async def query_sales_database(region: str) -> AsyncGenerator[dict, None]:
    """Runs the quarterly sales report. Call this ONCE; results arrive when ready."""
    report = await run_expensive_query(region)   # Background execution (~13s)
    yield report                                 # Single yield upon completion

Because it’s a generator, ADK issues an immediate pending response. The agent acknowledges (“I’m pulling that report together”), stays quiet while it runs, then presents the result. (Google Developers Blog)

Option B: keep a normal coroutine and set response_scheduling

If you’d rather keep async def ... -> dict, wrap it in FunctionTool and set when the response should land. Add this to agent.py:

async def export_report(region: str) -> dict:
    """Exports the quarterly report as a file. Call this ONCE; it returns when done.

    Args:
      region: Sales region to export.
    """
    await asyncio.sleep(13)  # stand-in for a slow, plain return-once job
    return {"status": "done", "region": region}


export_tool = FunctionTool(export_report)
export_tool.response_scheduling = types.FunctionResponseScheduling.WHEN_IDLE

In ADK 2.10+ you can instead set export_tool.behavior = types.Behavior.NON_BLOCKING to run it in the background without choosing a scheduling mode. ADK’s own non-blocking sample now uses behavior, per a 2.12.0 changelog entry.

The trade-off Google calls out: with a return-once coroutine, the model receives nothing until the function finishes, so it usually won’t say an initial acknowledgement unless the user keeps talking. Use Option B when you expect the user to move on to other questions and you want the result folded in at the next pause. If you need an instant “on it”, use Option A.

7. Control when results land: WHEN_IDLE, INTERRUPT, SILENT

response_scheduling decides when the model reacts to an incoming tool response. The three values (ADK live tools):

ValueBehaviorUse it for
WHEN_IDLEWaits for a natural pauseReports and lookups, the usual choice
INTERRUPTDelivers immediatelyAlarms, failures, “the transfer failed”
SILENTEnters context, announced only if relevantBackground info the model may use later

In the installed package, types.FunctionResponseScheduling has exactly these members plus SCHEDULING_UNSPECIFIED, and types.Behavior has UNSPECIFIED, BLOCKING and NON_BLOCKING.

What Google observed on gemini-3.8-live:

  • WHEN_IDLE: a sales report arrived while the model was answering a separate 20-second question. It waited, then delivered the figures when the answer finished.
  • INTERRUPT: the result cut off speech mid-sentence; the model then finished its answer and appended the figures.
  • SILENT: the result went into history and was available for later turns, with no unrequested announcement.

Mix modes inside one streaming tool (ADK 2.8+)

Execution shape (streaming vs return-once) and delivery timing are independent. Setting response_scheduling on a streaming tool stamps every partial update with that mode. Since ADK 2.8.0 (“allow streaming tools to set response scheduling per yield” in the changelog), a streaming tool can choose per update by yielding a types.FunctionResponse:

# Acknowledge and keep intermediate progress quiet:
yield types.FunctionResponse(
    response={"stage": "aggregating"},
    scheduling=types.FunctionResponseScheduling.SILENT,
)
# Deliver the final result when the speaker finishes:
yield types.FunctionResponse(
    response={"report": final_data},
    scheduling=types.FunctionResponseScheduling.WHEN_IDLE,
)

This is the pattern to reach for when you want progress in the model’s context without spoken updates, then a clean spoken result at the end. If you yield types.FunctionResponse, change the generator’s type hint to AsyncGenerator[types.FunctionResponse, None].

8. Let users cancel with stop_streaming

ADK reserves a tool name, stop_streaming. Add an empty function with that name to the agent’s tools, and when the model calls it ADK intercepts the call and cancels the running generator’s asyncio.Task. Your tool sees asyncio.CancelledError, which is why the streaming tool above has a try/except where you can close connections or file handles before re-raising. (ADK live tools, Google Developers Blog)

def stop_streaming(function_name: str):
    """Stop a running streaming tool.

    Args:
      function_name: The name of the streaming function to stop.
    """

In the 2.12.0 source (flows/llm_flows/tools/_live_caller.py), ADK cancels the task and waits up to 1.0 second for it to finish. If it hasn’t stopped by then, the model gets back a status saying the task “is not cancelled yet.” Keep your cleanup fast.

Barge-in is not cancellation. When the user talks over the agent, the Live API stops the model’s audio and ADK emits an event with interrupted=True so your client can stop playback. Running streaming or background tools keep going and their results are still delivered. The only way to stop a tool is for the model to call stop_streaming, so tell it when to do that in the instruction.

9. Assemble the agent

Finish agent.py with the agent definition:

root_agent = Agent(
    name="sales_analyst",
    model=os.getenv("DEMO_AGENT_MODEL", "gemini-live-2.5-flash-native-audio"),
    tools=[query_sales_database, export_tool, FunctionTool(stop_streaming)],
    instruction=(
        "You are a helpful sales assistant. When the user asks about sales data, "
        "call query_sales_database once. Acknowledge the request immediately, "
        "briefly paraphrase each progress update, and present the final figures "
        "when completed. When the user asks to export a report, call export_report "
        "once and keep talking about anything else they ask. If the user says to "
        "cancel the sales query, call stop_streaming with "
        "function_name='query_sales_database'. Keep answers conversational and concise."
    ),
)

The first instruction sentence is Google’s; we added the export and cancel lines. The default model is ADK’s DEFAULT_LIVE_MODEL, used only if DEMO_AGENT_MODEL isn’t set.

Check that the tools really are non-blocking

You don’t need a voice session to confirm the wiring. In 2.12.0, ADK’s mark_live_async_tools_non_blocking() sets behavior = NON_BLOCKING on the Live function declaration for streaming tools and for tools with response_scheduling or behavior set. When we ran it against this agent’s tools, the declarations came out as:

query_sales_database Behavior.NON_BLOCKING
export_report Behavior.NON_BLOCKING
stop_streaming None

That’s what you want: two background-capable tools, and the cancel tool left as a normal call. This uses an internal ADK module, so treat it as a debugging check, not production code.

10. Run it in adk web and test by voice

From the app/ folder (ADK’s quickstart says to run from there):

cd app
export SSL_CERT_FILE=$(python3 -m certifi)   # Windows PowerShell: $env:SSL_CERT_FILE = (python3 -m certifi)
adk web

The quickstart says SSL_CERT_FILE is required for the voice and video tests. Open the URL adk web prints (usually http://localhost:8000), pick sales_agent, reload the page, click the microphone button and talk. Use adk web --port 8765 if 8000 is taken; on our test run, GET /list-apps returned ["sales_agent"].

Try these four checks, one for each behavior:

  1. Streaming: “How did EMEA do last quarter?” You should hear an acknowledgement within a second or two, then short progress paraphrases, then the figures.
  2. Parallel talk: while it’s running, ask “What’s the capital of Japan?” The agent should answer without abandoning the report.
  3. Background with WHEN_IDLE: “Export the APAC report,” then keep asking questions. The export result should arrive at a pause, not mid-sentence.
  4. Cancel: start the EMEA query again and say “never mind, cancel that.” The model should call stop_streaming, and the report shouldn’t arrive.

Google also hosts an interactive demo that runs the same 13-second query through every pattern so you can hear the differences.

adk web is for development only. ADK’s docs say not to use it in production; the next step there is building a custom server around runner.run_live() or using the LiveKit integration.

11. Limits and costs to plan for

Session limits (ADK’s supported-models page, which says to verify against upstream docs because they change):

LimitAI StudioAgent Platform
Session duration, audio-only15 min15 min
Session duration, audio + video2 min2 min
Connection lifetime~10 min~10 min
Concurrent sessionsSee rate limitsUp to 1,000 per project on pay-as-you-go

Agent Platform also caps a conversation session at 10 minutes by default. Enabling context window compression lets a session run past the duration limits. A long-running background tool doesn’t change these limits, so a tool that needs minutes of work should hand off to a job queue and report back rather than hold the live session open.

Pricing for gemini-3.8-live on the Gemini API paid tier (pricing page, checked October 20, 2026): audio input $3.00 per 1M tokens or $0.005/min, audio output $12.00 per 1M tokens or $0.018/min, text input $0.75 and text output $4.50 per 1M tokens. The free tier is listed as free of charge. Progress updates the model speaks aloud are output audio, so chatty streaming tools cost more than SILENT updates.

Audio formats if you build your own client: the Live API takes raw 16-bit PCM at 16 kHz and returns 24 kHz PCM, over a stateful WebSocket. (Gemini Live API overview)

12. Troubleshooting

The agent still goes silent while the tool runs. Check the model first. On Gemini 3.1 Flash Live, function calling is synchronous and response_scheduling is not supported, so every tool blocks. Switch DEMO_AGENT_MODEL to gemini-3.8-live or Gemini 2.5 Flash Live. Then check your ADK version: response_scheduling needs 2.4+, behavior needs 2.10+.

My .env model is ignored and the default model is used. If you run your own script, call load_dotenv() before importing the agent module. The agent reads os.getenv("DEMO_AGENT_MODEL") at import time, so loading .env afterwards is too late. (ADK supported models)

The model calls the streaming tool twice. Keep “Call this ONCE” in the docstring and add a latch or idempotency key in the tool. Google flags this as a known behavior of multi-turn streaming tools.

Text messages fail in adk web. ADK’s quickstart notes you can’t use text chat with native-audio models in adk web. Use the microphone.

NotImplementedError from _make_subprocess_transport on Windows. Run adk web --no-reload, as the quickstart suggests.

Agent Platform returns errors at connect time. Make sure GOOGLE_CLOUD_LOCATION is a region like us-central1, not global.

Moving from Gemini 2.5 Flash Live to a 3.x Live model breaks the session. ADK’s notes for moving from 2.5 to 3.1 Flash Live say the most common upgrade failure is leaving RunConfig.proactivity or RunConfig.enable_affective_dialog set; remove them. Also iterate over event.content.parts instead of reading parts[0], because one server event can carry several parts. Check the same settings if you move to a newer 3.x Live model.

A cancelled tool keeps running. stop_streaming waits about one second for the task to stop. If your cleanup blocks (for example a synchronous database close), move it off the event loop or make it faster.

13. FAQ

Do I need ADK for this, or can I use the raw Live API? The raw Gemini Live API supports the same idea: you mark a function declaration "behavior": "NON_BLOCKING" and put "scheduling": "INTERRUPT" (or WHEN_IDLE / SILENT) in the function response. But the Live API doesn’t handle tool responses for you, so you write the plumbing. ADK runs tools inside run_live(), marks streaming tools non-blocking and handles cancellation. (Gemini Live tools, ADK live overview)

Streaming tool or response_scheduling? Stream when the stages are worth hearing about or you want an instant acknowledgement. Use response_scheduling (or behavior = NON_BLOCKING) when the wait has nothing to say and the user is likely to keep talking.

Does barge-in stop my tool? No. It stops the model’s audio and sets interrupted=True on an event. Tools keep running until they finish or the model calls stop_streaming.

Can a streaming tool watch video? Yes. Add an input_stream: LiveRequestQueue parameter and ADK feeds the user’s real-time input into a queue for that tool. ADK’s live tools page has a people-counting example.

Is this Python-only? Streaming tools are documented for Python and Java in ADK. This tutorial covers Python.

Where do Google’s 14.4 s and 1.2 s numbers come from? From Google’s own sales-report benchmark on gemini-3.8-live, reported in the October 19 post. We didn’t reproduce them. Your numbers depend on your model, region and network.

Next steps

Sources: Google Developers Blog: Build a Voice Agent That Never Goes Silent using ADK (Oct 19, 2026) · ADK: Live and voice agents · ADK: Tools for live agents · ADK: Supported models for live agents · ADK: Python streaming quickstart · adk-python CHANGELOG · google-adk on PyPI · Gemini API: Tool use with Live API · Gemini API models · Gemini API pricing