#
# Copyright (c) 2024-2026, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Event and session models for the OpenAI Live API (WebSocket transport).
The Live API is a full-duplex speech-to-speech API: the model listens and
speaks at the same time and decides on its own when to talk. Clients start a
session, stream audio in, receive audio and transcript events out, and answer
the model's *delegations* (units of work it hands to a backend model).
Server events are parsed leniently: unrecognised fields are kept, and event
types this module doesn't model become :class:`UnknownServerEvent` rather than
errors, so an addition to the API doesn't break the receive loop.
"""
import json
import uuid
from typing import Any, Literal
from pydantic import BaseModel, ConfigDict, Field
#: Maximum number of startup ``input`` messages a session accepts.
MAX_INPUT_ITEMS = 128
#
# session configuration
#
[docs]
class InputTextContent(BaseModel):
"""A text content part supplied by the client.
Parameters:
type: Content type, always "input_text".
text: The text.
"""
type: Literal["input_text"] = "input_text"
text: str
[docs]
class OutputTextContent(BaseModel):
"""A text content part attributed to the assistant.
Parameters:
type: Content type, always "output_text".
text: The text.
"""
type: Literal["output_text"] = "output_text"
text: str
[docs]
class AudioOutputConfig(BaseModel):
"""Output audio configuration.
Parameters:
voice: Output voice name (defaults to ``marin`` server-side).
"""
voice: str | None = None
[docs]
class AudioConfig(BaseModel):
"""Session audio configuration.
Parameters:
output: Output audio configuration.
format: WebSocket audio format, shared by input and output.
"""
output: AudioOutputConfig | None = None
format: AudioFormat | None = None
[docs]
class ClientDelegationConfig(BaseModel):
"""Delegation configuration selecting client delegation.
Parameters:
type: Delegation type, always "client".
"""
type: Literal["client"] = "client"
[docs]
class ResponsesDelegationConfig(BaseModel):
"""Delegation configuration selecting Responses delegation.
Parameters:
type: Delegation type, always "responses".
responses: Configuration of the backend Responses model: ``model``,
``instructions``, ``tools``, ``tool_choice``, ``reasoning``,
``max_output_tokens``, ``service_tier``, and other Responses
request fields the Live API accepts.
"""
type: Literal["responses"] = "responses"
responses: dict[str, Any]
[docs]
class SessionConfig(BaseModel):
"""Live session configuration, sent once with ``session.start``.
Parameters:
model: The live model. Required, and immutable after start.
instructions: Instructions for the live model. Immutable after start;
:class:`SessionInstructionsAppendEvent` adds more later.
audio: Audio configuration. The output voice and format are immutable
after start.
delegation: Delegation mode. Omitted selects client delegation. The
mode is immutable: switching needs a new session.
input: Prior text-only conversation messages. Startup-only.
"""
model: str
instructions: str | None = None
audio: AudioConfig | None = None
delegation: ClientDelegationConfig | ResponsesDelegationConfig | None = None
input: list[InputItem] | None = None
[docs]
class SessionUpdateConfig(BaseModel):
"""The sparse session update a running session accepts.
Only delegation settings can change, and only within the mode chosen at
startup. Startup fields (``model``, ``instructions``, ``audio``, ``input``)
are not update fields.
Parameters:
delegation: Replacement delegation settings, of the session's mode.
"""
delegation: ClientDelegationConfig | ResponsesDelegationConfig | None = None
#
# client events
#
[docs]
class ClientEvent(BaseModel):
"""Base class for events sent to the Live API.
Parameters:
event_id: Client-chosen identifier, echoed back as ``client_event_id``
on acknowledgments and errors.
"""
event_id: str = Field(default_factory=lambda: str(uuid.uuid4()))
[docs]
def to_payload(self) -> dict[str, Any]:
"""Render the event as the JSON payload to send.
Unset optional fields are omitted: the session schema is strict about
unknown fields, and a ``null`` is not the same as an absent field.
Returns:
The payload.
"""
return self.model_dump(exclude_none=True)
[docs]
class SessionStartEvent(ClientEvent):
"""Start the session. The first message on a WebSocket, sent exactly once.
Parameters:
type: Event type, always "session.start".
session: The session configuration, including the model.
"""
type: Literal["session.start"] = "session.start"
session: SessionConfig
[docs]
class SessionUpdateEvent(ClientEvent):
"""Apply a sparse update to a running session.
Parameters:
type: Event type, always "session.update".
session: The sparse update.
"""
type: Literal["session.update"] = "session.update"
session: SessionUpdateConfig
[docs]
class ContextAppendEvent(ClientEvent):
"""Base class for the three context-append events.
Parameters:
delegation_id: The client delegation this content belongs to, or
``None`` for general session context. Always sent, including when
it is ``None``.
content: The text, at most 500 tokens.
"""
delegation_id: str | None
content: str
[docs]
def to_payload(self) -> dict[str, Any]:
"""Render the event, keeping ``delegation_id`` even when it is ``None``.
The field is required on these events, and ``None`` is meaningful: it
marks the content as general session context rather than belonging to
a delegation.
Returns:
The payload.
"""
return {**super().to_payload(), "delegation_id": self.delegation_id}
[docs]
class SessionInstructionsAppendEvent(ContextAppendEvent):
"""Append instructions the live model will follow, without replacing the startup ones.
Parameters:
type: Event type, always "session.instructions.append".
"""
type: Literal["session.instructions.append"] = "session.instructions.append"
[docs]
class SessionThinkingAppendEvent(ContextAppendEvent):
"""Append information to the live model's internal reasoning.
The content is not spoken when appended, though the model can draw on it
when a later user request makes it relevant. It is not a secrecy boundary.
Parameters:
type: Event type, always "session.thinking.append".
"""
type: Literal["session.thinking.append"] = "session.thinking.append"
[docs]
class FunctionCallOutputItem(BaseModel):
"""The result of a client-actionable function call from a Responses delegation.
Parameters:
type: Item type, always "function_call_output".
call_id: The ``call_id`` of the completed ``function_call`` item.
output: The function output, as a string.
"""
type: Literal["function_call_output"] = "function_call_output"
call_id: str
output: str
[docs]
class ResponseItemCreateEvent(ClientEvent):
"""Queue an input item for the backend Responses model. Not acknowledged.
Queueing an item does not start inference: :class:`ResponseCreateEvent`
does. Responses delegation only.
Parameters:
type: Event type, always "response.item.create".
item: The input item, such as a function call output.
"""
type: Literal["response.item.create"] = "response.item.create"
item: FunctionCallOutputItem | dict[str, Any]
[docs]
class ResponseCreateEvent(ClientEvent):
"""Start or continue delegated Responses work. Responses delegation only.
Every output required by the pending response must be queued first, or
the command is rejected with ``function_call_outputs_required``.
Parameters:
type: Event type, always "response.create".
"""
type: Literal["response.create"] = "response.create"
[docs]
class SessionCloseEvent(ClientEvent):
"""Request a graceful shutdown; the server answers with ``session.closed``.
Parameters:
type: Event type, always "session.close".
"""
type: Literal["session.close"] = "session.close"
#
# server events
#
[docs]
class ServerEvent(BaseModel):
"""Base class for events received from the Live API.
Unknown fields are retained so events keep parsing as the API grows.
Parameters:
type: The event type.
event_id: Identifier of this server event.
client_event_id: The client event this one answers, when correlatable.
"""
model_config = ConfigDict(extra="allow")
type: str
event_id: str | None = None
client_event_id: str | None = None
[docs]
class SessionResource(BaseModel):
"""The resolved session, echoed by ``session.started``, ``session.updated`` and ``session.closed``.
Parameters:
id: Session identifier. Opaque: forward it unchanged.
expires_at: Session expiry as a Unix timestamp in seconds.
status: Session status.
model: The live model.
instructions: The startup instructions.
audio: Audio configuration.
delegation: Delegation configuration.
input: The startup conversation history.
context_management: Long-session context handling.
"""
model_config = ConfigDict(extra="allow")
id: str | None = None
expires_at: int | None = None
status: str | None = None
model: str | None = None
instructions: str | None = None
audio: dict[str, Any] | None = None
delegation: dict[str, Any] | None = None
input: list[dict[str, Any]] | None = None
context_management: dict[str, Any] | None = None
[docs]
class SessionStartedEvent(ServerEvent):
"""The session is configured and ready for audio.
Parameters:
type: Event type, always "session.started".
session: The resolved session.
"""
type: Literal["session.started"]
session: SessionResource
[docs]
class SessionUpdatedEvent(ServerEvent):
"""A sparse ``session.update`` was applied.
Parameters:
type: Event type, always "session.updated".
session: The complete resolved session, not just the changed fields.
"""
type: Literal["session.updated"]
session: SessionResource
[docs]
class OutputAudioDeltaEvent(ServerEvent):
"""A chunk of output audio in the session's audio format.
Parameters:
type: Event type, always "session.output_audio.delta".
delta: Base64-encoded audio bytes.
"""
type: Literal["session.output_audio.delta"]
delta: str
[docs]
class TranscriptDeltaEvent(ServerEvent):
"""A timed transcript fragment of user or assistant speech.
Fragments are frame-aligned, not word- or turn-aligned: accumulate them in
order, and group them into turns in the application if it needs turns.
Parameters:
type: "session.input_transcript.delta" (the user) or
"session.output_transcript.delta" (the assistant).
delta: The fragment text.
start_ms: Start of the fragment on the session timeline.
end_ms: End of the fragment on the session timeline.
"""
type: Literal["session.input_transcript.delta", "session.output_transcript.delta"]
delta: str = ""
start_ms: int | None = None
end_ms: int | None = None
@property
def role(self) -> Literal["user", "assistant"]:
"""The speaker."""
return "user" if self.type == "session.input_transcript.delta" else "assistant"
[docs]
class SessionDelegationCreatedEvent(ServerEvent):
"""The live model created a delegation.
Parameters:
type: Event type, always "session.delegation.created".
offset_ms: Position on the session timeline.
delegation: The delegation metadata.
"""
type: Literal["session.delegation.created"]
offset_ms: int | None = None
delegation: DelegationMetadata
[docs]
class ResponseOutputItem(BaseModel):
"""An output item of a Responses delegation.
Parameters:
id: Item identifier.
type: Item type, e.g. "function_call" or "message".
status: Item status, e.g. "completed".
call_id: For function calls, the call identifier to answer with.
name: For function calls, the function name.
arguments: For function calls, the JSON-encoded arguments.
"""
model_config = ConfigDict(extra="allow")
id: str | None = None
type: str | None = None
status: str | None = None
call_id: str | None = None
name: str | None = None
arguments: str | None = None
[docs]
class ResponseEventEnvelope(ServerEvent):
"""A Responses lifecycle event, wrapped with the delegation it belongs to.
The nested event keeps its Responses meaning and is dispatched by its own
complete ``type``. Lifecycle snapshots inside are reduced — ``output`` is
an empty array even at ``response.completed`` — so collect output items
from their individual events rather than from a snapshot.
Parameters:
type: Event type, always "response.event".
delegation_id: The delegation this response belongs to, when known.
event: The nested Responses event.
"""
type: Literal["response.event"]
delegation_id: str | None = None
event: dict[str, Any] = Field(default_factory=dict)
@property
def inner_type(self) -> str:
"""The nested Responses event's type, or ``""`` if it has none."""
inner = self.event.get("type")
return inner if isinstance(inner, str) else ""
[docs]
class ContextAppendedEvent(ServerEvent):
"""Appended context was accepted and placed on the session timeline.
Acceptance is not proof the model has spoken the content.
Parameters:
type: "session.instructions.appended", "session.thinking.appended" or
"session.commentary.appended".
start_ms: Start of the accepted range.
end_ms: End of the accepted range.
"""
type: Literal[
"session.instructions.appended",
"session.thinking.appended",
"session.commentary.appended",
]
start_ms: int | None = None
end_ms: int | None = None
[docs]
class ContextWindowUsage(BaseModel):
"""Context utilization, when reported.
Parameters:
usage_ratio: Fraction of the context window in use.
"""
model_config = ConfigDict(extra="allow")
usage_ratio: float | None = None
[docs]
class Usage(BaseModel):
"""Cumulative live session usage.
Backend token usage is not here: it belongs to the wrapped Responses
lifecycle, on the nested ``response.completed`` event.
Parameters:
seconds: Cumulative live audio duration in seconds.
"""
model_config = ConfigDict(extra="allow")
seconds: float | None = None
[docs]
class SessionUsageUpdatedEvent(ServerEvent):
"""Periodic cumulative usage report.
Parameters:
type: Event type, always "session.usage.updated".
usage: Cumulative usage, not an increment to sum.
context_window: Context utilization, when reported.
"""
type: Literal["session.usage.updated"]
usage: Usage
context_window: ContextWindowUsage | None = None
[docs]
class SessionClosedEvent(ServerEvent):
"""Graceful shutdown completed. Transport closure alone is not finalization.
Parameters:
type: Event type, always "session.closed".
reason: Why the session closed.
session: The session snapshot, as configuration rather than a live session.
usage: Final cumulative usage.
"""
type: Literal["session.closed"]
reason: str | None = None
session: SessionResource | None = None
usage: Usage | None = None
[docs]
class ErrorDetails(BaseModel):
"""The error envelope shared by Live API and normalized Responses errors.
Parameters:
type: Error type, e.g. "invalid_request_error".
code: Error code.
message: Human-readable message.
param: The offending field, when one caused the failure.
client_event_id: The client event the error correlates to, when known.
"""
model_config = ConfigDict(extra="allow")
type: str | None = None
code: str | None = None
message: str | None = None
param: str | None = None
client_event_id: str | None = None
[docs]
class ErrorEvent(ServerEvent):
"""A startup, validation, command or delegated-Responses error.
Parameters:
type: Event type, always "error".
error: The error details.
"""
type: Literal["error"]
error: ErrorDetails
[docs]
class UnknownServerEvent(ServerEvent):
"""A server event this module doesn't model; its fields are retained as extras."""
_server_event_types: dict[str, type[ServerEvent]] = {
"error": ErrorEvent,
"session.started": SessionStartedEvent,
"session.updated": SessionUpdatedEvent,
"session.usage.updated": SessionUsageUpdatedEvent,
"session.closed": SessionClosedEvent,
"session.output_audio.delta": OutputAudioDeltaEvent,
"session.input_transcript.delta": TranscriptDeltaEvent,
"session.output_transcript.delta": TranscriptDeltaEvent,
"session.delegation.created": SessionDelegationCreatedEvent,
"session.instructions.appended": ContextAppendedEvent,
"session.thinking.appended": ContextAppendedEvent,
"session.commentary.appended": ContextAppendedEvent,
"response.event": ResponseEventEnvelope,
}
[docs]
def parse_server_event(message: str | bytes) -> ServerEvent:
"""Parse a server event from its JSON text.
Event types this module doesn't model become :class:`UnknownServerEvent`.
Args:
message: The JSON text of one server event, as ``str`` or UTF-8 ``bytes``.
Returns:
The parsed event.
Raises:
ValueError: If the text isn't a JSON object with a string ``type``, or
a modeled event fails validation.
"""
try:
data = json.loads(message)
except json.JSONDecodeError as e:
raise ValueError(f"Invalid server event JSON: {e}") from e
if not isinstance(data, dict) or not isinstance(data.get("type"), str):
raise ValueError(f"Server event is not an object with a string type: {message}")
event_type = data["type"]
model = _server_event_types.get(event_type, UnknownServerEvent)
try:
return model.model_validate(data)
except Exception as e:
raise ValueError(f"Invalid {event_type} server event: {e}") from e