agentchat.realtime_agent.websocket_audio_adapter

WebSocketAudioAdapter

class WebSocketAudioAdapter(RealtimeObserver)

__init__

def __init__(websocket: "WebSocket",
             *,
             logger: Optional[Logger] = None) -> None

Observer for handling function calls from the OpenAI Realtime API.

Arguments:

• websocket WebSocket - The websocket connection.
• logger Logger - The logger for the observer.

on_event

async def on_event(event: dict[str, Any]) -> None

Receive events from the OpenAI Realtime API, send audio back to websocket.

handle_speech_started_event

async def handle_speech_started_event() -> None

Handle interruption when the caller’s speech starts.

initialize_session

async def initialize_session() -> None

Control initial session with OpenAI.

run_loop

async def run_loop() -> None

Reads data from websocket and sends it to the RealtimeClient.