strands.experimental.bidi.models.openai_realtime

OpenAI Realtime API provider for Strands bidirectional streaming.

Provides real-time audio and text communication through OpenAI’s Realtime API with WebSocket connections, voice activity detection, and function calling.

OPENAI_MAX_TIMEOUT_S

Max timeout before closing connection.

OpenAI documents a 60 minute limit on realtime sessions (docs). However, OpenAI does not emit any warnings when approaching the limit. As a workaround, we configure a max timeout client side to gracefully handle the connection closure. We set the max to 50 minutes to provide enough buffer before hitting the real limit.

BidiOpenAIRealtimeModel

class BidiOpenAIRealtimeModel(BidiModel)

Defined in: src/strands/experimental/bidi/models/openai_realtime.py:79

OpenAI Realtime API implementation for bidirectional streaming.

Combines model configuration and connection state in a single class. Manages WebSocket connection to OpenAI’s Realtime API with automatic VAD, function calling, and event conversion to Strands format.

__init__

def __init__(model_id: str = DEFAULT_MODEL,
             provider_config: dict[str, Any] | None = None,
             client_config: dict[str, Any] | None = None,
             **kwargs: Any) -> None

Defined in: src/strands/experimental/bidi/models/openai_realtime.py:90

Initialize OpenAI Realtime bidirectional model.

Arguments:

• model_id - Model identifier (default: gpt-realtime)
• provider_config - Model behavior (audio, instructions, turn_detection, etc.)
• client_config - Authentication (api_key, organization, project) Falls back to OPENAI_API_KEY, OPENAI_ORGANIZATION, OPENAI_PROJECT env vars
• **kwargs - Reserved for future parameters.

start

async def start(system_prompt: str | None = None,
                tools: list[ToolSpec] | None = None,
                messages: Messages | None = None,
                **kwargs: Any) -> None

Defined in: src/strands/experimental/bidi/models/openai_realtime.py:180

Establish bidirectional connection to OpenAI Realtime API.

Arguments:

• system_prompt - System instructions for the model.
• tools - List of tools available to the model.
• messages - Conversation history to initialize with.
• **kwargs - Additional configuration options.

receive

async def receive() -> AsyncGenerator[BidiOutputEvent, None]

Defined in: src/strands/experimental/bidi/models/openai_realtime.py:423

Receive OpenAI events and convert to Strands TypedEvent format.

send

async def send(content: BidiInputEvent | ToolResultEvent) -> None

Defined in: src/strands/experimental/bidi/models/openai_realtime.py:703

Unified send method for all content types. Sends the given content to OpenAI.

Dispatches to appropriate internal handler based on content type.

Arguments:

• content - Typed event (BidiTextInputEvent, BidiAudioInputEvent, BidiImageInputEvent, or ToolResultEvent).

Raises:

• ValueError - If content type not supported (e.g., image content).

stop

async def stop() -> None

Defined in: src/strands/experimental/bidi/models/openai_realtime.py:772

Close session and cleanup resources.