Skip to content

strands.experimental.bidi.types

Media input types for bidirectional streaming.

@dataclass
class AudioDelta()

Defined in: src/strands/experimental/bidi/types/media.py:15

Audio samples to append to the live input stream.

Sending a delta does not explicitly end the user’s turn.

Attributes:

  • format - Audio format.
  • source - Source containing the audio samples.
def to_dict() -> _AudioDeltaData

Defined in: src/strands/experimental/bidi/types/media.py:28

Return the dictionary form of this delta.

Agent-related type definitions for bidirectional streaming.

This module defines the types used for BidiAgent.

Input accepted by a bidirectional agent.

Bidirectional streaming types for real-time audio/text conversations.

Type definitions for bidirectional streaming that extends Strands’ existing streaming capabilities with real-time audio and persistent connection support.

Key features:

  • Audio output events with standardized formats
  • Interruption detection and handling
  • Connection lifecycle management
  • Provider-agnostic event types
  • Type-safe discriminated unions with TypedEvent
  • JSON-serializable output events (audio stored as base64 strings)

Audio format normalization:

  • Supports PCM, WAV, Opus, and MP3 formats
  • Describes sample rates in Hz
  • Normalizes channel configurations (mono/stereo)
  • Abstracts provider-specific encodings
  • Audio output stored as base64-encoded strings for JSON compatibility

Number of audio channels.

  • Mono: 1
  • Stereo: 2

Audio encoding format.

Role of a message sender.

  • “user”: Messages from the user to the assistant.
  • “assistant”: Messages from the assistant to the user.

Reason for the model ending its response generation.

  • “complete”: Model completed its response.
  • “error”: Model encountered an error.
  • “interrupted”: Model was interrupted by the user.
  • “tool_use”: Model is requesting a tool use.
class BidiConnectionStartEvent(TypedEvent)

Defined in: src/strands/experimental/bidi/types/events.py:94

Streaming connection established and ready for interaction.

Arguments:

  • connection_id - Unique identifier for this streaming connection.
  • model - Model identifier (e.g., “gpt-realtime”, “gemini-2.0-flash-live”).
def __init__(connection_id: str, model: str)

Defined in: src/strands/experimental/bidi/types/events.py:102

Initialize connection start event.

@property
def connection_id() -> str

Defined in: src/strands/experimental/bidi/types/events.py:113

Unique identifier for this streaming connection.

@property
def model() -> str

Defined in: src/strands/experimental/bidi/types/events.py:118

Model identifier (e.g., ‘gpt-realtime’, ‘gemini-2.0-flash-live’).

class BidiConnectionRestartEvent(TypedEvent)

Defined in: src/strands/experimental/bidi/types/events.py:123

Agent is restarting the model connection.

Emitted on both reconnect paths: reactively after the model reports a timeout, and proactively when the reconnect timer fires ahead of the provider’s limit.

Arguments:

  • reason - What triggered the restart (“timeout” reactively, “scheduled” proactively).
  • timeout_error - The model’s timeout error on the reactive path; None when scheduled.
  • turn_interrupted - True if the restart cut an in-progress or owed turn (the alignment wait could not complete it before the deadline, or a timeout struck mid-turn). The provider replays history as context, so that turn will not be answered on its own — an app can re-prompt or notify the user when this is set.
def __init__(reason: Literal["timeout", "scheduled"],
timeout_error: "BidiModelTimeoutError | None" = None,
turn_interrupted: bool = False)

Defined in: src/strands/experimental/bidi/types/events.py:138

Initialize connection restart event.

@property
def reason() -> str

Defined in: src/strands/experimental/bidi/types/events.py:155

What triggered the restart (“timeout” or “scheduled”).

@property
def timeout_error() -> "BidiModelTimeoutError | None"

Defined in: src/strands/experimental/bidi/types/events.py:160

Model timeout error on the reactive path; None when scheduled.

@property
def turn_interrupted() -> bool

Defined in: src/strands/experimental/bidi/types/events.py:165

True if the restart cut an in-progress or owed turn that will not be answered.

class BidiConnectionWarningEvent(TypedEvent)

Defined in: src/strands/experimental/bidi/types/events.py:170

Agent is approaching a proactive reconnect.

Emitted by the proactive reconnect timer before a reconnect; informational only.

Arguments:

  • time_left_s - Approximate seconds until the scheduled reconnect.
def __init__(time_left_s: float)

Defined in: src/strands/experimental/bidi/types/events.py:179

Initialize connection warning event.

@property
def time_left_s() -> float

Defined in: src/strands/experimental/bidi/types/events.py:189

Approximate seconds until the scheduled reconnect.

class BidiResponseStartEvent(TypedEvent)

Defined in: src/strands/experimental/bidi/types/events.py:194

Model starts generating a response.

