Module livekit.plugins.ultravox.realtime.events

Ultravox WebSocket event definitions based on the Ultravox data message protocol.

This module defines all the event types that can be sent and received through Ultravox's WebSocket API for real-time voice AI communication. Please note that audio data is not included in these events.

Reference: https://docs.ultravox.ai/datamessages

As of 2025-05-28, the following events are supported:

Message Sender Description
Ping Client Measures round-trip data latency.
Pong Server Server reply to a ping message.
State Server Indicates the server's current state.
Transcript Server Contains text for an utterance made during the call.
InputTextMessage Client Used to send a user message to the agent via text.
SetOutputMedium Client Sets server's output medium to text or voice.
ClientToolInvocation Server Asks the client to invoke a client tool.
ClientToolResult Client Contains the result of a client tool invocation.
Debug Server Useful for application debugging. Excluded by default.
CallStarted Server Notifies that a call has started.
PlaybackClearBuffer Server Used to clear buffered output audio. WebSocket only.

Functions

def parse_ultravox_event(data: dict[str, Any]) ‑> PingEvent | UserTextMessageEvent | SetOutputMediumEvent | ClientToolResultEvent | CallStartedEvent | PongEvent | StateEvent | TranscriptEvent | ClientToolInvocationEvent | DebugEvent | PlaybackClearBufferEvent
Expand source code
def parse_ultravox_event(data: dict[str, Any]) -> UltravoxEventType:
    """Parse a raw WebSocket message into an Ultravox event object.

    Parameters
    ----------
    data : Dict[str, Any]
        Raw JSON data from WebSocket message

    Returns
    -------
    UltravoxEventType
        Parsed event object

    Raises
    ------
    ValueError
        If the event type is unknown or data is invalid
    """
    try:
        return UltravoxEventAdapter.validate_python(data)
    except ValidationError as e:
        raise ValueError(f"Invalid event data: {data}\n{e}") from e

Parse a raw WebSocket message into an Ultravox event object.

Parameters

data : Dict[str, Any]
Raw JSON data from WebSocket message

Returns

UltravoxEventType
Parsed event object

Raises

ValueError
If the event type is unknown or data is invalid

Classes

class CallStartedEvent (**data: Any)
Expand source code
class CallStartedEvent(UltravoxEvent):
    """Server message indicating that a call has started."""

    type: Literal["call_started"] = "call_started"
    call_id: str = Field(..., alias="callId", description="Unique call ID")

Server message indicating that a call has started.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Ancestors

Class variables

var call_id : str
var model_config
var type : Literal['call_started']
class ClientToolInvocationEvent (**data: Any)
Expand source code
class ClientToolInvocationEvent(UltravoxEvent):
    """Server request for client to invoke a client-implemented tool."""

    type: Literal["client_tool_invocation"] = "client_tool_invocation"
    tool_name: str = Field(..., alias="toolName", description="Tool to invoke")
    invocation_id: str = Field(..., alias="invocationId", description="Unique invocation ID")
    parameters: dict[str, Any] = Field(..., description="Tool parameters")

Server request for client to invoke a client-implemented tool.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Ancestors

Class variables

var invocation_id : str
var model_config
var parameters : dict[str, typing.Any]
var tool_name : str
var type : Literal['client_tool_invocation']
class ClientToolResultEvent (**data: Any)
Expand source code
class ClientToolResultEvent(UltravoxEvent):
    """Contains the result of a client-implemented tool invocation."""

    type: Literal["client_tool_result"] = "client_tool_result"
    invocation_id: str = Field(
        ..., alias="invocationId", description="Matches corresponding invocation"
    )
    result: str | None = Field(None, description="Tool execution result, often JSON string")
    agent_reaction: Literal["speaks", "listens", "speaks-once"] | None = Field(
        "speaks", alias="agentReaction", description="How the agent should react to the tool result"
    )
    response_type: str = Field(
        "tool-response", alias="responseType", description="Type of response"
    )
    error_type: Literal["undefined", "implementation-error"] | None = Field(
        None, alias="errorType", description="Error type if tool execution failed"
    )
    error_message: str | None = Field(
        None, alias="errorMessage", description="Error details if failed"
    )

Contains the result of a client-implemented tool invocation.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Ancestors

Class variables

