ADK Streaming Tools: Non-Blocking Yield for Gemini Live

In a Gemini Live voice session, a standard tool call blocks the model until the function returns. If your tool takes 13 seconds to run a database query, the agent produces 13 seconds of dead air and users assume the call dropped. The fix in ADK is to change return to yield, turning your tool into an AsyncGenerator that ADK marks as NON_BLOCKING and streams partial responses back to the model as they arrive.

This works because ADK detects the generator return type and dispatches an immediate status acknowledgment to the model. Each yield sends a partial function response upstream, so the agent can acknowledge the request, narrate progress, and answer unrelated questions while the background work continues. Google's ADK benchmark on gemini-3.8-live cut the time to first words from 14.4 seconds to 1.2 seconds by switching from a blocking tool to a streaming tool.

Why standard tools block in Live sessions#

A standard ADK tool is an async def that returns a single value at the end:

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

When you pass this to an Agent and run a Live session with run_live, the execution loop awaits the return before generating further model turns. The model cannot say "I'm working on it" because its turn is blocked waiting for the function response payload. The silence is architectural, not a prompt problem. Adding instructions like "tell the user you are working on it" to the agent's system prompt does not help, because the model cannot generate any tokens until the tool returns.

The Live API uses bidirectional streaming over WebSockets. The model generates audio tokens incrementally, but it can only do so when it has the turn. A blocking tool call hands the turn to your function and waits. Until the function returns, the model has no opportunity to speak, ask clarifying questions, or respond to anything the user says. This is why a 13-second database query produces 13 seconds of complete silence, not 13 seconds of conversational buffering.

Changing return to yield for streaming tools#

The smallest change that unblocks the conversation is swapping return for yield and typing the function as an AsyncGenerator. ADK sees the generator signature, marks the tool declaration as NON_BLOCKING, and sends an immediate pending status to the model:

