bidirectional-tool-streaming

bidirectional-tool-streaming

Use tools inside a Gemini Live streaming session so the agent can fetch data, run actions, or call APIs without interrupting the live audio/video stream.

When to use

• Voice agent that books appointments mid-call
• Vision agent that logs observations while watching
• Live coding assistant that runs code while streaming explanation
• Any streaming flow where tools must fire without blocking response

Pattern: streaming agent with tools

from google.adk.agents import LlmAgent
from google.adk.streaming import StreamingMode

def book_appointment(date: str, time: str, name: str) -> dict:
    """Book an appointment.

    Args:
        date: ISO date (YYYY-MM-DD).
        time: 24-hour time (HH:MM).
        name: Patron name.

    Returns:
        {'confirmation_id': str, 'success': bool}.
    """
    return {"confirmation_id": "APT-1234", "success": True}

root_agent = LlmAgent(
    name="voice_concierge",
    model="gemini-2.5-flash-live",
    instruction=(
        "Book appointments by voice. Confirm date/time/name verbally before booking. "
        "Use the book_appointment tool only after the user confirms."
    ),
    streaming_mode=StreamingMode.BIDI,
    response_modalities=["AUDIO"],
    tools=[book_appointment],
)

ADK handles streaming tool calls automatically — Gemini Live signals a tool call, ADK executes, returns the result inline, and the agent continues speaking.

Streaming-aware tools (yield progress)

For long tools, stream partial results:

from typing import AsyncGenerator

async def deep_search(query: str) -> AsyncGenerator[dict, None]:
    """Search and stream results as they arrive."""
    yield {"status": "searching"}
    async for hit in search_api.stream(query):
        yield {"hit": hit.title, "url": hit.url}
    yield {"status": "complete"}

The agent receives each yield and can speak progress updates.

Cancellation

If the user interrupts, ADK cancels in-flight tool calls:

from google.adk.callbacks import on_tool_cancelled

@on_tool_cancelled
async def cleanup(ctx, tool_name, args):
    print(f"Tool {tool_name} cancelled mid-execution. Cleaning up.")

Concurrent tool calls

Gemini Live can issue multiple tool calls in parallel; ADK runs them concurrently:

LlmAgent(
    ...,
    tools=[get_weather, get_stocks, get_news],
    parallel_tool_calls=True,
)

Validation

• Tool call doesn't pause audio output (test with a slow tool)
• Cancellation fires when user interrupts
• Parallel tools complete in approximately max(individual_time), not sum
• Tool results appear in the spoken response

See also

• tool-confirmation-hitl to require approval for streaming tool calls
• audio-streaming-agent for the underlying voice setup