var agent_reaction : Literal['speaks', 'listens', 'speaks-once'] | None
var error_message : str | None
var error_type : Literal['undefined', 'implementation-error'] | None
var invocation_id : str
var model_config
var response_type : str
var result : str | None
var type : Literal['client_tool_result']
class DebugEvent (**data: Any)
Expand source code
class DebugEvent(UltravoxEvent):
    """Server message for debugging information."""

    type: Literal["debug"] = "debug"
    message: str = Field(..., description="Debug information")

Server message for debugging information.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Ancestors

Class variables

var message : str
var model_config
var type : Literal['debug']
class PingEvent (**data: Any)
Expand source code
class PingEvent(UltravoxEvent):
    """Client message to measure round-trip data message latency."""

    type: Literal["ping"] = "ping"
    timestamp: float = Field(..., description="Client timestamp for latency measurement")

Client message to measure round-trip data message latency.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Ancestors

Class variables

var model_config
var timestamp : float
var type : Literal['ping']
class PlaybackClearBufferEvent (**data: Any)
Expand source code
class PlaybackClearBufferEvent(UltravoxEvent):
    """Server message to clear buffered output audio (WebSocket only)."""

    type: Literal["playback_clear_buffer"] = "playback_clear_buffer"

Server message to clear buffered output audio (WebSocket only).

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Ancestors

Class variables

var model_config
var type : Literal['playback_clear_buffer']
class PongEvent (**data: Any)
Expand source code
class PongEvent(UltravoxEvent):
    """Server reply to a ping message."""

    type: Literal["pong"] = "pong"
    timestamp: float = Field(..., description="Matching ping timestamp")

Server reply to a ping message.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Ancestors

Class variables

var model_config
var timestamp : float
var type : Literal['pong']
class SetOutputMediumEvent (**data: Any)
Expand source code
class SetOutputMediumEvent(UltravoxEvent):
    """Message to set the server's output medium."""

    type: Literal["set_output_medium"] = "set_output_medium"
    medium: Literal["voice", "text"] = Field(..., description="Output medium type")

Message to set the server's output medium.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Ancestors

Class variables

var medium : Literal['voice', 'text']
var model_config
var type : Literal['set_output_medium']
class StateEvent (**data: Any)
Expand source code
class StateEvent(UltravoxEvent):
    """Server message indicating its current state."""

    type: Literal["state"] = "state"
    state: str = Field(..., description="Current session state")

Server message indicating its current state.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Ancestors

Class variables

var model_config
var state : str
var type : Literal['state']
class TranscriptEvent (**data: Any)
Expand source code
class TranscriptEvent(UltravoxEvent):
    """Message containing text transcripts of user and agent utterances."""

    type: Literal["transcript"] = "transcript"
    role: Literal["user", "agent"] = Field(..., description="Who emitted the utterance")
    medium: Literal["text", "voice"] = Field(
        ..., description="Medium through which utterance was emitted"
    )
    text: str | None = Field(None, description="Full transcript text (exclusive with delta)")
    delta: str | None = Field(
        None, description="Incremental transcript update (exclusive with text)"
    )
    final: bool = Field(..., description="Whether more updates are expected for this utterance")
    ordinal: int = Field(..., description="Used for ordering transcripts within a call")

Message containing text transcripts of user and agent utterances.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Ancestors

Class variables

var delta : str | None
var final : bool
var medium : Literal['voice', 'text']
var model_config
var ordinal : int
var role : Literal['user', 'agent']
var text : str | None
var type : Literal['transcript']
class UltravoxEvent (**data: Any)
Expand source code
class UltravoxEvent(BaseModel):
    """Base class for all Ultravox WebSocket events."""

    type: str = Field(..., description="Event type identifier")

Base class for all Ultravox WebSocket events.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Ancestors

  • pydantic.main.BaseModel

Subclasses

Class variables

var model_config
var type : str
class UserTextMessageEvent (**data: Any)
Expand source code
class UserTextMessageEvent(UltravoxEvent):
    """User message sent via text (UserTextMessage in Ultravox docs)."""

    type: Literal["user_text_message"] = "user_text_message"
    text: str = Field(..., description="The content of the user message")
    urgency: Literal["immediate", "soon", "later"] | None = Field(
        None,
        description="Message urgency level - immediate for barge-in, later for context updates",
    )
    defer_response: bool | None = Field(
        None,
        alias="deferResponse",
        description="If true, allows adding text without inducing immediate response",
    )

User message sent via text (UserTextMessage in Ultravox docs).

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Ancestors

Class variables

var defer_response : bool | None
var model_config
var text : str
var type : Literal['user_text_message']
var urgency : Literal['immediate', 'soon', 'later'] | None