import asyncio
from typing import AsyncGenerator
from google.adk.agents import Agent

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".

    Yields:
        Progress updates while the query runs, followed by the final summary.
    """
    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),
    ]
    for message, delay in stages:
        yield f"PROGRESS: {message}"
        await asyncio.sleep(delay)

    yield f"DONE: {region} Q2 revenue reached $4.81M, up 12.4% year over year."

agent = Agent(
    name="sales_analyst",
    model="gemini-3.8-live",
    tools=[query_sales_database],
    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. Keep answers conversational and concise."
    ),
)

When the model triggers query_sales_database, ADK immediately dispatches an asynchronous status event. The agent speaks right away, something like "Looking into the EMEA figures for you now..." Each subsequent yield sends a partial function response upstream, and the model paraphrases each update as it arrives, though narration can trail the tool by several seconds.

The AsyncGenerator return type is what ADK uses to detect streaming tools. Make sure the type annotation is correct, not just the yield statement inside the function body.

If the user asks an unrelated question while the query runs, the agent answers it without canceling or interrupting the background task. This multitasking is what makes streaming tools feel natural in voice. The GPT-Live decisions integration guide covers related patterns for acting while the user speaks in other Live API contexts.

Running example: yield vs return#

The core difference between blocking and streaming is visible without ADK. Run this to see how a generator interleaves output with work while a coroutine holds everything until the end:

import asyncio

async def blocking_query():
    await asyncio.sleep(0.3)
    return "Done after 0.3s of silence"

async def streaming_query():
    for stage in ["Connecting", "Scanning", "Aggregating", "Formatting"]:
        yield f"PROGRESS: {stage}"
        await asyncio.sleep(0.1)
    yield "DONE: Results ready"

async def main():
    print("Blocking (return):")
    result = await blocking_query()
    print(f"  {result}")

    print("\nStreaming (yield):")
    async for update in streaming_query():
        print(f"  {update}")

asyncio.run(main())

The blocking version prints nothing until the full delay elapses. The streaming version prints each stage as it happens. In a Live voice session, that difference is the gap between a responsive agent and dead air. The Python typing documentation describes AsyncGenerator as a generic requiring two type parameters: the yield type and the send type, which is None when the generator does not receive values via send().

Background execution without progress narration#

Not every operation has a story to narrate. If the user asks for a report and immediately pivots to another topic, continuous progress updates can feel intrusive. Two patterns handle this.

Single-yield generator. Yield once at completion. ADK still issues an immediate pending response so the agent acknowledges the request, but no progress chatter follows:

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

FunctionTool with response_scheduling. Keep the standard coroutine and wrap it in a FunctionTool with scheduling config:

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

sales_tool = FunctionTool(query_sales_database)
sales_tool.response_scheduling = types.FunctionResponseScheduling.WHEN_IDLE

In ADK 2.10+, you can also set sales_tool.behavior = types.Behavior.NON_BLOCKING to run a tool in the background without modifying scheduling mode.

Both approaches let the agent stay responsive. The single-yield generator is the fastest to set up because it requires no FunctionTool wrapper or scheduling configuration. The FunctionTool approach gives you explicit control over when results enter the conversation without changing the function's implementation.

With response_scheduling on a return-once coroutine, the model receives no payload until the entire function completes. The agent typically does not utter an initial verbal acknowledgment unless the user continues speaking. This fits cases where the user immediately continues with other questions and you want the result woven into the dialogue at the next natural pause.

Controlling when results enter the conversation#

The response_scheduling attribute controls when the model reacts to an incoming tool response. ADK supports three scheduling behaviors:

In benchmark sessions on gemini-3.8-live, WHEN_IDLE let an incoming sales report wait politely while the model finished answering a separate 20-second question, then delivered the figures at the next natural pause. INTERRUPT cut off current speech to deliver the result, then the model finished its previous answer and appended the figures. SILENT added the result to history for subsequent turns without an unrequested announcement.

Think of response_scheduling as a politeness dial. WHEN_IDLE is the well-mannered assistant who waits for a conversational gap. INTERRUPT is the urgent tap on the shoulder. SILENT is the note left on the desk, available when needed but never announced.

You can set response_scheduling on a streaming tool, which stamps every partial update with that behavior. In ADK 2.8 and newer, streaming tools can dynamically specify scheduling per update by yielding types.FunctionResponse objects instead of plain strings.

Execution shape and delivery timing are independent. A streaming tool can use WHEN_IDLE to narrate progress politely, or INTERRUPT to cut in with urgent alerts. A single-yield background tool can use SILENT to update context without speaking.

Production gotchas#

Single-flight tool calls. Streaming tools push updates across multiple turns, and models may attempt to call the tool again if prompted repeatedly. Include a clear "Call this ONCE" directive in the docstring and consider an in-memory latch or idempotency key in production code.

User barge-in does not cancel tools. When the user talks over the agent, the Live API stops the model's output and ADK emits an event with interrupted=True so the client can stop audio playback. Running streaming tools are not cancelled, and their results are still delivered. To stop a tool, the model must call stop_streaming.

Built-in cancellation with stop_streaming. Include a function named stop_streaming(function_name: str) in the agent's tools list. ADK intercepts the call and cancels the generator's underlying asyncio.Task, raising asyncio.CancelledError so the function can close database connections or file handles cleanly.

Token costs add up. Each progress yield generates an additional model turn for paraphrasing. Keep progress stages concise to limit output token usage. The Gemini API pricing page lists gemini-3.8-flash at $0.75 per 1M input tokens and $3.75 per 1M output tokens through December 31, 2026, with prices doubling on January 1, 2027. The page does not list gemini-3.8-live pricing separately.

Framework and model requirements. Streaming tools require google-adk >= 2.10.0 and gemini-3.8-live on Gemini Enterprise. These features work in bidirectional live sessions using run_live. The models page on ai.google.dev lists gemini-3.8-live as the default Live API model for low-latency voice agent experiences without reasoning delays. For general Gemini API setup in Python, the Gemini API Python guide covers installation and streaming basics. If you are building custom tool servers, the MCP server Python guide walks through testing tools locally.

Quick reference checklist#

• Change async def ... -> dict to async def ... -> AsyncGenerator[str, None]
• Replace return value with yield value (one yield for background, many for streaming)
• Add "Call this ONCE" to the docstring to prevent duplicate calls
• Include stop_streaming(function_name: str) as a tool for user-initiated cancellation
• Use response_scheduling = WHEN_IDLE for polite delivery, INTERRUPT for urgent alerts
• Set behavior = types.Behavior.NON_BLOCKING on FunctionTool in ADK 2.10+ for background execution
• Require google-adk >= 2.10.0 and model="gemini-3.8-live" for streaming tool support

Quick answers