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.
View on GitHubFebruary 1, 2026
Select agents to install to:
npx add-skill https://github.com/vanman2024/ai-dev-marketplace/blob/main/plugins/google-adk/skills/streaming-patterns/SKILL.md -a claude-code --skill streaming-patternsInstallation paths:
.claude/skills/streaming-patterns/# ADK Streaming Patterns
Comprehensive patterns for configuring Google ADK bidi-streaming (bidirectional streaming) to build real-time, multimodal AI agents with voice, video, and streaming tool capabilities.
## Core Concepts
ADK bidi-streaming enables low-latency, bidirectional communication between users and AI agents with:
- **Real-time interaction**: Process and respond while user is still providing input
- **Natural interruption**: User can interrupt agent mid-response
- **Multimodal support**: Text, audio, and video inputs/outputs
- **Streaming tools**: Tools that yield intermediate results over time
- **Session persistence**: Maintain context across ~10-minute connection timeouts
## Quick Start Patterns
### 1. Basic Bidi-Streaming Setup
```python
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)
```
### 2. LiveRequestQueue Pattern
```python
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
)]
))
```
## Configuration Patterns
### Response Modalities
**CRITICAL**: Only ONE response modality per session. Cannot switch mid-session.
```python
# Audio output (voice agent)
RunConfig(response_modalities=["AUDIO"])
# Text output (chat agent)
RunConfig(response_modalities=["TEXT"])
```