class OpenAIRealtimeWebSocketModel(RealtimeModel):
"""A model that uses OpenAI's WebSocket API."""
def __init__(self, *, transport_config: TransportConfig | None = None) -> None:
self.model = DEFAULT_REALTIME_MODEL
self._websocket: ClientConnection | None = None
self._websocket_task: asyncio.Task[None] | None = None
self._response_create_tasks: set[asyncio.Task[None]] = set()
self._listeners: list[RealtimeModelListener] = []
self._current_item_id: str | None = None
self._audio_state_tracker: ModelAudioTracker = ModelAudioTracker()
self._response_create_sequencer = _ResponseCreateSequencer()
self._tracing_config: RealtimeModelTracingConfig | Literal["auto"] | None = None
self._playback_tracker: RealtimePlaybackTracker | None = None
self._created_session: OpenAISessionCreateRequest | None = None
self._server_event_type_adapter = get_server_event_type_adapter()
self._call_id: str | None = None
self._transport_config: TransportConfig | None = transport_config
@property
def _ongoing_response(self) -> bool:
return self._response_create_sequencer.ongoing_response
@_ongoing_response.setter
def _ongoing_response(self, value: bool) -> None:
self._response_create_sequencer.set_ongoing_response_for_test(value)
@property
def _response_control(self) -> Literal["free", "create_requested", "cancel_requested"]:
return self._response_create_sequencer.response_control
@property
def _pending_response_create_event_id(self) -> str | None:
return self._response_create_sequencer.pending_response_create_event_id
async def connect(self, options: RealtimeModelConfig) -> None:
"""Establish a connection to the model and keep it alive."""
assert self._websocket is None, "Already connected"
assert self._websocket_task is None, "Already connected"
model_settings: RealtimeSessionModelSettings = options.get("initial_model_settings", {})
self._playback_tracker = options.get("playback_tracker", None)
call_id = options.get("call_id")
model_name = model_settings.get("model_name")
if call_id and model_name:
error_message = (
"Cannot specify both `call_id` and `model_name` "
"when attaching to an existing realtime call."
)
raise UserError(error_message)
if model_name:
self.model = model_name
self._call_id = call_id
api_key = await get_api_key(options.get("api_key"))
if "tracing" in model_settings:
self._tracing_config = model_settings["tracing"]
else:
self._tracing_config = "auto"
if call_id:
url = options.get("url", f"wss://api.openai.com/v1/realtime?call_id={call_id}")
else:
url = options.get("url", f"wss://api.openai.com/v1/realtime?model={self.model}")
headers: dict[str, str] = {}
if options.get("headers") is not None:
# For customizing request headers
headers.update(options["headers"])
else:
# OpenAI's Realtime API
if not api_key:
raise UserError("API key is required but was not provided.")
headers.update({"Authorization": f"Bearer {api_key}"})
self._websocket = await self._create_websocket_connection(
url=url,
headers=headers,
transport_config=self._transport_config,
)
self._websocket_task = asyncio.create_task(self._listen_for_messages())
await self._update_session_config(model_settings)
async def _create_websocket_connection(
self,
url: str,
headers: dict[str, str],
transport_config: TransportConfig | None = None,
) -> ClientConnection:
"""Create a WebSocket connection with the given configuration.
Args:
url: The WebSocket URL to connect to.
headers: HTTP headers to include in the connection request.
transport_config: Optional low-level transport configuration.
Returns:
A connected WebSocket client connection.
"""
connect_kwargs: dict[str, Any] = {
"user_agent_header": _USER_AGENT,
"additional_headers": headers,
"max_size": None, # Allow any size of message
}
if transport_config:
if "ping_interval" in transport_config:
connect_kwargs["ping_interval"] = transport_config["ping_interval"]
if "ping_timeout" in transport_config:
connect_kwargs["ping_timeout"] = transport_config["ping_timeout"]
if "handshake_timeout" in transport_config:
connect_kwargs["open_timeout"] = transport_config["handshake_timeout"]
return await websockets.connect(url, **connect_kwargs)
async def _send_tracing_config(
self, tracing_config: RealtimeModelTracingConfig | Literal["auto"] | None
) -> None:
"""Update tracing configuration via session.update event."""
if tracing_config is not None:
converted_tracing_config = _ConversionHelper.convert_tracing_config(tracing_config)
await self._send_raw_message(
OpenAISessionUpdateEvent(
session=OpenAISessionCreateRequest(
model=self.model,
type="realtime",
tracing=converted_tracing_config,
),
type="session.update",
)
)
def add_listener(self, listener: RealtimeModelListener) -> None:
"""Add a listener to the model."""
if listener not in self._listeners:
self._listeners.append(listener)
def remove_listener(self, listener: RealtimeModelListener) -> None:
"""Remove a listener from the model."""
if listener in self._listeners:
self._listeners.remove(listener)
async def _emit_event(self, event: RealtimeModelEvent) -> None:
"""Emit an event to the listeners."""
# Copy list to avoid modification during iteration
for listener in list(self._listeners):
await listener.on_event(event)
async def _listen_for_messages(self):
assert self._websocket is not None, "Not connected"
try:
async for message in self._websocket:
try:
parsed = json.loads(message)
await self._handle_ws_event(parsed)
except json.JSONDecodeError as e:
await self._emit_event(
RealtimeModelExceptionEvent(
exception=e, context="Failed to parse WebSocket message as JSON"
)
)
except Exception as e:
await self._emit_event(
RealtimeModelExceptionEvent(
exception=e, context="Error handling WebSocket event"
)
)
except websockets.exceptions.ConnectionClosedOK:
# Normal connection closure - no exception event needed
logger.debug("WebSocket connection closed normally")
except websockets.exceptions.ConnectionClosed as e:
await self._emit_event(
RealtimeModelExceptionEvent(
exception=e, context="WebSocket connection closed unexpectedly"
)
)
except Exception as e:
await self._emit_event(
RealtimeModelExceptionEvent(
exception=e, context="WebSocket error in message listener"
)
)
finally:
await self._cancel_response_create_tasks()
await self._release_response_waiters()
async def send_event(self, event: RealtimeModelSendEvent) -> None:
"""Send an event to the model."""
if isinstance(event, RealtimeModelSendRawMessage):
converted = _ConversionHelper.try_convert_raw_message(event)
if converted is not None:
if converted.type == "response.create":
request_version = await self._reserve_response_create_request(manual=True)
self._start_response_create(
request_version,
response_create=converted,
manual=True,
)
else:
await self._send_raw_message(converted)
else:
logger.error(f"Failed to convert raw message: {event}")
elif isinstance(event, RealtimeModelSendUserInput):
await self._send_user_input(event)
elif isinstance(event, RealtimeModelSendAudio):
await self._send_audio(event)
elif isinstance(event, RealtimeModelSendToolOutput):
await self._send_tool_output(event)
elif isinstance(event, RealtimeModelSendInterrupt):
await self._send_interrupt(event)
elif isinstance(event, RealtimeModelSendSessionUpdate):
await self._send_session_update(event)
else:
assert_never(event)
raise ValueError(f"Unknown event type: {type(event)}")
async def _send_raw_message(self, event: OpenAIRealtimeClientEvent) -> None:
"""Send a raw message to the model."""
assert self._websocket is not None, "Not connected"
payload = event.model_dump_json(exclude_unset=True)
await self._websocket.send(payload)
async def _set_response_control(
self, control: Literal["free", "create_requested", "cancel_requested"]
) -> None:
await self._response_create_sequencer.set_response_control(control)
async def _mark_response_created(self) -> None:
await self._response_create_sequencer.mark_response_created()
async def _mark_response_done(self) -> None:
await self._response_create_sequencer.mark_response_done()
async def _release_response_waiters(self) -> None:
# Connection teardown means no response.done will arrive, so local
# response sequencing must be released explicitly.
await self._response_create_sequencer.release_waiters()
async def _reserve_response_create_request(self, *, manual: bool = False) -> int:
return await self._response_create_sequencer.reserve_response_create_request(manual=manual)
async def _clear_pending_response_create(self, event_id: str | None = None) -> bool:
return await self._response_create_sequencer.clear_pending_response_create(event_id)
async def _send_response_create_when_idle(
self,
request_version: int,
*,
response_create: OpenAIResponseCreateEvent | None = None,
manual: bool = False,
) -> None:
pending = await self._response_create_sequencer.wait_for_response_create_slot(
request_version,
manual=manual,
event_id=response_create.event_id if response_create is not None else None,
)
if pending is None:
return
try:
response_create_event = (
response_create.model_copy(update={"event_id": pending.event_id})
if response_create is not None
else OpenAIResponseCreateEvent(type="response.create", event_id=pending.event_id)
)
await self._send_raw_message(response_create_event)
except BaseException:
await self._clear_pending_response_create(pending.event_id)
raise
await self._response_create_sequencer.mark_response_create_sent(pending)
async def _send_response_create_in_background(
self,
request_version: int,
*,
response_create: OpenAIResponseCreateEvent | None = None,
manual: bool = False,
) -> None:
try:
await self._send_response_create_when_idle(
request_version,
response_create=response_create,
manual=manual,
)
except asyncio.CancelledError:
logger.debug("Deferred response.create task was cancelled")
except AssertionError as exc:
if str(exc) != "Not connected":
await self._emit_event(
RealtimeModelExceptionEvent(
exception=exc, context="Error sending deferred response.create"
)
)
except websockets.exceptions.ConnectionClosed:
logger.debug("Skipping deferred response.create because the websocket is closed")
except Exception as exc:
await self._emit_event(
RealtimeModelExceptionEvent(
exception=exc, context="Error sending deferred response.create"
)
)
def _start_response_create(
self,
request_version: int,
*,
response_create: OpenAIResponseCreateEvent | None = None,
manual: bool = False,
) -> None:
task = asyncio.create_task(
self._send_response_create_in_background(
request_version,
response_create=response_create,
manual=manual,
)
)
self._response_create_tasks.add(task)
task.add_done_callback(self._response_create_tasks.discard)
async def _cancel_response_create_tasks(self) -> None:
if not self._response_create_tasks:
return
current_task = asyncio.current_task()
tasks_to_await = []
for task in list(self._response_create_tasks):
task.cancel()
if task is not current_task:
tasks_to_await.append(task)
if tasks_to_await:
await asyncio.gather(*tasks_to_await, return_exceptions=True)
async def _send_user_input(self, event: RealtimeModelSendUserInput) -> None:
converted = _ConversionHelper.convert_user_input_to_item_create(event)
await self._send_raw_message(converted)
request_version = await self._reserve_response_create_request()
self._start_response_create(request_version)
async def _send_audio(self, event: RealtimeModelSendAudio) -> None:
converted = _ConversionHelper.convert_audio_to_input_audio_buffer_append(event)
await self._send_raw_message(converted)
if event.commit:
await self._send_raw_message(
OpenAIInputAudioBufferCommitEvent(type="input_audio_buffer.commit")
)
async def _send_tool_output(self, event: RealtimeModelSendToolOutput) -> None:
converted = _ConversionHelper.convert_tool_output(event)
await self._send_raw_message(converted)
tool_item = RealtimeToolCallItem(
item_id=event.tool_call.id or "",
previous_item_id=event.tool_call.previous_item_id,
call_id=event.tool_call.call_id,
type="function_call",
status="completed",
arguments=event.tool_call.arguments,
name=event.tool_call.name,
output=event.output,
)
await self._emit_event(RealtimeModelItemUpdatedEvent(item=tool_item))
if event.start_response:
request_version = await self._reserve_response_create_request()
self._start_response_create(request_version)
def _get_playback_state(self) -> RealtimePlaybackState:
if self._playback_tracker:
return self._playback_tracker.get_state()
if last_audio_item_id := self._audio_state_tracker.get_last_audio_item():
item_id, item_content_index = last_audio_item_id
audio_state = self._audio_state_tracker.get_state(item_id, item_content_index)
if audio_state:
elapsed_ms = (
datetime.now() - audio_state.initial_received_time
).total_seconds() * 1000
return {
"current_item_id": item_id,
"current_item_content_index": item_content_index,
"elapsed_ms": elapsed_ms,
}
return {
"current_item_id": None,
"current_item_content_index": None,
"elapsed_ms": None,
}
def _get_audio_limits(self, item_id: str, item_content_index: int) -> tuple[float, int] | None:
audio_state = self._audio_state_tracker.get_state(item_id, item_content_index)
if audio_state is None:
return None
max_audio_ms = int(math.ceil(audio_state.audio_length_ms))
return audio_state.audio_length_ms, max_audio_ms
async def _send_interrupt(self, event: RealtimeModelSendInterrupt) -> None:
playback_state = self._get_playback_state()
current_item_id = playback_state.get("current_item_id")
current_item_content_index = playback_state.get("current_item_content_index")
elapsed_ms = playback_state.get("elapsed_ms")
if current_item_id is None or elapsed_ms is None:
logger.debug(
"Skipping interrupt. "
f"Item id: {current_item_id}, "
f"elapsed ms: {elapsed_ms}, "
f"content index: {current_item_content_index}"
)
else:
current_item_content_index = current_item_content_index or 0
if elapsed_ms > 0:
await self._emit_event(
RealtimeModelAudioInterruptedEvent(
item_id=current_item_id,
content_index=current_item_content_index,
)
)
max_audio_ms: int | None = None
audio_limits = self._get_audio_limits(current_item_id, current_item_content_index)
if audio_limits is not None:
_, max_audio_ms = audio_limits
truncated_ms = max(int(elapsed_ms), 0)
if self._ongoing_response or max_audio_ms is None or truncated_ms < max_audio_ms:
converted = _ConversionHelper.convert_interrupt(
current_item_id,
current_item_content_index,
truncated_ms,
)
await self._send_raw_message(converted)
else:
logger.debug(
"Didn't interrupt bc elapsed ms is < 0. "
f"Item id: {current_item_id}, "
f"elapsed ms: {elapsed_ms}, "
f"content index: {current_item_content_index}"
)
session = self._created_session
automatic_response_cancellation_enabled = (
session
and session.audio is not None
and session.audio.input is not None
and session.audio.input.turn_detection is not None
and session.audio.input.turn_detection.interrupt_response is True
)
should_cancel_response = event.force_response_cancel or (
not automatic_response_cancellation_enabled
)
if should_cancel_response:
await self._cancel_response()
if current_item_id is not None and elapsed_ms is not None:
self._audio_state_tracker.on_interrupted()
if self._playback_tracker:
self._playback_tracker.on_interrupted()
async def _send_session_update(self, event: RealtimeModelSendSessionUpdate) -> None:
"""Send a session update to the model."""
await self._update_session_config(event.session_settings)
async def _handle_audio_delta(self, parsed: ResponseAudioDeltaEvent) -> None:
"""Handle audio delta events and update audio tracking state."""
self._current_item_id = parsed.item_id
audio_bytes = base64.b64decode(parsed.delta)
self._audio_state_tracker.on_audio_delta(parsed.item_id, parsed.content_index, audio_bytes)
await self._emit_event(
RealtimeModelAudioEvent(
data=audio_bytes,
response_id=parsed.response_id,
item_id=parsed.item_id,
content_index=parsed.content_index,
)
)
async def _handle_output_item(self, item: ConversationItem) -> None:
"""Handle response output item events (function calls and messages)."""
if item.type == "function_call" and item.status == "completed":
tool_call = RealtimeToolCallItem(
item_id=item.id or "",
previous_item_id=None,
call_id=item.call_id,
type="function_call",
# We use the same item for tool call and output, so it will be completed by the
# output being added
status="in_progress",
arguments=item.arguments or "",
name=item.name or "",
output=None,
)
await self._emit_event(RealtimeModelItemUpdatedEvent(item=tool_call))
await self._emit_event(
RealtimeModelToolCallEvent(
call_id=item.call_id or "",
name=item.name or "",
arguments=item.arguments or "",
id=item.id or "",
)
)
elif item.type == "message":
# Handle message items from output_item events (no previous_item_id)
message_item: RealtimeMessageItem = TypeAdapter(RealtimeMessageItem).validate_python(
{
"item_id": item.id or "",
"type": item.type,
"role": item.role,
"content": (
[content.model_dump() for content in item.content] if item.content else []
),
"status": "in_progress",
}
)
await self._emit_event(RealtimeModelItemUpdatedEvent(item=message_item))
async def _handle_conversation_item(
self, item: ConversationItem, previous_item_id: str | None
) -> None:
"""Handle conversation item creation/retrieval events."""
message_item = _ConversionHelper.conversation_item_to_realtime_message_item(
item, previous_item_id
)
await self._emit_event(RealtimeModelItemUpdatedEvent(item=message_item))
async def close(self) -> None:
"""Close the session."""
await self._cancel_response_create_tasks()
if self._websocket:
await self._websocket.close()
self._websocket = None
if self._websocket_task:
self._websocket_task.cancel()
try:
await self._websocket_task
except asyncio.CancelledError:
pass
self._websocket_task = None
else:
await self._release_response_waiters()
async def _cancel_response(self) -> None:
if not await self._response_create_sequencer.begin_cancel_response():
return
try:
await self._send_raw_message(OpenAIResponseCancelEvent(type="response.cancel"))
except Exception:
await self._set_response_control("free")
raise
def _error_matches_pending_response_create(self, error: Any) -> bool:
if error.event_id is not None:
return True
code = getattr(error, "code", None)
message = (getattr(error, "message", None) or "").lower()
return code == "bad_response_create" or "response.create" in message
async def _handle_ws_event(self, event: dict[str, Any]):
await self._emit_event(RealtimeModelRawServerEvent(data=event))
# The public interface definedo on this Agents SDK side (e.g., RealtimeMessageItem)
# must be the same even after the GA migration, so this part does the conversion
if isinstance(event, dict) and event.get("type") in (
"response.output_item.added",
"response.output_item.done",
):
item = event.get("item")
if isinstance(item, dict) and item.get("type") == "message":
raw_content = item.get("content") or []
converted_content: list[dict[str, Any]] = []
for part in raw_content:
if not isinstance(part, dict):
continue
if part.get("type") == "audio":
converted_content.append(
{
"type": "audio",
"audio": part.get("audio"),
"transcript": part.get("transcript"),
}
)
elif part.get("type") in ("text", "output_text"):
converted_content.append({"type": "text", "text": part.get("text")})
status = item.get("status")
if status not in ("in_progress", "completed", "incomplete"):
is_done = event.get("type") == "response.output_item.done"
status = "completed" if is_done else "in_progress"
# Explicitly type the adapter for mypy
type_adapter: TypeAdapter[RealtimeMessageItem] = TypeAdapter(RealtimeMessageItem)
message_item: RealtimeMessageItem = type_adapter.validate_python(
{
"item_id": item.get("id", ""),
"type": "message",
"role": item.get("role", "assistant"),
"content": converted_content,
"status": status,
}
)
await self._emit_event(RealtimeModelItemUpdatedEvent(item=message_item))
return
try:
if "previous_item_id" in event and event["previous_item_id"] is None:
event["previous_item_id"] = "" # TODO (rm) remove
parsed: AllRealtimeServerEvents = self._server_event_type_adapter.validate_python(event)
except pydantic.ValidationError as e:
logger.error(f"Failed to validate server event: {event}", exc_info=True)
await self._emit_event(RealtimeModelErrorEvent(error=e))
return
except Exception as e:
event_type = event.get("type", "unknown") if isinstance(event, dict) else "unknown"
logger.error(f"Failed to validate server event: {event}", exc_info=True)
exception_event = RealtimeModelExceptionEvent(
exception=e,
context=f"Failed to validate server event: {event_type}",
)
await self._emit_event(exception_event)
return
if parsed.type == "response.output_audio.delta":
await self._handle_audio_delta(parsed)
elif parsed.type == "response.output_audio.done":
audio_done_event = RealtimeModelAudioDoneEvent(
item_id=parsed.item_id,
content_index=parsed.content_index,
)
await self._emit_event(audio_done_event)
elif parsed.type == "input_audio_buffer.speech_started":
# On VAD speech start, immediately stop local playback so the user can
# barge‑in without overlapping assistant audio.
last_audio = self._audio_state_tracker.get_last_audio_item()
if last_audio is not None:
item_id, content_index = last_audio
playback_state = self._get_playback_state()
playback_item_id = playback_state.get("current_item_id")
playback_content_index = playback_state.get("current_item_content_index") or 0
playback_elapsed_ms = playback_state.get("elapsed_ms")
await self._emit_event(
RealtimeModelAudioInterruptedEvent(item_id=item_id, content_index=content_index)
)
elapsed_override = getattr(parsed, "audio_end_ms", None)
if elapsed_override is None or elapsed_override <= 0:
effective_elapsed_ms = playback_elapsed_ms
else:
effective_elapsed_ms = float(elapsed_override)
if playback_item_id and effective_elapsed_ms is not None:
max_audio_ms: int | None = None
audio_limits = self._get_audio_limits(playback_item_id, playback_content_index)
if audio_limits is not None:
_, max_audio_ms = audio_limits
truncated_ms = max(int(round(effective_elapsed_ms)), 0)
if (
max_audio_ms is not None
and truncated_ms >= max_audio_ms
and not self._ongoing_response
):
logger.debug(
"Skipping truncate because playback appears complete. "
f"Item id: {playback_item_id}, "
f"elapsed ms: {effective_elapsed_ms}, "
f"content index: {playback_content_index}, "
f"audio length ms: {max_audio_ms}"
)
else:
if max_audio_ms is not None:
truncated_ms = min(truncated_ms, max_audio_ms)
await self._send_raw_message(
_ConversionHelper.convert_interrupt(
playback_item_id,
playback_content_index,
truncated_ms,
)
)
# Reset trackers so subsequent playback state queries don't
# reference audio that has been interrupted client‑side.
self._audio_state_tracker.on_interrupted()
if self._playback_tracker:
self._playback_tracker.on_interrupted()
# If server isn't configured to auto‑interrupt/cancel, cancel the
# response to prevent further audio.
session = self._created_session
automatic_response_cancellation_enabled = (
session
and session.audio is not None
and session.audio.input is not None
and session.audio.input.turn_detection is not None
and session.audio.input.turn_detection.interrupt_response is True
)
if not automatic_response_cancellation_enabled:
await self._cancel_response()
elif parsed.type == "response.created":
await self._mark_response_created()
await self._emit_event(RealtimeModelTurnStartedEvent())
elif parsed.type == "response.done":
await self._mark_response_done()
await self._emit_event(RealtimeModelTurnEndedEvent())
elif parsed.type == "session.created":
await self._send_tracing_config(self._tracing_config)
self._update_created_session(parsed.session)
elif parsed.type == "session.updated":
self._update_created_session(parsed.session)
elif parsed.type == "error":
if (
not self._ongoing_response
and self._response_control == "create_requested"
and self._error_matches_pending_response_create(parsed.error)
):
await self._clear_pending_response_create(parsed.error.event_id)
await self._emit_event(RealtimeModelErrorEvent(error=parsed.error))
elif parsed.type == "conversation.item.deleted":
await self._emit_event(RealtimeModelItemDeletedEvent(item_id=parsed.item_id))
elif (
parsed.type == "conversation.item.added"
or parsed.type == "conversation.item.created"
or parsed.type == "conversation.item.retrieved"
):
previous_item_id = (
parsed.previous_item_id if parsed.type == "conversation.item.created" else None
)
if parsed.item.type == "message":
await self._handle_conversation_item(parsed.item, previous_item_id)
elif (
parsed.type == "conversation.item.input_audio_transcription.completed"
or parsed.type == "conversation.item.truncated"
):
if self._current_item_id:
await self._send_raw_message(
OpenAIConversationItemRetrieveEvent(
type="conversation.item.retrieve",
item_id=self._current_item_id,
)
)
if parsed.type == "conversation.item.input_audio_transcription.completed":
await self._emit_event(
RealtimeModelInputAudioTranscriptionCompletedEvent(
item_id=parsed.item_id, transcript=parsed.transcript
)
)
elif parsed.type == "response.output_audio_transcript.delta":
await self._emit_event(
RealtimeModelTranscriptDeltaEvent(
item_id=parsed.item_id, delta=parsed.delta, response_id=parsed.response_id
)
)
elif (
parsed.type == "conversation.item.input_audio_transcription.delta"
or parsed.type == "response.output_text.delta"
or parsed.type == "response.function_call_arguments.delta"
):
# No support for partials yet
pass
elif (
parsed.type == "response.output_item.added"
or parsed.type == "response.output_item.done"
):
await self._handle_output_item(parsed.item)
elif parsed.type == "input_audio_buffer.timeout_triggered":
await self._emit_event(
RealtimeModelInputAudioTimeoutTriggeredEvent(
item_id=parsed.item_id,
audio_start_ms=parsed.audio_start_ms,
audio_end_ms=parsed.audio_end_ms,
)
)
def _update_created_session(
self,
session: OpenAISessionCreateRequest
| OpenAIRealtimeTranscriptionSessionCreateRequest
| Mapping[str, object]
| pydantic.BaseModel,
) -> None:
# Only store/playback-format information for realtime sessions (not transcription-only)
normalized_session = self._normalize_session_payload(session)
if not normalized_session:
return
self._created_session = normalized_session
normalized_format = self._extract_audio_format(normalized_session)
if normalized_format is None:
return
self._audio_state_tracker.set_audio_format(normalized_format)
if self._playback_tracker:
self._playback_tracker.set_audio_format(normalized_format)
@staticmethod
def _normalize_session_payload(
session: OpenAISessionCreateRequest
| OpenAIRealtimeTranscriptionSessionCreateRequest
| Mapping[str, object]
| pydantic.BaseModel,
) -> OpenAISessionCreateRequest | None:
if isinstance(session, OpenAISessionCreateRequest):
return session
if isinstance(session, OpenAIRealtimeTranscriptionSessionCreateRequest):
return None
session_payload: Mapping[str, object]
if isinstance(session, pydantic.BaseModel):
session_payload = cast(Mapping[str, object], session.model_dump())
elif isinstance(session, Mapping):
session_payload = session
else:
return None
if OpenAIRealtimeWebSocketModel._is_transcription_session(session_payload):
return None
try:
return OpenAISessionCreateRequest.model_validate(session_payload)
except pydantic.ValidationError:
return None
@staticmethod
def _is_transcription_session(payload: Mapping[str, object]) -> bool:
try:
OpenAIRealtimeTranscriptionSessionCreateRequest.model_validate(payload)
except pydantic.ValidationError:
return False
else:
return True
@staticmethod
def _extract_audio_format(session: OpenAISessionCreateRequest) -> str | None:
audio = session.audio
if not audio or not audio.output or not audio.output.format:
return None
return OpenAIRealtimeWebSocketModel._normalize_audio_format(audio.output.format)
@staticmethod
def _normalize_audio_format(fmt: object) -> str:
if isinstance(fmt, AudioPCM):
return "pcm16"
if isinstance(fmt, AudioPCMU):
return "g711_ulaw"
if isinstance(fmt, AudioPCMA):
return "g711_alaw"
fmt_type = OpenAIRealtimeWebSocketModel._read_format_type(fmt)
if isinstance(fmt_type, str) and fmt_type:
return fmt_type
return str(fmt)
@staticmethod
def _read_format_type(fmt: object) -> str | None:
if isinstance(fmt, str):
return fmt
if isinstance(fmt, Mapping):
type_value = fmt.get("type")
return type_value if isinstance(type_value, str) else None
if isinstance(fmt, pydantic.BaseModel):
type_value = fmt.model_dump().get("type")
return type_value if isinstance(type_value, str) else None
try:
type_value = fmt.type # type: ignore[attr-defined]
except AttributeError:
return None
return type_value if isinstance(type_value, str) else None
@staticmethod
def _normalize_turn_detection_config(config: object) -> object:
"""Normalize camelCase turn detection keys to snake_case for API compatibility."""
if not isinstance(config, Mapping):
return config
normalized = dict(config)
key_map = {
"createResponse": "create_response",
"interruptResponse": "interrupt_response",
"prefixPaddingMs": "prefix_padding_ms",
"silenceDurationMs": "silence_duration_ms",
"idleTimeoutMs": "idle_timeout_ms",
"modelVersion": "model_version",
}
for camel_key, snake_key in key_map.items():
if camel_key in normalized and snake_key not in normalized:
normalized[snake_key] = normalized[camel_key]
normalized.pop(camel_key, None)
return normalized
async def _update_session_config(self, model_settings: RealtimeSessionModelSettings) -> None:
session_config = self._get_session_config(model_settings)
await self._send_raw_message(
OpenAISessionUpdateEvent(session=session_config, type="session.update")
)
def _get_session_config(
self, model_settings: RealtimeSessionModelSettings
) -> OpenAISessionCreateRequest:
"""Get the session config."""
audio_input_args: dict[str, Any] = {}
audio_output_args: dict[str, Any] = {}
audio_config = model_settings.get("audio")
audio_config_mapping = audio_config if isinstance(audio_config, Mapping) else None
input_audio_config: Mapping[str, Any] = (
cast(Mapping[str, Any], audio_config_mapping.get("input", {}))
if audio_config_mapping
else {}
)
output_audio_config: Mapping[str, Any] = (
cast(Mapping[str, Any], audio_config_mapping.get("output", {}))
if audio_config_mapping
else {}
)
input_format_source: FormatInput = (
input_audio_config.get("format") if input_audio_config else None
)
if input_format_source is None:
if self._call_id:
input_format_source = model_settings.get("input_audio_format")
else:
input_format_source = model_settings.get(
"input_audio_format", DEFAULT_MODEL_SETTINGS.get("input_audio_format")
)
input_format = to_realtime_audio_format(input_format_source)
if input_format is not None:
audio_input_args["format"] = input_format
if "noise_reduction" in input_audio_config:
audio_input_args["noise_reduction"] = input_audio_config.get("noise_reduction")
elif "input_audio_noise_reduction" in model_settings:
audio_input_args["noise_reduction"] = model_settings.get("input_audio_noise_reduction")
if "transcription" in input_audio_config:
audio_input_args["transcription"] = input_audio_config.get("transcription")
elif "input_audio_transcription" in model_settings:
audio_input_args["transcription"] = model_settings.get("input_audio_transcription")
else:
audio_input_args["transcription"] = DEFAULT_MODEL_SETTINGS.get(
"input_audio_transcription"
)
if "turn_detection" in input_audio_config:
audio_input_args["turn_detection"] = self._normalize_turn_detection_config(
input_audio_config.get("turn_detection")
)
elif "turn_detection" in model_settings:
audio_input_args["turn_detection"] = self._normalize_turn_detection_config(
model_settings.get("turn_detection")
)
else:
audio_input_args["turn_detection"] = DEFAULT_MODEL_SETTINGS.get("turn_detection")
requested_voice = output_audio_config.get("voice") if output_audio_config else None
audio_output_args["voice"] = requested_voice or model_settings.get(
"voice", DEFAULT_MODEL_SETTINGS.get("voice")
)
output_format_source: FormatInput = (
output_audio_config.get("format") if output_audio_config else None
)
if output_format_source is None:
if self._call_id:
output_format_source = model_settings.get("output_audio_format")
else:
output_format_source = model_settings.get(
"output_audio_format", DEFAULT_MODEL_SETTINGS.get("output_audio_format")
)
output_format = to_realtime_audio_format(output_format_source)
if output_format is not None:
audio_output_args["format"] = output_format
if "speed" in output_audio_config:
audio_output_args["speed"] = output_audio_config.get("speed")
elif "speed" in model_settings:
audio_output_args["speed"] = model_settings.get("speed")
output_modalities = (
model_settings.get("output_modalities")
or model_settings.get("modalities")
or DEFAULT_MODEL_SETTINGS.get("modalities")
)
# Construct full session object. `type` will be excluded at serialization time for updates.
session_create_request = OpenAISessionCreateRequest(
type="realtime",
model=(model_settings.get("model_name") or self.model) or DEFAULT_REALTIME_MODEL,
output_modalities=output_modalities,
audio=OpenAIRealtimeAudioConfig(
input=OpenAIRealtimeAudioInput(**audio_input_args),
output=OpenAIRealtimeAudioOutput(**audio_output_args),
),
tools=cast(
Any,
self._tools_to_session_tools(
tools=model_settings.get("tools", []),
handoffs=model_settings.get("handoffs", []),
),
),
)
if "instructions" in model_settings:
session_create_request.instructions = model_settings.get("instructions")
if "prompt" in model_settings:
_passed_prompt: Prompt = model_settings["prompt"]
variables: dict[str, Any] | None = _passed_prompt.get("variables")
session_create_request.prompt = ResponsePrompt(
id=_passed_prompt["id"],
variables=variables,
version=_passed_prompt.get("version"),
)
if "max_output_tokens" in model_settings:
session_create_request.max_output_tokens = cast(
Any, model_settings.get("max_output_tokens")
)
if "tool_choice" in model_settings:
tool_choice = model_settings.get("tool_choice")
ensure_tool_choice_supports_backend(
tool_choice,
backend_name="OpenAI Responses models",
)
session_create_request.tool_choice = cast(Any, tool_choice)
return session_create_request
def _tools_to_session_tools(
self, tools: list[Tool], handoffs: list[Handoff]
) -> list[OpenAISessionFunction]:
converted_tools: list[OpenAISessionFunction] = []
for tool in tools:
if not isinstance(tool, FunctionTool):
raise UserError(f"Tool {tool.name} is unsupported. Must be a function tool.")
ensure_function_tool_supports_responses_only_features(
tool,
backend_name="Realtime models",
)
converted_tools.append(
OpenAISessionFunction(
name=tool.name,
description=tool.description,
parameters=tool.params_json_schema,
type="function",
)
)
for handoff in handoffs:
converted_tools.append(
OpenAISessionFunction(
name=handoff.tool_name,
description=handoff.tool_description,
parameters=handoff.input_json_schema,
type="function",
)
)
return converted_tools