Arguments:

  • response_id - Unique identifier for this response (used in response.complete).
def __init__(response_id: str)

Defined in: src/strands/experimental/bidi/types/events.py:201

Initialize response start event.

@property
def response_id() -> str

Defined in: src/strands/experimental/bidi/types/events.py:206

Unique identifier for this response.

class BidiAudioStreamEvent(TypedEvent)

Defined in: src/strands/experimental/bidi/types/events.py:211

Streaming audio output from the model.

Arguments:

  • audio - Base64-encoded audio string.
  • format - Audio encoding format.
  • sample_rate - Number of audio samples per second in Hz.
  • channels - Number of audio channels (1=mono, 2=stereo).
def __init__(audio: str, format: AudioFormat, sample_rate: int,
channels: AudioChannel)

Defined in: src/strands/experimental/bidi/types/events.py:221

Initialize audio stream event.

@property
def audio() -> str

Defined in: src/strands/experimental/bidi/types/events.py:240

Base64-encoded audio string.

@property
def format() -> AudioFormat

Defined in: src/strands/experimental/bidi/types/events.py:245

Audio encoding format.

@property
def sample_rate() -> int

Defined in: src/strands/experimental/bidi/types/events.py:250

Number of audio samples per second in Hz.

@property
def channels() -> AudioChannel

Defined in: src/strands/experimental/bidi/types/events.py:255

Number of audio channels (1=mono, 2=stereo).

class BidiTranscriptStreamEvent(TypedEvent)

Defined in: src/strands/experimental/bidi/types/events.py:260

Incremental transcription of user or assistant speech.

Arguments:

  • delta - The incremental transcript text.
  • role - Who is speaking (“user” or “assistant”).
def __init__(delta: str, role: Role)

Defined in: src/strands/experimental/bidi/types/events.py:268

Initialize transcript stream event.

@property
def delta() -> str

Defined in: src/strands/experimental/bidi/types/events.py:279

The incremental transcript text.

@property
def role() -> Role

Defined in: src/strands/experimental/bidi/types/events.py:284

The role of the message sender.

class BidiTranscriptCompleteEvent(TypedEvent)

Defined in: src/strands/experimental/bidi/types/events.py:289

Complete transcript for one user or assistant turn.

Arguments:

  • transcript - The complete transcript text.
  • role - Who spoke (“user” or “assistant”).
def __init__(transcript: str, role: Role)

Defined in: src/strands/experimental/bidi/types/events.py:297

Initialize transcript complete event.

@property
def transcript() -> str

Defined in: src/strands/experimental/bidi/types/events.py:308

The complete transcript text.

@property
def role() -> Role

Defined in: src/strands/experimental/bidi/types/events.py:313

The role of the speaker.

class BidiInterruptionEvent(TypedEvent)

Defined in: src/strands/experimental/bidi/types/events.py:318

Model generation was interrupted.

Arguments:

  • reason - Why the interruption occurred.
def __init__(reason: Literal["user_speech", "error"])

Defined in: src/strands/experimental/bidi/types/events.py:325

Initialize interruption event.

@property
def reason() -> str

Defined in: src/strands/experimental/bidi/types/events.py:335

Why the interruption occurred.

class BidiResponseCompleteEvent(TypedEvent)

Defined in: src/strands/experimental/bidi/types/events.py:340

Model finished generating response.

Arguments:

  • response_id - ID of the response that completed (matches response.start).
  • stop_reason - Why the response ended.
def __init__(response_id: str, stop_reason: StopReason)

Defined in: src/strands/experimental/bidi/types/events.py:348

Initialize response complete event.

@property
def response_id() -> str

Defined in: src/strands/experimental/bidi/types/events.py:363

Unique identifier for this response.

@property
def stop_reason() -> StopReason

Defined in: src/strands/experimental/bidi/types/events.py:368

Why the response ended.

class ModalityUsage(dict)

Defined in: src/strands/experimental/bidi/types/events.py:373

Token usage for a specific modality.

Attributes:

  • modality - Type of content.
  • input_tokens - Tokens used for this modality’s input.
  • output_tokens - Tokens used for this modality’s output.
class BidiUsageEvent(TypedEvent)

Defined in: src/strands/experimental/bidi/types/events.py:387

Token usage event with modality breakdown for bidirectional streaming.

Tracks token consumption across different modalities (audio, text, images) during bidirectional streaming sessions.

Arguments:

  • input_tokens - Total tokens used for all input modalities.
  • output_tokens - Total tokens used for all output modalities.
  • total_tokens - Sum of input and output tokens.
  • modality_details - Optional list of token usage per modality.
  • cache_read_input_tokens - Optional tokens read from cache.
  • cache_write_input_tokens - Optional tokens written to cache.
