Google Adk Streaming

Google ADK — Streaming (Live API)

Overview

ADK Live enables bidirectional streaming for real-time voice/video interactions. Built on top of Gemini Live API.

Runner Modes

Basic Streaming (SSE)

For server-sent events streaming (text):

from google.adk.agents import Agent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types

agent = Agent(
    name="streaming_agent",
    model="gemini-2.5-flash",
    instruction="You are a helpful assistant.",
)

runner = Runner(
    agent=agent,
    session_service=InMemorySessionService(),
    app_name="my_app",
)

session = await session_service.create_session(app_name="my_app", user_id="user1")

message = types.Content(role="user", parts=[types.Part(text="Tell me a story.")])

# Stream events as they arrive
async for event in runner.run_async(
    session_id=session.id,
    user_id="user1",
    new_message=message,
):
    if event.content and event.content.parts:
        for part in event.content.parts:
            if part.text:
                print(part.text, end="", flush=True)

Live Streaming (Bidi-Streaming / Audio)

from google.adk.agents import Agent, RunConfig, LiveRequestQueue, LiveRequest
from google.adk.agents.run_config import StreamingMode
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types

agent = Agent(
    name="voice_agent",
    model="gemini-2.5-flash",
    instruction="You are a voice assistant. Respond naturally to spoken queries.",
)

runner = Runner(
    agent=agent,
    session_service=InMemorySessionService(),
    app_name="voice_app",
)

# Configure for live streaming
run_config = RunConfig(
    streaming_mode=StreamingMode.BIDI,
    speech_config=types.SpeechConfig(
        voice_config=types.VoiceConfig(
            prebuilt_voice_config=types.PrebuiltVoiceConfig(
                voice_name="Aoede"
            )
        )
    ),
)

# Create live session
session = await session_service.create_session(app_name="voice_app", user_id="user1")

# Create request queue for sending audio
request_queue = LiveRequestQueue()

# Start live stream
async for event in runner.run_live(
    session_id=session.id,
    user_id="user1",
    live_request_queue=request_queue,
    run_config=run_config,
):
    # Handle streamed audio/text responses
    if event.content:
        for part in event.content.parts:
            if part.inline_data:
                # Audio data
                process_audio(part.inline_data.data)
            elif part.text:
                print(part.text)

Sending Audio Input

import base64
from google.adk.agents import LiveRequest
from google.genai import types

# Send audio chunk to the live stream
audio_chunk = read_microphone_chunk()  # Your audio capture

request_queue.send(
    LiveRequest(
        content=types.Content(
            role="user",
            parts=[types.Part(
                inline_data=types.Blob(
                    mime_type="audio/pcm;rate=16000",
                    data=audio_chunk,
                )
            )],
        )
    )
)

# Close when done
request_queue.close()

FastAPI Streaming Endpoint

The adk api_server provides SSE streaming out of the box:

adk api_server my_agent/ --port 8080

SSE Endpoint

POST /run_sse
Content-Type: application/json

{
  "app_name": "my_app",
  "user_id": "user1",
  "session_id": "session_abc",
  "new_message": {
    "role": "user",
    "parts": [{"text": "Hello!"}]
  }
}

Response: Server-Sent Events stream of agent events.

WebSocket Streaming

For full-duplex audio streaming via WebSocket:

# The ADK web UI uses WebSocket for live streaming
# Custom WebSocket integration:
from google.adk.cli.fast_api import get_fast_api_app

app = get_fast_api_app(agents_dir="./my_agents")
# WebSocket endpoint available at /ws

Streaming Tools

Tools that stream partial results during execution:

from google.adk.agents import Agent
from google.adk.agents.active_streaming_tool import ActiveStreamingTool

# Active streaming tools can send intermediate results
# while still executing (useful for progress updates)

RunConfig for Streaming

from google.adk.agents import RunConfig
from google.adk.agents.run_config import StreamingMode
from google.genai import types

config = RunConfig(
    streaming_mode=StreamingMode.BIDI,  # or StreamingMode.SSE
    speech_config=types.SpeechConfig(
        voice_config=types.VoiceConfig(
            prebuilt_voice_config=types.PrebuiltVoiceConfig(
                voice_name="Aoede"  # Available: Aoede, Charon, Fenrir, Kore, Puck
            )
        )
    ),
    response_modalities=["AUDIO", "TEXT"],  # What the model outputs
    input_audio_transcription=types.AudioTranscriptionConfig(),  # Transcribe input audio
    output_audio_transcription=types.AudioTranscriptionConfig(),  # Transcription of audio output
)

Available Voices

Key Rules

• run_live() is for real-time bidirectional audio/video streaming
• run_async() with SSE is for text streaming (simpler)
• Live streaming requires Gemini models with Live API support
• Audio format: PCM 16-bit, 16kHz sample rate
• In multi-agent live scenarios, audio is transcribed to text for sub-agents
• Transcriptions are stored as Events in the session
• Audio artifacts are saved separately with references in Events
• WebSocket endpoint available via adk web for development

Related Skills

• google-adk-llm-agent — Agent configuration (streaming uses standard agents)
• google-adk-session — Sessions store transcription events