# Google Adk Streaming

> ADK bidi-streaming / Live API. Use when building real-time audio/video streaming agents — WebSocket connections, run_live, audio transcription, and streaming tools.

- Skill: `eagleisbatman/google-adk-streaming` (Agent Skill)
- Install (CLI): `npx skillmds@latest add eagleisbatman/google-adk-streaming`
- Raw SKILL.md: https://api.skillmd.com/api/skills/eagleisbatman/google-adk-streaming/raw
- Safety review: pending
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: Integrations & APIs
- Author: eagleisbatman (https://skillmd.com/u/eagleisbatman)
- Updated: 2026-09-22
- Page: https://skillmd.com/skills/eagleisbatman/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

| Mode | Method | Use Case |
|------|--------|----------|
| Sync | `runner.run()` | Local testing, simple scripts |
| Async | `runner.run_async()` | Production, API servers |
| Live | `runner.run_live()` | Real-time streaming (audio/video) |

## Basic Streaming (SSE)

For server-sent events streaming (text):

```python
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)

```python
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

```python
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:

```bash
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:

```python
# 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:

```python
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

```python
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

| Voice | Description |
|-------|-------------|
| Aoede | Bright and clear |
| Charon | Deep and authoritative |
| Fenrir | Warm and friendly |
| Kore | Professional |
| Puck | Energetic |

## 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