Configure ADK bidi-streaming for real-time multimodal interactions. Use when building live voice/video agents, implementing real-time streaming, configuring LiveRequestQueue, setting up audio/video processing, or when user mentions bidi-streaming, real-time agents, streaming tools, multimodal streaming, or Gemini Live API.
/plugin marketplace add vanman2024/ai-dev-marketplace/plugin install google-adk@ai-dev-marketplaceThis skill is limited to using the following tools:
examples/multi-agent-handoff.pyexamples/streaming-tool-agent.pyexamples/video-agent.pyexamples/voice-agent.pyscripts/check-modality-support.pyscripts/test-liverequest-queue.pyscripts/validate-streaming-config.pytemplates/audio-config.pytemplates/bidi-streaming-config.pytemplates/event-handler.pytemplates/liverequest-queue.pytemplates/streaming-tool-template.pytemplates/video-config.pyComprehensive patterns for configuring Google ADK bidi-streaming (bidirectional streaming) to build real-time, multimodal AI agents with voice, video, and streaming tool capabilities.
ADK bidi-streaming enables low-latency, bidirectional communication between users and AI agents with:
from google.adk.agents import Agent
from google.adk.agents.run_config import RunConfig, StreamingMode
from google.genai import types
# Configure for audio streaming
run_config = RunConfig(
response_modalities=["AUDIO"],
streaming_mode=StreamingMode.BIDI
)
# Run agent with bidi-streaming
async for event in agent.run_live(request_queue, run_config=run_config):
if event.server_content:
# Handle streaming response
handle_response(event)
from google.adk.agents import LiveRequestQueue
# Create queue for multimodal inputs
request_queue = LiveRequestQueue()
# Enqueue text
await request_queue.put("What's the weather?")
# Enqueue audio chunks
await request_queue.put(audio_bytes)
# Signal activity boundaries
await request_queue.put(types.LiveClientRealtimeInput(
media_chunks=[types.LiveClientRealtimeInputMediaChunk(
data=audio_chunk
)]
))
CRITICAL: Only ONE response modality per session. Cannot switch mid-session.
# Audio output (voice agent)
RunConfig(response_modalities=["AUDIO"])
# Text output (chat agent)
RunConfig(response_modalities=["TEXT"])
Session Resumption (automatic reconnection):
RunConfig(
session_resumption=types.SessionResumptionConfig()
)
Context Window Compression (unlimited sessions):
RunConfig(
context_window_compression=types.ContextWindowCompressionConfig(
trigger_tokens=100000,
sliding_window=types.SlidingWindow(target_tokens=80000)
)
)
See templates/audio-config.py for speech and transcription settings.
Use environment variable (no code changes needed):
# Google AI Studio (Gemini Live API)
export GOOGLE_GENAI_USE_VERTEXAI=FALSE
# Vertex AI (Live API)
export GOOGLE_GENAI_USE_VERTEXAI=TRUE
Define tools as async generators for continuous results:
@streaming_tool
async def monitor_stock(symbol: str):
"""Stream real-time stock price updates."""
while True:
price = await fetch_current_price(symbol)
yield f"Current price: ${price}"
await asyncio.sleep(1)
See templates/streaming-tool-template.py for complete pattern.
Process events from run_live():
async for event in agent.run_live(request_queue, run_config=run_config):
# Server content (agent responses)
if event.server_content:
if event.server_content.model_turn:
# Text/audio from model
process_model_response(event.server_content.model_turn)
if event.server_content.turn_complete:
# Agent finished speaking
handle_turn_complete()
# Tool calls
if event.tool_call:
# ADK executes tools automatically
log_tool_execution(event.tool_call)
# Interruptions
if event.interrupted:
handle_interruption()
Transfer stateful sessions between agents:
# Agent 1 creates session
session = await agent1.run_live(request_queue, run_config=config)
# Transfer to Agent 2 (seamless handoff)
await agent2.run_live(
request_queue,
run_config=config,
session=session # Maintains conversation context
)
Configure RunConfig
Create LiveRequestQueue
Implement Event Handling
Define Streaming Tools (optional)
Test and Deploy
See examples/audio-video-agent.py for complete multimodal setup including:
All templates use placeholders only (no hardcoded API keys):
Utility scripts for validation and setup:
Real-world streaming agent implementations:
See examples/voice-agent.py
See examples/streaming-tool-agent.py
See examples/multi-agent-handoff.py
This skill follows strict security rules:
.gitignore protection documentedThis skill should be used when the user asks to "create an agent", "add an agent", "write a subagent", "agent frontmatter", "when to use description", "agent examples", "agent tools", "agent colors", "autonomous agent", or needs guidance on agent structure, system prompts, triggering conditions, or agent development best practices for Claude Code plugins.
This skill should be used when the user asks to "create a slash command", "add a command", "write a custom command", "define command arguments", "use command frontmatter", "organize commands", "create command with file references", "interactive command", "use AskUserQuestion in command", or needs guidance on slash command structure, YAML frontmatter fields, dynamic arguments, bash execution in commands, user interaction patterns, or command development best practices for Claude Code.
This skill should be used when the user asks to "create a hook", "add a PreToolUse/PostToolUse/Stop hook", "validate tool use", "implement prompt-based hooks", "use ${CLAUDE_PLUGIN_ROOT}", "set up event-driven automation", "block dangerous commands", or mentions hook events (PreToolUse, PostToolUse, Stop, SubagentStop, SessionStart, SessionEnd, UserPromptSubmit, PreCompact, Notification). Provides comprehensive guidance for creating and implementing Claude Code plugin hooks with focus on advanced prompt-based hooks API.