def __init__(input_tokens: int,
output_tokens: int,
total_tokens: int,
modality_details: list[ModalityUsage] | None = None,
cache_read_input_tokens: int | None = None,
cache_write_input_tokens: int | None = None)

Defined in: src/strands/experimental/bidi/types/events.py:402

Initialize usage event.

@property
def input_tokens() -> int

Defined in: src/strands/experimental/bidi/types/events.py:427

Total tokens used for all input modalities.

@property
def output_tokens() -> int

Defined in: src/strands/experimental/bidi/types/events.py:432

Total tokens used for all output modalities.

@property
def total_tokens() -> int

Defined in: src/strands/experimental/bidi/types/events.py:437

Sum of input and output tokens.

@property
def modality_details() -> list[ModalityUsage]

Defined in: src/strands/experimental/bidi/types/events.py:442

Optional list of token usage per modality.

@property
def cache_read_input_tokens() -> int | None

Defined in: src/strands/experimental/bidi/types/events.py:447

Optional tokens read from cache.

@property
def cache_write_input_tokens() -> int | None

Defined in: src/strands/experimental/bidi/types/events.py:452

Optional tokens written to cache.

class BidiConnectionCloseEvent(TypedEvent)

Defined in: src/strands/experimental/bidi/types/events.py:457

Streaming connection closed.

Arguments:

  • connection_id - Unique identifier for this streaming connection (matches BidiConnectionStartEvent).
  • reason - Why the connection was closed.
def __init__(connection_id: str,
reason: Literal["client_disconnect", "timeout", "error",
"complete", "user_request"])

Defined in: src/strands/experimental/bidi/types/events.py:465

Initialize connection close event.

@property
def connection_id() -> str

Defined in: src/strands/experimental/bidi/types/events.py:480

Unique identifier for this streaming connection.

@property
def reason() -> str

Defined in: src/strands/experimental/bidi/types/events.py:485

Why the interruption occurred.

class BidiErrorEvent(TypedEvent)

Defined in: src/strands/experimental/bidi/types/events.py:490

Error occurred during the session.

Stores the full Exception object as an instance attribute for debugging while keeping the event dict JSON-serializable. The exception can be accessed via the error property for re-raising or type-based error handling.

Arguments:

  • error - The exception that occurred.
  • details - Optional additional error information.
def __init__(error: Exception, details: dict[str, Any] | None = None)

Defined in: src/strands/experimental/bidi/types/events.py:502

Initialize error event.

@property
def error() -> Exception

Defined in: src/strands/experimental/bidi/types/events.py:521

The original exception that occurred.

Can be used for re-raising or type-based error handling.

@property
def code() -> str

Defined in: src/strands/experimental/bidi/types/events.py:529

Error code derived from exception class name.

@property
def message() -> str

Defined in: src/strands/experimental/bidi/types/events.py:534

Human-readable error message from the exception.

@property
def details() -> dict[str, Any] | None

Defined in: src/strands/experimental/bidi/types/events.py:539

Additional error context beyond the exception itself.

Union of different bidi output event types.

Content-related type definitions for bidirectional streaming.

A complete text or image block.

An audio delta for the live input stream.

Dictionary form of one text or image block.

Dictionary form of an audio delta.

Protocol for bidirectional streaming IO channels.

Defines callable protocols for input and output channels that can be used with BidiAgent. This approach provides better typing and flexibility by separating input and output concerns into independent callables.

@runtime_checkable
class BidiInput(Protocol)

Defined in: src/strands/experimental/bidi/types/io.py:19

Protocol for bidirectional input callables.

Input callables read data from a source (microphone, camera, websocket, etc.) and return events to be sent to the agent.

async def start(agent: "BidiAgent") -> None

Defined in: src/strands/experimental/bidi/types/io.py:26

Start input.

async def stop() -> None

Defined in: src/strands/experimental/bidi/types/io.py:30

Stop input.

def __call__() -> Awaitable[BidiAgentInput]

Defined in: src/strands/experimental/bidi/types/io.py:34

Read input data from the source.

Returns:

Awaitable that resolves to input content (audio, text, image, etc.)

@runtime_checkable
class BidiOutput(Protocol)

Defined in: src/strands/experimental/bidi/types/io.py:44

Protocol for bidirectional output callables.

Output callables receive events from the agent and handle them appropriately (play audio, display text, send over websocket, etc.).

async def start(agent: "BidiAgent") -> None

Defined in: src/strands/experimental/bidi/types/io.py:51

Start output.

async def stop() -> None

Defined in: src/strands/experimental/bidi/types/io.py:55

Stop output.

def __call__(event: BidiOutputEvent) -> Awaitable[None]

Defined in: src/strands/experimental/bidi/types/io.py:59

Process output events from the agent.

Arguments:

  • event - Output event from the agent (audio, text, tool calls, etc.)