# Part 3: Event handling with run_live()
The `run_live()` method is ADK's primary entry point for streaming conversations, implementing an async generator that yields events as the conversation unfolds. This part focuses on understanding and handling these events—the core communication mechanism that enables real-time interaction between your application, users, and AI models.
You'll learn how to process different event types (text, audio, transcriptions, tool calls), manage conversation flow with interruption and turn completion signals, serialize events for network transport, and leverage ADK's automatic tool execution. Understanding event handling is essential for building responsive streaming applications that feel natural and real-time to users.
Async Context Required
All `run_live()` code requires async context. See [Part 1: FastAPI Application Example](https://adk.dev/streaming/dev-guide/part1/#fastapi-application-example) for details and production examples.
## How run_live() Works
`run_live()` is an async generator that streams conversation events in real-time. It yields events immediately as they're generated—no buffering, no polling, no callbacks. Events are streamed without internal buffering. Overall memory depends on session persistence (e.g., in-memory vs database), making it suitable for both quick exchanges and extended sessions.
### Method Signature and Flow
**Usage:**
Source reference: runners.py
```python
# The method signature reveals the thoughtful design
async def run_live(
self,
*, # Keyword-only arguments
user_id: Optional[str] = None, # User identification (required unless session provided)
session_id: Optional[str] = None, # Session tracking (required unless session provided)
live_request_queue: LiveRequestQueue, # The bidirectional communication channel
run_config: Optional[RunConfig] = None, # Streaming behavior configuration
session: Optional[Session] = None, # Deprecated: use user_id and session_id instead
) -> AsyncGenerator[Event, None]: # Generator yielding conversation events
```
As its signature tells, every streaming conversation needs identity (user_id), continuity (session_id), communication (live_request_queue), and configuration (run_config). The return type—an async generator of Events—promises real-time delivery without overwhelming system resources.
```
sequenceDiagram
participant Client
participant Runner
participant Agent
participant LLMFlow
participant Gemini
Client->>Runner: runner.run_live(user_id, session_id, queue, config)
Runner->>Agent: agent.run_live(context)
Agent->>LLMFlow: _llm_flow.run_live(context)
LLMFlow->>Gemini: Connect and stream
loop Continuous Streaming
Gemini-->>LLMFlow: LlmResponse
LLMFlow-->>Agent: Event
Agent-->>Runner: Event
Runner-->>Client: Event (yield)
end
```
### Basic Usage Pattern
The simplest way to consume events from `run_live()` is to iterate over the async generator with a for-loop:
Demo implementation: main.py:225-233
```python
async for event in runner.run_live(
user_id=user_id,
session_id=session_id,
live_request_queue=live_request_queue,
run_config=run_config
):
event_json = event.model_dump_json(exclude_none=True, by_alias=True)
logger.debug(f"[SERVER] Event: {event_json}")
await websocket.send_text(event_json)
```
Session Identifiers
Both `user_id` and `session_id` must match the identifiers you used when creating the session via `SessionService.create_session()`. These can be any string values based on your application's needs (e.g., UUIDs, email addresses, custom tokens). See [Part 1: Get or Create Session](https://adk.dev/streaming/dev-guide/part1/#get-or-create-session) for detailed guidance on session identifiers.
### Connection Lifecycle in run_live()
The `run_live()` method manages the underlying Live API connection lifecycle automatically:
**Connection States:**
1. **Initialization**: Connection established when `run_live()` is called
1. **Active Streaming**: Bidirectional communication via `LiveRequestQueue` (upstream to the model) and `run_live()` (downstream from the model)
1. **Graceful Closure**: Connection closes when `LiveRequestQueue.close()` is called
1. **Error Recovery**: ADK supports transparent session resumption; enable via `RunConfig.session_resumption` to handle transient failures. See [Part 4: Live API Session Resumption](https://adk.dev/streaming/dev-guide/part4/#live-api-session-resumption) for details.
#### What run_live() Yields
The `run_live()` method yields a stream of `Event` objects in real-time as the agent processes user input and generates responses. Understanding the different event types helps you build responsive UIs that handle text, audio, transcriptions, tool calls, metadata, and errors appropriately. Each event type is explained in detail in the sections below.
| Event Type | Description |
| --------------------------------------------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------- |
| **[Text Events](#text-events)** | Model's text responses when using `response_modalities=["TEXT"]`; includes `partial`, `turn_complete`, and `interrupted` flags for streaming UI management |
| **[Audio Events with Inline Data](#audio-events)** | Raw audio bytes (`inline_data`) streamed in real-time when using `response_modalities=["AUDIO"]`; ephemeral (not persisted to session) |
| **[Audio Events with File Data](#audio-events-with-file-data)** | Audio aggregated into files and stored in artifacts; contains `file_data` references instead of raw bytes; can be persisted to session history |
| **[Metadata Events](#metadata-events)** | Token usage information (`prompt_token_count`, `candidates_token_count`, `total_token_count`) for cost monitoring and quota tracking |
| **[Transcription Events](#transcription-events)** | Speech-to-text for user input (`input_transcription`) and model output (`output_transcription`) when transcription is enabled in `RunConfig` |
| **[Tool Call Events](#tool-call-events)** | Function call requests from the model; ADK handles execution automatically |
| **[Error Events](#error-events)** | Not an event type: model and connection failures are raised out of the `async for` loop rather than yielded |
Source Reference
See the complete event type handling implementation in [`runners.py`](https://github.com/google/adk-python/blob/427a983b18088bdc22272d02714393b0a779ecdf/src/google/adk/runners.py)
#### When run_live() Exits
The `run_live()` event loop can exit under various conditions. Understanding these exit scenarios is crucial for proper resource cleanup and error handling:
| Exit Condition | Trigger | Graceful? | Description |
| ----------------------- | ------------------------------------------------------ | -------------------- | ------------------------------------------------------------------------------------- |
| **Manual close** | `live_request_queue.close()` | ✅ Yes | User explicitly closes the queue, sending `LiveRequest(close=True)` signal |
| **All agents complete** | Last agent in SequentialAgent calls `task_completed()` | ✅ Yes | After all sequential agents finish their tasks |
| **Session timeout** | Live API duration limit reached | ⚠️ Connection closed | Session exceeds maximum duration (see limits below) |
| **Early exit** | `end_invocation` flag set | ✅ Yes | Set during preprocessing, or by an agent's own `_run_live_impl()`, to terminate early |
| **Empty event** | Queue closure signal | ✅ Yes | Internal signal indicating event stream has ended |
| **Errors** | Connection errors, exceptions | ❌ No | Unhandled exceptions or connection failures |
SequentialAgent Behavior
When using `SequentialAgent`, the `task_completed()` function does NOT exit your application's `run_live()` loop. It only signals the end of the current agent's work, triggering a seamless transition to the next agent in the sequence. Your event loop continues receiving events from subsequent agents. The loop only exits when the **last** agent in the sequence completes.
Learn More
For session resumption and connection recovery details, see [Part 4: Live API Session Resumption](https://adk.dev/streaming/dev-guide/part4/#live-api-session-resumption). For multi-agent workflows, see [Best Practices for Multi-Agent Workflows](#best-practices-for-multi-agent-workflows).
#### Events Saved to ADK `Session`
Not all events yielded by `run_live()` are persisted to the ADK `Session`. When `run_live()` exits, only certain events are saved to the session while others remain ephemeral. Understanding which events are saved versus which are ephemeral is crucial for applications that use session persistence, resumption, or need to review conversation history.
Source Reference
See session event persistence logic in [`runners.py`](https://github.com/google/adk-python/blob/427a983b18088bdc22272d02714393b0a779ecdf/src/google/adk/runners.py)
**Events Saved to the ADK `Session`:**
These events are persisted to the ADK `Session` and available in session history:
- **Audio Events with File Data**: Saved to ADK `Session` only if `RunConfig.save_live_blob` is `True`; audio data is aggregated into files in artifacts with `file_data` references
- **Usage Metadata Events**: Always saved to track token consumption across the ADK `Session`
- **Non-Partial Transcription Events**: Final transcriptions are saved; partial transcriptions are not persisted
- **Function Call and Response Events**: Always saved to maintain tool execution history
- **Other Control Events**: Most control events (e.g., `turn_complete`, `interrupted`) are saved
**Events NOT Saved to the ADK `Session`:**
These events are ephemeral and only yielded to callers during active streaming:
- **Audio Events with Inline Data**: Raw audio `Blob` data in `inline_data` is never saved to the ADK `Session` (only yielded for real-time playback)
- **Partial Transcription Events**: Only yielded for real-time display; final transcriptions are saved
Audio Persistence
To save audio conversations to the ADK `Session` for review or resumption, enable `RunConfig.save_live_blob = True`. This persists audio streams to artifacts. See [Part 4: save_live_blob](https://adk.dev/streaming/dev-guide/part4/#save_live_blob) for configuration details.
## Understanding Events
Events are the core communication mechanism in ADK Gemini Live API Toolkit's streaming system. This section explores the complete lifecycle of events—from how they're generated through multiple pipeline layers, to concurrent processing patterns that enable true real-time interaction, to practical handling of interruptions and turn completion. You'll learn about event types (text, audio, transcriptions, tool calls), serialization strategies for network transport, and the connection lifecycle that manages streaming sessions across both Gemini Live API and Gemini Live API platforms.
### The Event Class
ADK's `Event` class is a Pydantic model that represents all communication in a streaming conversation. It extends `LlmResponse` and serves as the unified container for model responses, user input, transcriptions, and control signals.
Source Reference
See Event class implementation in [`event.py:30-128`](https://github.com/google/adk-python/blob/427a983b18088bdc22272d02714393b0a779ecdf/src/google/adk/events/event.py#L30-L128) and [`llm_response.py:28-200`](https://github.com/google/adk-python/blob/427a983b18088bdc22272d02714393b0a779ecdf/src/google/adk/models/llm_response.py#L28-L200)
#### Key Fields
**Essential for all applications:**
- `content`: Contains text, audio, or function calls as `Content.parts`
- `author`: Identifies who created the event (`"user"` or agent name)
- `partial`: Distinguishes incremental chunks from complete text
- `turn_complete`: Signals when to enable user input again
- `interrupted`: Indicates when to stop rendering current output
**For voice/audio applications:**
- `input_transcription`: User's spoken words (when enabled in `RunConfig`)
- `output_transcription`: Model's spoken words (when enabled in `RunConfig`)
- `content.parts[].inline_data`: Audio data for playback
**For tool execution:**
- `content.parts[].function_call`: Model's tool invocation requests
- `content.parts[].function_response`: Tool execution results
- `long_running_tool_ids`: Track async tool execution
**For debugging and diagnostics:**
- `usage_metadata`: Token counts and billing information
- `cache_metadata`: Context cache hit/miss statistics
- `turn_complete_reason`: Why the Live API ended the turn, when it reports one (e.g., RESPONSE_REJECTED, BLOCKLIST); `None` on a normal turn
- `finish_reason` / `error_code` / `error_message`: Inherited from `LlmResponse` and populated on the `run_async()` path only; always `None` under `run_live()`
Author Semantics
Transcription events have author `"user"`; model responses/events use the agent's name as `author` (not `"model"`). See [Event Authorship](#event-authorship) for details.
#### Understanding Event Identity
Events have two important ID fields:
- **`event.id`**: Unique identifier for this specific event (format: UUID). Each event gets a new ID, even partial text chunks.
- **`event.invocation_id`**: Shared identifier for all events in the current invocation (format: `"e-" + UUID`). In `run_live()`, all events from a single streaming session share the same invocation_id. (See [InvocationContext](#invocationcontext-the-execution-state-container) for more about invocations)
**Usage:**
```python
# All events in this streaming session will have the same invocation_id
async for event in runner.run_live(...):
print(f"Event ID: {event.id}") # Unique per event
print(f"Invocation ID: {event.invocation_id}") # Same for all events in session
```
**Use cases:**
- **event.id**: Track individual events in logs, deduplicate events
- **event.invocation_id**: Group events by conversation session, filter session-specific events
### Event Authorship
In live streaming mode, the `Event.author` field follows special semantics to maintain conversation clarity:
**Model responses**: Authored by the **agent name** (e.g., `"my_agent"`), not the literal string `"model"`
- This enables multi-agent scenarios where you need to track which agent generated the response
- Example: `Event(author="customer_service_agent", content=...)`
**User transcriptions**: Authored as `"user"` when the event contains transcribed user audio
**How it works**:
1. Gemini Live API returns user audio transcriptions with `content.role == 'user'`
1. ADK's `get_author_for_event()` function checks for this role marker
1. If `content.role == 'user'`, ADK sets `Event.author` to `"user"`
1. Otherwise, ADK sets `Event.author` to the agent name (e.g., `"my_agent"`)
This transformation ensures that transcribed user input is correctly attributed to the user in your application's conversation history, even though it flows through the model's response stream.
- Example: Input audio transcription → `Event(author="user", input_transcription=..., content.role="user")`
**Why this matters**:
- In multi-agent applications, you can filter events by agent: `events = [e for e in stream if e.author == "my_agent"]`
- When displaying conversation history, use `event.author` to show who said what
- Transcription events are correctly attributed to the user even though they flow through the model
Source Reference
See author attribution logic in [`base_llm_flow.py:674-708`](https://github.com/google/adk-python/blob/427a983b18088bdc22272d02714393b0a779ecdf/src/google/adk/flows/llm_flows/base_llm_flow.py#L674-L708)
### Event Types and Handling
ADK streams distinct event types through `runner.run_live()` to support different interaction modalities: text responses for traditional chat, audio chunks for voice output, transcriptions for accessibility and logging, and tool call notifications for function execution. Each event includes metadata flags (`partial`, `turn_complete`, `interrupted`) that control UI state transitions and enable natural, human-like conversation flows. Understanding how to recognize and handle these event types is essential for building responsive streaming applications.
### Text Events
The most common event type, containing the model's text responses when you specifying `response_modalities` in `RunConfig` to `["TEXT"]` mode:
**Usage:**
```python
async for event in runner.run_live(...):
if event.content and event.content.parts:
if event.content.parts[0].text:
text = event.content.parts[0].text
if event.partial:
# Your logic to update streaming display
update_streaming_display(text)
```
#### Default Response Modality Behavior
When `response_modalities` is not explicitly set (i.e., `None`), ADK automatically defaults to `["AUDIO"]` mode at the start of `run_live()`. This means:
- **If you provide no RunConfig**: Defaults to `["AUDIO"]`
- **If you provide RunConfig without response_modalities**: Defaults to `["AUDIO"]`
- **If you explicitly set response_modalities**: Uses your setting (no default applied)
**Why this default exists**: Some native audio models require the response modality to be explicitly set. To ensure compatibility with all models, ADK defaults to `["AUDIO"]`.
**For text-only applications**: Always explicitly set `response_modalities=["TEXT"]` in your RunConfig to avoid receiving unexpected audio events.
**Example:**
```python
# Explicit text mode
run_config = RunConfig(
response_modalities=["TEXT"],
streaming_mode=StreamingMode.BIDI
)
```
**Key Event Flags:**
These flags help you manage streaming text display and conversation flow in your UI:
- `event.partial`: `True` for incremental text chunks during streaming; `False` for complete merged text
- `event.turn_complete`: `True` when the model has finished its complete response
- `event.interrupted`: `True` when user interrupted the model's response
Learn More
For detailed guidance on using `partial` `turn_complete` and `interrupted` flags to manage conversation flow and UI state, see [Handling Text Events](#handling-text-events).
### Audio Events
When `response_modalities` is configured to `["AUDIO"]` in your `RunConfig`, the model generates audio output instead of text, and you'll receive audio data in the event stream:
**Configuration:**
```python
# Configure RunConfig for audio responses
run_config = RunConfig(
response_modalities=["AUDIO"],
streaming_mode=StreamingMode.BIDI
)
# Audio arrives as inline_data in event.content.parts
async for event in runner.run_live(..., run_config=run_config):
if event.content and event.content.parts:
part = event.content.parts[0]
if part.inline_data:
# Audio event structure:
# part.inline_data.data: bytes (raw PCM audio)
# part.inline_data.mime_type: str (e.g., "audio/pcm;rate=24000")
audio_data = part.inline_data.data
mime_type = part.inline_data.mime_type
print(f"Received {len(audio_data)} bytes of {mime_type}")
# Your logic to play audio
await play_audio(audio_data)
```
Learn More
- **`response_modalities` controls how the model generates output**—you must choose either `["TEXT"]` for text responses or `["AUDIO"]` for audio responses per session. You cannot use both modalities simultaneously. See [Part 4: Response Modalities](https://adk.dev/streaming/dev-guide/part4/#response-modalities) for configuration details.
- For comprehensive coverage of audio formats, sending/receiving audio, and audio processing flow, see [Part 5: How to Use Audio, Image and Video](https://adk.dev/streaming/dev-guide/part5/index.md).
### Audio Events with File Data
When audio data is aggregated and saved as files in artifacts, ADK yields events containing `file_data` references instead of raw `inline_data`. This is useful for persisting audio to session history.
Source Reference
See audio file aggregation logic in [`audio_cache_manager.py:156-178`](https://github.com/google/adk-python/blob/427a983b18088bdc22272d02714393b0a779ecdf/src/google/adk/flows/llm_flows/audio_cache_manager.py#L156-L178)
**Receiving Audio File References:**
```python
async for event in runner.run_live(
user_id=user_id,
session_id=session_id,
live_request_queue=queue,
run_config=run_config
):
if event.content and event.content.parts:
for part in event.content.parts:
if part.file_data:
# Audio aggregated into a file saved in artifacts
file_uri = part.file_data.file_uri
mime_type = part.file_data.mime_type
print(f"Audio file saved: {file_uri} ({mime_type})")
# Retrieve audio file from artifact service for playback
```
**File Data vs Inline Data:**
- **Inline Data** (`part.inline_data`): Raw audio bytes streamed in real-time; ephemeral and not saved to session
- **File Data** (`part.file_data`): Reference to audio file stored in artifacts; can be persisted to session history
When `RunConfig.save_live_blob` is `True`, both input and output audio data are aggregated into audio files and saved in the artifact service. The file reference is included in the event as `file_data`, allowing you to retrieve the audio later.
Session Persistence
To save audio events with file data to session history, enable `RunConfig.save_live_blob = True`. This allows audio conversations to be reviewed or replayed from persisted sessions.
### Metadata Events
Usage metadata events contain token usage information for monitoring costs and quota consumption. The `run_live()` method yields these events separately from content events.
Source Reference
See usage metadata structure in [`llm_response.py:105`](https://github.com/google/adk-python/blob/427a983b18088bdc22272d02714393b0a779ecdf/src/google/adk/models/llm_response.py#L105)
**Accessing Token Usage:**
```python
async for event in runner.run_live(
user_id=user_id,
session_id=session_id,
live_request_queue=queue,
run_config=run_config
):
if event.usage_metadata:
print(f"Prompt tokens: {event.usage_metadata.prompt_token_count}")
print(f"Response tokens: {event.usage_metadata.candidates_token_count}")
print(f"Total tokens: {event.usage_metadata.total_token_count}")
# Track cumulative usage across the session
total_tokens += event.usage_metadata.total_token_count or 0
```
**Available Metadata Fields:**
- `prompt_token_count`: Number of tokens in the input (prompt and context)
- `candidates_token_count`: Number of tokens in the model's response
- `total_token_count`: Sum of prompt and response tokens
- `cached_content_token_count`: Number of tokens served from cache (when using context caching)
Cost Monitoring
Usage metadata events allow real-time cost tracking during streaming sessions. You can implement quota limits, display usage to users, or log metrics for billing and analytics.
### Transcription Events
Input and output transcription are enabled by default in `RunConfig`, and you receive transcriptions as separate events:
**Configuration:**
```python
async for event in runner.run_live(...):
# User's spoken words (when input_audio_transcription enabled)
if event.input_transcription:
# Your logic to display user transcription
display_user_transcription(event.input_transcription)
# Model's spoken words (when output_audio_transcription enabled)
if event.output_transcription:
# Your logic to display model transcription
display_model_transcription(event.output_transcription)
```
These enable accessibility features and conversation logging without separate transcription services.
Learn More
For details on enabling transcription in `RunConfig` and understanding transcription delivery, see [Part 5: Audio Transcription](https://adk.dev/streaming/dev-guide/part5/#audio-transcription).
### Tool Call Events
When the model requests tool execution:
**Usage:**
```python
async for event in runner.run_live(...):
if event.content and event.content.parts:
for part in event.content.parts:
if part.function_call:
# Model is requesting a tool execution
tool_name = part.function_call.name
tool_args = part.function_call.args
# ADK handles execution automatically
```
ADK processes tool calls automatically—you typically don't need to handle these directly unless implementing custom tool execution logic.
Learn More
For details on how ADK automatically executes tools, handles function responses, and supports long-running and streaming tools, see [Automatic Tool Execution in run_live()](#automatic-tool-execution-in-run_live).
### Error Events
Failures in a live session are raised, not yielded. `run_live()` does not emit an error event—when the live connection fails, the exception propagates out of the `async for` loop. Error handling therefore belongs in a `try`/`except` around the loop, not in an `if` inside it.
`event.error_code` is not set by `run_live()`
`Event` inherits `error_code` and `error_message` from `LlmResponse`, so both attributes exist on every live event—but nothing on the `run_live()` path assigns them. `GeminiLlmConnection.receive()` never sets them, and `_postprocess_live()` only reads them. They are populated on the `run_async()` path only, where a blocking `finish_reason` or `prompt_feedback.block_reason` is mapped onto them. An `if event.error_code:` check inside a `run_live()` loop never fires.
**Usage:**
```python
import logging
from google.genai import errors
from websockets.exceptions import ConnectionClosed
logger = logging.getLogger(__name__)
try:
async for event in runner.run_live(...):
if event.content and event.content.parts:
# ... handle content
pass
except ConnectionClosed as e:
# The WebSocket dropped and ADK could not resume the session
logger.error(f"Live connection closed: {e}")
await websocket.send_json({"type": "error", "message": "Connection lost"})
except errors.APIError as e:
# The Live API rejected or failed the request (auth, quota, bad model, ...)
logger.error(f"Live API error {e.code}: {e.message}")
await websocket.send_json({"type": "error", "message": "Service error"})
finally:
queue.close() # Always cleanup connection
```
Whether the loop finishes naturally, breaks, or raises, `finally` ensures the connection closes properly.
**What ADK retries before the exception reaches you:**
ADK's live flow retries a dropped connection itself, but only when session resumption is enabled and a handle has already been cached:
| Failure | ADK behavior |
| ------------------------------------------------------------------------------ | ---------------------------------------------------------------------------------------------------------- |
| `ConnectionClosed` / `ConnectionClosedOK`, resumption handle cached | Reconnects with the handle, up to 5 times, then re-raises (the counter resets on every successful connect) |
| `ConnectionClosed` / `ConnectionClosedOK`, no handle | Re-raised immediately |
| `errors.APIError` with close code 1000, 1006 or 1011, resumption handle cached | Treated as a recoverable drop and reconnected the same way |
| Any other `errors.APIError` (including 1000/1006/1011 with no handle) | Re-raised immediately |
| Any other exception | Logged and re-raised |
Learn More
To let ADK ride out transient drops instead of raising, enable session resumption. See [Part 4: Live API Session Resumption](https://adk.dev/streaming/dev-guide/part4/#live-api-session-resumption).
**Reading why a turn ended:**
A model-side termination is not a transport error and does not raise. It arrives on the event itself:
- `event.turn_complete_reason`: a `types.TurnCompleteReason` carried on the `turn_complete=True` event when the Live API reports why generation stopped (for example `RESPONSE_REJECTED`, `PROHIBITED_INPUT_CONTENT`, `BLOCKLIST`, `GENERATED_CONTENT_SAFETY`, `MALFORMED_FUNCTION_CALL`). It is `None` when the turn ended normally.
- `event.interrupted`: `True` when the user barged in and the model stopped mid-response. See [Handling `interrupted` Flag](#handling-interrupted-flag).
**Usage:**
```python
async for event in runner.run_live(...):
if event.turn_complete and event.turn_complete_reason:
logger.warning(f"Turn ended early: {event.turn_complete_reason}")
await websocket.send_json({
"type": "turn_ended",
"reason": event.turn_complete_reason.name
})
```
**Best practices for error handling:**
- **Wrap the loop, not the events**: put the `try`/`except` around `async for`, and always close the queue in `finally`
- **Log errors with context**: Include session_id and user_id in error logs for debugging
- **Distinguish transport failures from turn terminations**: an exception ends the whole `run_live()` call, while `turn_complete_reason` only ends the current turn
- **Notify users gracefully**: Show user-friendly error messages instead of raw exception text
- **Enable session resumption**: it is what lets ADK reconnect instead of raising on a dropped connection
- **Monitor error rates**: Track exception types and frequencies to identify systemic issues
## Handling Text Events
Understanding the `partial`, `interrupted`, and `turn_complete` flags is essential for building responsive streaming UIs. These flags enable you to provide real-time feedback during streaming, handle user interruptions gracefully, and detect conversation boundaries for proper state management.
### Handling `partial`
This flag helps you distinguish between incremental text chunks and complete merged text, enabling smooth streaming displays with proper final confirmation.
**Usage:**
```python
async for event in runner.run_live(...):
if event.content and event.content.parts:
if event.content.parts[0].text:
text = event.content.parts[0].text
if event.partial:
# Your streaming UI update logic here
update_streaming_display(text)
else:
# Your complete message display logic here
display_complete_message(text)
```
**`partial` Flag Semantics:**
- `partial=True`: The text in this event is **incremental**—it contains ONLY the new text since the last event
- `partial=False`: The text in this event is **complete**—it contains the full merged text for this response segment
Note
The `partial` flag is meaningful for text content (`event.content.parts[].text`) and for transcriptions. For each content type:
- **Audio events**: Each audio chunk in `inline_data` is independent (no merging occurs)
- **Tool calls**: Function calls and responses are always complete (partial doesn't apply)
- **Transcriptions**: Incremental transcription chunks are yielded with `partial=True`; the final accumulated transcription is yielded with `partial=False`
**Example Stream:**
```text
Event 1: partial=True, text="Hello", turn_complete=None
Event 2: partial=True, text=" world", turn_complete=None
Event 3: partial=False, text="Hello world", turn_complete=None
Event 4: partial=None, content=None, turn_complete=True # Turn done
```
**Important timing relationships**:
- `partial=False` can occur **multiple times** in a turn (e.g., after each sentence)
- `turn_complete=True` occurs **once** at the very end of the model's complete response, in a **separate event**
- You may receive: `partial=False` (sentence 1) → `partial=False` (sentence 2) → `turn_complete=True`
- The merged text event (`partial=False` with content) is always yielded **before** the `turn_complete=True` event
Note
ADK internally accumulates all text from `partial=True` events. When you receive an event with `partial=False`, the text content equals the sum of all preceding `partial=True` chunks. This means:
- You can safely ignore all `partial=True` events and only process `partial=False` events if you don't need streaming display
- If you do display `partial=True` events, the `partial=False` event provides the complete merged text for validation or storage
- This accumulation is handled automatically by ADK's live connection layer (`GeminiLlmConnection.receive()`)—you don't need to manually concatenate partial text chunks
#### Handling `interrupted` Flag
This enables natural conversation flow by detecting when users interrupt the model mid-response, allowing you to stop rendering outdated content immediately.
When users send new input while the model is still generating a response (common in voice conversations), you'll receive an event with `interrupted=True`:
**Usage:**
```python
async for event in runner.run_live(...):
if event.interrupted:
# Your logic to stop displaying partial text and clear typing indicators
stop_streaming_display()
# Your logic to show interruption in UI (optional)
show_user_interruption_indicator()
```
**Example - Interruption Scenario:**
```text
Model: "The weather in San Francisco is currently..."
User: [interrupts] "Actually, I meant San Diego"
→ event.interrupted=True received
→ Your app: stop rendering model response, clear UI
→ Model processes new input
Model: "The weather in San Diego is..."
```
**When to use interruption handling:**
- **Voice conversations**: Stop audio playback immediately when user starts speaking
- **Clear UI state**: Remove typing indicators and partial text displays
- **Conversation logging**: Mark which responses were interrupted (incomplete)
- **User feedback**: Show visual indication that interruption was recognized
#### Handling `turn_complete` Flag
This signals conversation boundaries, allowing you to update UI state (enable input controls, hide indicators) and mark proper turn boundaries in logs and analytics.
When the model finishes its complete response, you'll receive an event with `turn_complete=True`:
**Usage:**
```python
async for event in runner.run_live(...):
if event.turn_complete:
# Your logic to update UI to show "ready for input" state
enable_user_input()
# Your logic to hide typing indicator
hide_typing_indicator()
# Your logic to mark conversation boundary in logs
log_turn_boundary()
```
**Event Flag Combinations:**
Understanding how `turn_complete` and `interrupted` combine helps you handle all conversation states:
| Scenario | turn_complete | interrupted | Your App Should |
| ----------------------------- | ------------- | ----------- | ---------------------------------------- |
| Normal completion | True | False | Enable input, show "ready" state |
| User interrupted mid-response | False | True | Stop display, clear partial content |
| Interrupted at end | True | True | Same as normal completion (turn is done) |
| Mid-response (partial text) | False | False | Continue displaying streaming text |
**Implementation:**
```python
async for event in runner.run_live(...):
# Handle streaming text
if event.content and event.content.parts and event.content.parts[0].text:
if event.partial:
# Your logic to show typing indicator and update partial text
update_streaming_text(event.content.parts[0].text)
else:
# Your logic to display complete text chunk
display_text(event.content.parts[0].text)
# Handle interruption
if event.interrupted:
# Your logic to stop audio playback and clear indicators
stop_audio_playback()
clear_streaming_indicators()
# Handle turn completion
if event.turn_complete:
# Your logic to enable user input
show_input_ready_state()
enable_microphone()
```
**Common Use Cases:**
- **UI state management**: Show/hide "ready for input" indicators, typing animations, microphone states
- **Audio playback control**: Know when to stop rendering audio chunks from the model
- **Conversation logging**: Mark clear boundaries between turns for history/analytics
- **Streaming optimization**: Stop buffering when turn is complete
**Turn completion and caching:** When `RunConfig.save_live_blob` is enabled, audio caches are flushed automatically at specific points during streaming:
- **On turn completion** (`turn_complete=True`): Both user and model audio caches are flushed
- **On interruption** (`interrupted=True`): Model audio cache is flushed
## Serializing Events to JSON
ADK `Event` objects are Pydantic models, which means they come with powerful serialization capabilities. The `model_dump_json()` method is particularly useful for streaming events over network protocols like WebSockets or Server-Sent Events (SSE).
### Using event.model_dump_json()
This provides a simple one-liner to convert ADK events into JSON format that can be sent over network protocols like WebSockets or SSE.
The `model_dump_json()` method serializes an `Event` object to a JSON string:
Demo implementation: main.py:219-234
```python
async def downstream_task() -> None:
"""Receives Events from run_live() and sends to WebSocket."""
async for event in runner.run_live(
user_id=user_id,
session_id=session_id,
live_request_queue=live_request_queue,
run_config=run_config
):
event_json = event.model_dump_json(exclude_none=True, by_alias=True)
await websocket.send_text(event_json)
```
**What gets serialized:**
- Event metadata (id, invocation_id, author, timestamp)
- Content (text, audio data, function calls)
- Event flags (partial, turn_complete, interrupted)
- Transcription data (input_transcription, output_transcription)
- Tool execution information
**When to use `model_dump_json()`:**
- ✅ Streaming events over network (WebSocket, SSE)
- ✅ Logging/persistence to JSON files
- ✅ Debugging and inspection
- ✅ Integration with JSON-based APIs
**When NOT to use it:**
- ❌ In-memory processing (use event objects directly)
- ❌ High-frequency events where serialization overhead matters
- ❌ When you only need a few fields (extract them directly instead)
Performance Warning
Binary audio data in `event.content.parts[].inline_data` will be base64-encoded when serialized to JSON, significantly increasing payload size (~133% overhead). For production applications with audio, send binary data separately using WebSocket binary frames or multipart HTTP. See [Optimization for Audio Transmission](#optimization-for-audio-transmission) for details.
### Serialization options
This allows you to reduce payload sizes by excluding unnecessary fields, improving network performance and client processing speed.
Pydantic's `model_dump_json()` supports several useful parameters:
**Usage:**
```python
# Exclude None values for smaller payloads (with camelCase field names)
event_json = event.model_dump_json(exclude_none=True, by_alias=True)
# Custom exclusions (e.g., skip large binary audio)
event_json = event.model_dump_json(
exclude={'content': {'parts': {'__all__': {'inline_data'}}}},
by_alias=True
)
# Include only specific fields
event_json = event.model_dump_json(
include={'content', 'author', 'turn_complete', 'interrupted'},
by_alias=True
)
# Pretty-printed JSON (for debugging)
event_json = event.model_dump_json(indent=2, by_alias=True)
```
The bidi-demo uses `exclude_none=True` to minimize payload size by omitting fields with None values.
### Deserializing on the Client
This shows how to parse and handle serialized events on the client side, enabling responsive UI updates based on event properties like turn completion and interruptions.
On the client side (JavaScript/TypeScript), parse the JSON back to objects:
Demo implementation: app.js:339-688
```javascript
// Handle incoming messages
websocket.onmessage = function (event) {
// Parse the incoming ADK Event
const adkEvent = JSON.parse(event.data);
// Handle turn complete event
if (adkEvent.turnComplete === true) {
// Remove typing indicator from current message
if (currentBubbleElement) {
const textElement = currentBubbleElement.querySelector(".bubble-text");
const typingIndicator = textElement.querySelector(".typing-indicator");
if (typingIndicator) {
typingIndicator.remove();
}
}
currentMessageId = null;
currentBubbleElement = null;
return;
}
// Handle interrupted event
if (adkEvent.interrupted === true) {
// Stop audio playback if it's playing
if (audioPlayerNode) {
audioPlayerNode.port.postMessage({ command: "endOfAudio" });
}
// Keep the partial message but mark it as interrupted
if (currentBubbleElement) {
const textElement = currentBubbleElement.querySelector(".bubble-text");
// Remove typing indicator
const typingIndicator = textElement.querySelector(".typing-indicator");
if (typingIndicator) {
typingIndicator.remove();
}
// Add interrupted marker
currentBubbleElement.classList.add("interrupted");
}
currentMessageId = null;
currentBubbleElement = null;
return;
}
// Handle content events (text or audio)
if (adkEvent.content && adkEvent.content.parts) {
const parts = adkEvent.content.parts;
for (const part of parts) {
// Handle text
if (part.text) {
// Add a new message bubble for a new turn
if (currentMessageId == null) {
currentMessageId = Math.random().toString(36).substring(7);
currentBubbleElement = createMessageBubble(part.text, false, true);
currentBubbleElement.id = currentMessageId;
messagesDiv.appendChild(currentBubbleElement);
} else {
// Update the existing message bubble with accumulated text
const existingText = currentBubbleElement.querySelector(".bubble-text").textContent;
const cleanText = existingText.replace(/\.\.\.$/, '');
updateMessageBubble(currentBubbleElement, cleanText + part.text, true);
}
scrollToBottom();
}
}
}
};
```
Demo Implementation
See the complete WebSocket message handler in [`app.js:339-688`](https://github.com/google/adk-samples/blob/2f7b82f182659e0990bfb86f6ef400dd82633c07/python/agents/bidi-demo/app/static/js/app.js#L341-L690)
### Optimization for Audio Transmission
Base64-encoded binary audio in JSON significantly increases payload size. For production applications, use a single WebSocket connection with both binary frames (for audio) and text frames (for metadata):
**Usage:**
```python
async for event in runner.run_live(...):
# Check for binary audio
has_audio = (
event.content and
event.content.parts and
any(p.inline_data for p in event.content.parts)
)
if has_audio:
# Send audio via binary WebSocket frame
for part in event.content.parts:
if part.inline_data:
await websocket.send_bytes(part.inline_data.data)
# Send metadata only (much smaller)
metadata_json = event.model_dump_json(
exclude={'content': {'parts': {'__all__': {'inline_data'}}}},
by_alias=True
)
await websocket.send_text(metadata_json)
else:
# Text-only events can be sent as JSON
await websocket.send_text(event.model_dump_json(exclude_none=True, by_alias=True))
```
This approach reduces bandwidth by ~75% for audio-heavy streams while maintaining full event metadata.
## Automatic Tool Execution in run_live()
Source Reference
See automatic tool execution implementation in [`functions.py`](https://github.com/google/adk-python/blob/427a983b18088bdc22272d02714393b0a779ecdf/src/google/adk/flows/llm_flows/functions.py)
One of the most powerful features of ADK's `run_live()` is **automatic tool execution**. Unlike the raw Gemini Live API, which requires you to manually handle tool calls and responses, ADK abstracts this complexity entirely.
### The Challenge with Raw Live API
When using the Gemini Live API directly (without ADK), tool use requires manual orchestration:
1. **Receive** function calls from the model
1. **Execute** the tools yourself
1. **Format** function responses correctly
1. **Send** responses back to the model
This creates significant implementation overhead, especially in streaming contexts where you need to handle multiple concurrent tool calls, manage errors, and coordinate with ongoing audio/text streams.
### How ADK Simplifies Tool Use
With ADK, tool execution becomes declarative. Simply define tools on your Agent:
Demo implementation: agent.py:11-16
```python
import os
from google.adk.agents import Agent
from google.adk.tools import google_search
agent = Agent(
name="google_search_agent",
model=os.getenv("DEMO_AGENT_MODEL", "gemini-2.5-flash-native-audio-preview-12-2025"),
tools=[google_search],
instruction="You are a helpful assistant that can search the web."
)
```
When you call `runner.run_live()`, ADK automatically:
- **Detects** when the model returns function calls in streaming responses
- **Executes** tools in parallel for maximum performance
- **Handles** before/after tool callbacks for custom logic
- **Formats** function responses according to Live API requirements
- **Sends** responses back to the model seamlessly
- **Yields** both function call and response events to your application
### Tool Execution Events
When tools execute, you'll receive events through the `run_live()` async generator:
**Usage:**
```python
async for event in runner.run_live(...):
# Function call event - model requesting tool execution
if event.get_function_calls():
print(f"Model calling: {event.get_function_calls()[0].name}")
# Function response event - tool execution result
if event.get_function_responses():
print(f"Tool result: {event.get_function_responses()[0].response}")
```
You don't need to handle the execution yourself—ADK does it automatically. You just observe the events as they flow through the conversation.
Learn More
The bidi-demo sends all events (including function calls and responses) directly to the WebSocket client without server-side filtering. This allows the client to observe tool execution in real-time through the event stream. See the downstream task in [`main.py:219-234`](https://github.com/google/adk-samples/blob/31847c0723fbf16ddf6eed411eb070d1c76afd1a/python/agents/bidi-demo/app/main.py#L219-L234)
### Long-Running and Streaming Tools
ADK supports advanced tool patterns that integrate seamlessly with `run_live()`:
**Long-Running Tools**: Tools that require human approval or take extended time to complete. Wrap the function in `LongRunningFunctionTool`, which sets the tool's `is_long_running` attribute (a plain function in `tools=[...]` cannot carry it). In resumable async flows, ADK can pause after long-running calls. In live flows, streaming continues; `long_running_tool_ids` indicate pending operations and clients can display appropriate UI.
**Streaming Tools**: Tools that accept an `input_stream` parameter with type `LiveRequestQueue` can send real-time updates back to the model during execution, enabling progressive responses.
How Streaming Tools Work
When the model calls a tool, ADK identifies it as a streaming tool by looking for an `input_stream` parameter annotated with `LiveRequestQueue`.
**Queue creation and lifecycle**:
1. **Creation**: ADK lazily registers an `ActiveStreamingTool` with a dedicated `LiveRequestQueue` the first time the model calls the streaming tool
1. **Storage**: These queues are stored in `invocation_context.active_streaming_tools[tool_name]` for the duration of the invocation
1. **Injection**: ADK automatically injects the tool's queue as the `input_stream` parameter when it invokes the tool
1. **Usage**: The tool can use this queue to send real-time updates back to the model during execution
1. **Lifecycle**: The queues persist for the rest of the `run_live()` invocation (one InvocationContext = one `run_live()` call) and are destroyed when `run_live()` exits. A queue can be cleared earlier by a `stop_streaming(function_name=...)` call: ADK special-cases that function *name*, but ships no such tool—you define and register it yourself (see [Streaming Tools](/streaming/streaming-tools/))
**Queue distinction**:
- **Main queue** (`live_request_queue` parameter): Created by your application, used for client-to-model communication
- **Tool queues** (`active_streaming_tools[tool_name].stream`): Created automatically by ADK, used for tool-to-model communication during execution
Both types of queues are `LiveRequestQueue` instances, but they serve different purposes in the streaming architecture.
This enables tools to provide incremental updates, progress notifications, or partial results during long-running operations.
**Code reference**: See `_process_function_live_helper()` in `flows/llm_flows/functions.py` (tool detection and queue creation) and `FunctionTool._call_live()` in `tools/function_tool.py` (parameter injection) for implementation details.
See the [Tools Guide](/integrations/) for implementation examples.
### Key Takeaway
The difference between raw Live API tool use and ADK is stark:
| Aspect | Raw Live API | ADK `run_live()` |
| ------------------------- | ------------------------------- | ------------------------------- |
| **Tool Declaration** | Manual schema definition | Automatic from Python functions |
| **Tool Execution** | Manual handling in app code | Automatic parallel execution |
| **Response Formatting** | Manual JSON construction | Automatic |
| **Error Handling** | Manual try/catch and formatting | Automatic capture and reporting |
| **Streaming Integration** | Manual coordination | Automatic event yielding |
| **Developer Experience** | Complex, error-prone | Declarative, simple |
This automatic handling is one of the core value propositions of ADK—it transforms the complexity of Live API tool use into a simple, declarative developer experience.
## InvocationContext: The Execution State Container
Source Reference
See InvocationContext implementation in [`invocation_context.py`](https://github.com/google/adk-python/blob/427a983b18088bdc22272d02714393b0a779ecdf/src/google/adk/agents/invocation_context.py)
While `run_live()` returns an AsyncGenerator for consuming events, internally it creates and manages an `InvocationContext`—ADK's unified state carrier that encapsulates everything needed for a complete conversation invocation. **One InvocationContext corresponds to one `run_live()` loop**—it's created when you call `run_live()` and persists for the entire streaming session.
Think of it as a traveling notebook that accompanies a conversation from start to finish, collecting information, tracking progress, and providing context to every component along the way. It's ADK's runtime implementation of the Context concept, providing the execution-time state and services needed during a live conversation. For a broader overview of context in ADK, see [Context in ADK](/context/).
### What is an Invocation?
An **invocation** represents a complete interaction cycle:
- Starts with user input (text, audio, or control signal)
- May involve one or multiple agent calls
- Ends when a final response is generated or when explicitly terminated
- Is orchestrated by `runner.run_live()` or `runner.run_async()`
This is distinct from an **agent call** (execution of a single agent's logic) and a **step** (a single LLM call plus any resulting tool executions).
The hierarchy looks like this:
```text
┌─────────────────────── invocation ──────────────────────────┐
┌──────────── llm_agent_call_1 ────────────┐ ┌─ agent_call_2 ─┐
┌──── step_1 ────────┐ ┌───── step_2 ──────┐
[call_llm] [call_tool] [call_llm] [transfer]
```
### Who Uses InvocationContext?
InvocationContext serves different audiences at different levels:
- **ADK's internal components** (primary users): Runner, Agent, LLMFlow, and GeminiLlmConnection all receive, read from, and write to the InvocationContext as it flows through the stack. This shared context enables seamless coordination without tight coupling.
- **Application developers** (indirect beneficiaries): You don't typically create or manipulate InvocationContext directly in your application code. Instead, you benefit from the clean, simplified APIs that InvocationContext enables behind the scenes—like the elegant `async for event in runner.run_live()` pattern.
- **Tool and callback developers** (indirect access): When you implement custom tools or callbacks, ADK injects a `ToolContext` or `CallbackContext`, not the InvocationContext itself. That context gives you the invocation's conversation state and services, but not its control flags—see the warning below on `end_invocation`.
#### What InvocationContext Contains
When you implement custom tools or callbacks, you receive the invocation state through a `ToolContext` or `CallbackContext` parameter. Here's what's available to you:
**Essential Fields for Tool/Callback Developers:**
- **`tool_context.invocation_id`**: Current invocation identifier (unique per `run_live()` call)
- **`tool_context.session`**:
- **`tool_context.session.events`**: All events in the session history (across all invocations)
- **`tool_context.session.user_id`**: User identity
- **`tool_context.state`**: Delta-aware key-value store for session data; mutate it directly and ADK persists the change
- **`tool_context.run_config`**: Current streaming configuration (response modalities, transcription settings, cost limits)
- **`tool_context.actions`**: Event actions for this call (`skip_summarization`, `transfer_to_agent`, `escalate`, `state_delta`, `artifact_delta`)
Tools Cannot Set end_invocation
The `end_invocation` flag lives on the InvocationContext, and `tool_context.get_invocation_context()` returns a **copy** of it, so setting the flag on the returned object is a silent no-op. To stop early, set `ctx.end_invocation = True` inside the agent's own `_run_live_impl()`, or return content from a `before_agent_callback` and let ADK set the flag for you.
**Example Use Cases in Tool Development:**
```python
# Example: Comprehensive tool implementation showing common context patterns
from datetime import datetime
from google.adk.tools import ToolContext
from google.genai import types
async def my_tool(tool_context: ToolContext, query: str):
# Access user identity
user_id = tool_context.session.user_id
# Check if this is the user's first message
event_count = len(tool_context.session.events)
if event_count == 0:
return "Welcome! This is your first message."
# Access conversation history
recent_events = tool_context.session.events[-5:] # Last 5 events
# Access persistent session state
# Session state persists across invocations (not just this streaming session)
user_preferences = tool_context.state.get('user_preferences', {})
# Update session state (will be persisted)
tool_context.state['last_query_time'] = datetime.now().isoformat()
# Store large files/audio in the artifact service
result_bytes = b"..." # Your generated bytes
await tool_context.save_artifact(
filename="result.bin",
artifact=types.Part(inline_data=types.Blob(mime_type="application/octet-stream", data=result_bytes)),
)
# Process the query with context
result = process_query(query, context=recent_events, preferences=user_preferences)
# Whatever you return goes back to the model, errors included; a tool
# cannot end the invocation itself
return result
```
Understanding InvocationContext is essential for grasping how ADK maintains state, coordinates execution, and enables advanced features like multi-agent workflows and resumability. Even if you never touch it directly, knowing what flows through your application helps you design better agents and debug issues more effectively.
## Best Practices for Multi-Agent Workflows
ADK's bidirectional streaming supports three agent architectures: **single agent** (one agent handles the entire conversation), **multi-agent with sub-agents** (a coordinator agent dynamically routes to specialist agents using `transfer_to_agent`), and **sequential workflow agents** (agents execute in a fixed pipeline using `task_completed`). This section focuses on best practices for sequential workflows, where understanding agent transitions and state sharing is crucial for smooth BIDI communication.
Learn More
For comprehensive coverage of multi-agent patterns, see [Workflow Agents as Orchestrators](/agents/multi-agents/#workflow-agents-as-orchestrators) in the ADK documentation.
When building multi-agent systems with ADK, understanding how agents transition and share state during live streaming is crucial for smooth BIDI communication.
### SequentialAgent with BIDI Streaming
`SequentialAgent` enables workflow pipelines where agents execute one after another. Each agent completes its task before the next one begins. The challenge with live streaming is determining when an agent has finished processing continuous audio or video input.
Source Reference
See SequentialAgent implementation in [`sequential_agent.py:120-159`](https://github.com/google/adk-python/blob/427a983b18088bdc22272d02714393b0a779ecdf/src/google/adk/agents/sequential_agent.py#L120-L159)
**How it works:**
ADK automatically adds a `task_completed()` function to each agent in the sequence. When the model calls this function, it signals completion and triggers the transition to the next agent:
**Usage:**
```python
# SequentialAgent automatically adds this tool to each sub-agent
def task_completed():
"""
Signals that the agent has successfully completed the user's question
or task.
"""
return 'Task completion signaled.'
```
### Recommended Pattern: Transparent Sequential Flow
The key insight is that **agent transitions happen transparently** within the same `run_live()` event stream. Your application doesn't need to manage transitions—just consume events uniformly:
**Usage:**
```python
async def handle_sequential_workflow():
"""Recommended pattern for SequentialAgent with BIDI streaming."""
# 1. Single queue shared across all agents in the sequence
queue = LiveRequestQueue()
# 2. Background task captures user input continuously
async def capture_user_input():
while True:
# Your logic to read audio from microphone
audio_chunk = await microphone.read()
queue.send_realtime(
blob=types.Blob(data=audio_chunk, mime_type="audio/pcm;rate=16000")
)
input_task = asyncio.create_task(capture_user_input())
try:
# 3. Single event loop handles ALL agents seamlessly
async for event in runner.run_live(
user_id="user_123",
session_id="session_456",
live_request_queue=queue,
):
# Events flow seamlessly across agent transitions
current_agent = event.author
# Handle audio and text output
if event.content and event.content.parts:
for part in event.content.parts:
# Check for audio data
if part.inline_data and part.inline_data.mime_type.startswith("audio/"):
# Your logic to play audio
await play_audio(part.inline_data.data)
# Check for text data
if part.text:
await display_text(f"[{current_agent}] {part.text}")
# No special transition handling needed!
finally:
input_task.cancel()
queue.close()
```
### Event Flow During Agent Transitions
Here's what your application sees when agents transition:
```text
# Agent 1 (Researcher) completes its work
Event: author="researcher", text="I've gathered all the data."
Event: author="researcher", function_call: task_completed()
Event: author="researcher", function_response: task_completed
# --- Automatic transition (invisible to your code) ---
# Agent 2 (Writer) begins
Event: author="writer", text="Let me write the report based on the research..."
Event: author="writer", text=" The findings show..."
Event: author="writer", function_call: task_completed()
Event: author="writer", function_response: task_completed
# --- Automatic transition ---
# Agent 3 (Reviewer) begins - the last agent in sequence
Event: author="reviewer", text="Let me review the report..."
Event: author="reviewer", text="The report looks good. All done!"
Event: author="reviewer", function_call: task_completed()
Event: author="reviewer", function_response: task_completed
# --- Last agent completed: run_live() exits ---
# Your async for loop ends here
```
### Design Principles
#### 1. Single Event Loop
Use one event loop for all agents in the sequence:
**Usage:**
```python
# ✅ CORRECT: One loop handles all agents
async for event in runner.run_live(...):
# Your event handling logic here
await handle_event(event) # Works for Agent1, Agent2, Agent3...
# ❌ INCORRECT: Don't break the loop or create multiple loops
for agent in agents:
async for event in runner.run_live(...): # WRONG!
...
```
#### 2. Persistent Queue
The same `LiveRequestQueue` serves all agents:
```text
# User input flows to whichever agent is currently active
User speaks → Queue → Agent1 (researcher)
↓
User speaks → Queue → Agent2 (writer)
↓
User speaks → Queue → Agent3 (reviewer)
```
**Don't create new queues per agent:**
```python
# ❌ INCORRECT: New queue per agent
for agent in agents:
new_queue = LiveRequestQueue() # WRONG!
# ✅ CORRECT: Single queue for entire workflow
queue = LiveRequestQueue()
async for event in runner.run_live(live_request_queue=queue):
...
```
#### 3. Agent-Aware UI (Optional)
Track which agent is active for better user experience:
**Usage:**
```python
current_agent_name = None
async for event in runner.run_live(...):
# Detect agent transitions
if event.author and event.author != current_agent_name:
current_agent_name = event.author
# Your logic to update UI indicator
await update_ui_indicator(f"Now: {current_agent_name}")
# Your event handling logic here
await handle_event(event)
```
#### 4. Transition Notifications
Optionally notify users when agents hand off:
**Usage:**
```python
async for event in runner.run_live(...):
# Detect task completion (transition signal)
if event.content and event.content.parts:
for part in event.content.parts:
if (part.function_response and
part.function_response.name == "task_completed"):
# Your logic to display transition notification
await display_notification(
f"✓ {event.author} completed. Handing off to next agent..."
)
continue
# Your event handling logic here
await handle_event(event)
```
### Key Differences: transfer_to_agent vs task_completed
Understanding these two functions helps you choose the right multi-agent pattern:
| Function | Agent Pattern | When `run_live()` Exits | Use Case |
| ------------------- | ----------------------------- | ---------------------------------------------------------------- | ----------------------------------------- |
| `transfer_to_agent` | Coordinator (dynamic routing) | `LiveRequestQueue.close()` | Route user to specialist based on intent |
| `task_completed` | Sequential (pipeline) | `LiveRequestQueue.close()` or `task_completed` of the last agent | Fixed workflow: research → write → review |
**transfer_to_agent example:**
```text
# Coordinator routes based on user intent
User: "I need help with billing"
Event: author="coordinator", function_call: transfer_to_agent(agent_name="billing")
# Stream continues with billing agent - same run_live() loop
Event: author="billing", text="I can help with your billing question..."
```
**task_completed example:**
```text
# Sequential workflow progresses through pipeline
Event: author="researcher", function_call: task_completed()
# Current agent exits, next agent in sequence begins
Event: author="writer", text="Based on the research..."
```
### Best Practices Summary
| Practice | Reason |
| --------------------------------------- | ----------------------------------------- |
| Use single event loop | ADK handles transitions internally |
| Keep queue alive across agents | Same queue serves all sequential agents |
| Track `event.author` | Know which agent is currently responding |
| Don't reset session/context | Conversation state persists across agents |
| Handle events uniformly | All agents produce the same event types |
| Let `task_completed` signal transitions | Don't manually manage sequential flow |
The SequentialAgent design ensures smooth transitions—your application simply sees a continuous stream of events from different agents in sequence, with automatic handoffs managed by ADK.
## Summary
In this part, you mastered event handling in ADK Gemini Live API Toolkit's streaming architecture. We explored the different event types that agents generate—text responses, audio chunks, transcriptions, tool calls, and control signals—and learned how to process each event type effectively. You now understand how to handle interruptions and turn completion signals for natural conversation flow, serialize events for network transport using Pydantic's model serialization, leverage ADK's automatic tool execution to simplify agent workflows, and access InvocationContext for advanced state management scenarios. With these event handling patterns in place, you're equipped to build responsive streaming applications that provide real-time feedback to users. Next, you'll learn how to configure sophisticated streaming behaviors through RunConfig, including multimodal interactions, session resumption, and cost controls.
______________________________________________________________________
← [Previous: Part 2: Sending Messages with LiveRequestQueue](https://adk.dev/streaming/dev-guide/part2/index.md) | [Next: Part 4: Understanding RunConfig](https://adk.dev/streaming/dev-guide/part4/index.md) →