#
# Copyright (c) 2024–2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
"""RTVI protocol v1 message models.
Contains all RTVI protocol v1 message definitions and data structures.
Import this module under the ``RTVI`` alias to use as a namespace::
import pipecat.processors.frameworks.rtvi.models as RTVI
msg = RTVI.BotReady(id="1", data=RTVI.BotReadyData(version=RTVI.PROTOCOL_VERSION))
"""
from collections.abc import Mapping
from typing import (
Any,
Literal,
)
from pydantic import BaseModel
from pipecat.frames.frames import (
AggregationType,
)
# -- Constants --
PROTOCOL_VERSION = "1.2.0"
MESSAGE_LABEL = "rtvi-ai"
MessageLiteral = Literal["rtvi-ai"]
# -- Base Message Structure --
[docs]
class Message(BaseModel):
"""Base RTVI message structure.
Represents the standard format for RTVI protocol messages.
"""
label: MessageLiteral = MESSAGE_LABEL
type: str
id: str
data: dict[str, Any] | None = None
# -- Client -> Pipecat messages.
[docs]
class RawClientMessageData(BaseModel):
"""Data structure expected from client messages sent to the RTVI server."""
t: str
d: Any | None = None
[docs]
class ClientMessage(BaseModel):
"""Cleansed data structure for client messages for handling."""
msg_id: str
type: str
data: Any | None = None
[docs]
class RawServerResponseData(BaseModel):
"""Data structure for server responses to client messages."""
t: str
d: Any | None = None
[docs]
class ServerResponse(BaseModel):
"""The RTVI-formatted message response from the server to the client.
This message is used to respond to custom messages sent by the client.
"""
label: MessageLiteral = MESSAGE_LABEL
type: Literal["server-response"] = "server-response"
id: str
data: RawServerResponseData
[docs]
class AboutClientData(BaseModel):
"""Data about the RTVI client.
Contains information about the client, including which RTVI library it
is using, what platform it is on and any additional details, if available.
"""
library: str
library_version: str | None = None
platform: str | None = None
platform_version: str | None = None
platform_details: Any | None = None
[docs]
class ClientReadyData(BaseModel):
"""Data format of client ready messages.
Contains the RTVI protocol version and client information.
"""
version: str
about: AboutClientData
# -- Pipecat -> Client errors
[docs]
class ErrorResponseData(BaseModel):
"""Data for an RTVI error response.
Contains the error message to send back to the client.
"""
error: str
[docs]
class ErrorResponse(BaseModel):
"""RTVI error response message.
RTVI formatted error response message for relaying failed client requests.
"""
label: MessageLiteral = MESSAGE_LABEL
type: Literal["error-response"] = "error-response"
id: str
data: ErrorResponseData
[docs]
class ErrorData(BaseModel):
"""Data for an RTVI error event.
Contains error information including whether it's fatal.
"""
error: str
fatal: bool # Indicates the pipeline has stopped due to this error
[docs]
class Error(BaseModel):
"""RTVI error event message.
RTVI formatted error message for relaying errors in the pipeline.
"""
label: MessageLiteral = MESSAGE_LABEL
type: Literal["error"] = "error"
data: ErrorData
# -- Pipecat -> Client responses and messages.
[docs]
class BotReadyData(BaseModel):
"""Data for bot ready notification.
Contains protocol version and initial configuration.
"""
version: str
about: Mapping[str, Any] | None = None
[docs]
class BotReady(BaseModel):
"""Message indicating bot is ready for interaction.
Sent after bot initialization is complete.
"""
label: MessageLiteral = MESSAGE_LABEL
type: Literal["bot-ready"] = "bot-ready"
id: str
data: BotReadyData
[docs]
class LLMFunctionCallMessageData(BaseModel):
"""Data for LLM function call notification.
Contains function call details including name, ID, and arguments.
.. deprecated:: 0.0.102
Use ``LLMFunctionCallInProgressMessageData`` instead.
"""
function_name: str
tool_call_id: str
args: Mapping[str, Any]
[docs]
class LLMFunctionCallMessage(BaseModel):
"""Message notifying of an LLM function call.
Sent when the LLM makes a function call.
.. deprecated:: 0.0.102
Use ``LLMFunctionCallInProgressMessage`` with the
``llm-function-call-in-progress`` event type instead.
"""
label: MessageLiteral = MESSAGE_LABEL
type: Literal["llm-function-call"] = "llm-function-call"
data: LLMFunctionCallMessageData
[docs]
class SendTextOptions(BaseModel):
"""Options for sending text input to the LLM.
Contains options for how the pipeline should process the text input.
"""
run_immediately: bool = True
audio_response: bool = True
[docs]
class SendTextData(BaseModel):
"""Data format for sending text input to the LLM.
Contains the text content to send and any options for how the pipeline should process it.
"""
content: str
options: SendTextOptions | None = None
[docs]
class LLMFunctionCallStartMessageData(BaseModel):
"""Data for LLM function call start notification.
Contains the function name being called. Fields may be omitted based on
the configured function_call_report_level for security.
"""
function_name: str | None = None
[docs]
class LLMFunctionCallStartMessage(BaseModel):
"""Message notifying that an LLM function call has started.
Sent when the LLM begins a function call.
"""
label: MessageLiteral = MESSAGE_LABEL
type: Literal["llm-function-call-started"] = "llm-function-call-started"
data: LLMFunctionCallStartMessageData
[docs]
class LLMFunctionCallResultData(BaseModel):
"""Data for LLM function call result.
Contains function call details and result.
"""
function_name: str
tool_call_id: str
arguments: dict
result: dict | str
[docs]
class LLMFunctionCallInProgressMessageData(BaseModel):
"""Data for LLM function call in-progress notification.
Contains function call details including name, ID, and arguments.
Fields may be omitted based on the configured function_call_report_level for security.
"""
tool_call_id: str
function_name: str | None = None
arguments: Mapping[str, Any] | None = None
[docs]
class LLMFunctionCallInProgressMessage(BaseModel):
"""Message notifying that an LLM function call is in progress.
Sent when the LLM function call execution begins.
"""
label: MessageLiteral = MESSAGE_LABEL
type: Literal["llm-function-call-in-progress"] = "llm-function-call-in-progress"
data: LLMFunctionCallInProgressMessageData
[docs]
class LLMFunctionCallStoppedMessageData(BaseModel):
"""Data for LLM function call stopped notification.
Contains details about the function call that stopped, including
whether it was cancelled or completed with a result.
Fields may be omitted based on the configured function_call_report_level for security.
"""
tool_call_id: str
cancelled: bool
function_name: str | None = None
result: Any | None = None
[docs]
class LLMFunctionCallStoppedMessage(BaseModel):
"""Message notifying that an LLM function call has stopped.
Sent when a function call completes (with result) or is cancelled.
"""
label: MessageLiteral = MESSAGE_LABEL
type: Literal["llm-function-call-stopped"] = "llm-function-call-stopped"
data: LLMFunctionCallStoppedMessageData
[docs]
class BotLLMStartedMessage(BaseModel):
"""Message indicating bot LLM processing has started."""
label: MessageLiteral = MESSAGE_LABEL
type: Literal["bot-llm-started"] = "bot-llm-started"
[docs]
class BotLLMStoppedMessage(BaseModel):
"""Message indicating bot LLM processing has stopped."""
label: MessageLiteral = MESSAGE_LABEL
type: Literal["bot-llm-stopped"] = "bot-llm-stopped"
[docs]
class BotTTSStartedMessage(BaseModel):
"""Message indicating bot TTS processing has started."""
label: MessageLiteral = MESSAGE_LABEL
type: Literal["bot-tts-started"] = "bot-tts-started"
[docs]
class BotTTSStoppedMessage(BaseModel):
"""Message indicating bot TTS processing has stopped."""
label: MessageLiteral = MESSAGE_LABEL
type: Literal["bot-tts-stopped"] = "bot-tts-stopped"
[docs]
class TextMessageData(BaseModel):
"""Data for text-based RTVI messages.
Contains text content.
"""
text: str
[docs]
class BotOutputMessageData(TextMessageData):
"""Data for bot output RTVI messages.
Extends TextMessageData to include metadata about the output.
"""
spoken: bool = False # Indicates if the text has been spoken by TTS
aggregated_by: AggregationType | str
# Indicates what form the text is in (e.g., by word, sentence, etc.)
[docs]
class BotOutputMessage(BaseModel):
"""Message containing bot output text.
An event meant to holistically represent what the bot is outputting,
along with metadata about the output and if it has been spoken.
"""
label: MessageLiteral = MESSAGE_LABEL
type: Literal["bot-output"] = "bot-output"
data: BotOutputMessageData
[docs]
class BotTranscriptionMessage(BaseModel):
"""Message containing bot transcription text.
Sent when the bot's speech is transcribed.
"""
label: MessageLiteral = MESSAGE_LABEL
type: Literal["bot-transcription"] = "bot-transcription"
data: TextMessageData
[docs]
class BotLLMTextMessage(BaseModel):
"""Message containing bot LLM text output.
Sent when the bot's LLM generates text.
"""
label: MessageLiteral = MESSAGE_LABEL
type: Literal["bot-llm-text"] = "bot-llm-text"
data: TextMessageData
[docs]
class BotTTSTextMessage(BaseModel):
"""Message containing bot TTS text output.
Sent when text is being processed by TTS.
"""
label: MessageLiteral = MESSAGE_LABEL
type: Literal["bot-tts-text"] = "bot-tts-text"
data: TextMessageData
[docs]
class AudioMessageData(BaseModel):
"""Data for audio-based RTVI messages.
Contains audio data and metadata.
"""
audio: str
sample_rate: int
num_channels: int
[docs]
class BotTTSAudioMessage(BaseModel):
"""Message containing bot TTS audio output.
Sent when the bot's TTS generates audio.
"""
label: MessageLiteral = MESSAGE_LABEL
type: Literal["bot-tts-audio"] = "bot-tts-audio"
data: AudioMessageData
[docs]
class UserTranscriptionMessageData(BaseModel):
"""Data for user transcription messages.
Contains transcription text and metadata.
"""
text: str
user_id: str
timestamp: str
final: bool
[docs]
class UserTranscriptionMessage(BaseModel):
"""Message containing user transcription.
Sent when user speech is transcribed.
"""
label: MessageLiteral = MESSAGE_LABEL
type: Literal["user-transcription"] = "user-transcription"
data: UserTranscriptionMessageData
[docs]
class UserLLMTextMessage(BaseModel):
"""Message containing user text input for LLM.
Sent when user text is processed by the LLM.
"""
label: MessageLiteral = MESSAGE_LABEL
type: Literal["user-llm-text"] = "user-llm-text"
data: TextMessageData
[docs]
class UserStartedSpeakingMessage(BaseModel):
"""Message indicating user has started speaking."""
label: MessageLiteral = MESSAGE_LABEL
type: Literal["user-started-speaking"] = "user-started-speaking"
[docs]
class UserStoppedSpeakingMessage(BaseModel):
"""Message indicating user has stopped speaking."""
label: MessageLiteral = MESSAGE_LABEL
type: Literal["user-stopped-speaking"] = "user-stopped-speaking"
[docs]
class UserMuteStartedMessage(BaseModel):
"""Message indicating user has been muted."""
label: MessageLiteral = MESSAGE_LABEL
type: Literal["user-mute-started"] = "user-mute-started"
[docs]
class UserMuteStoppedMessage(BaseModel):
"""Message indicating user has been unmuted."""
label: MessageLiteral = MESSAGE_LABEL
type: Literal["user-mute-stopped"] = "user-mute-stopped"
[docs]
class BotStartedSpeakingMessage(BaseModel):
"""Message indicating bot has started speaking."""
label: MessageLiteral = MESSAGE_LABEL
type: Literal["bot-started-speaking"] = "bot-started-speaking"
[docs]
class BotStoppedSpeakingMessage(BaseModel):
"""Message indicating bot has stopped speaking."""
label: MessageLiteral = MESSAGE_LABEL
type: Literal["bot-stopped-speaking"] = "bot-stopped-speaking"
[docs]
class MetricsMessage(BaseModel):
"""Message containing performance metrics.
Sent to provide performance and usage metrics.
"""
label: MessageLiteral = MESSAGE_LABEL
type: Literal["metrics"] = "metrics"
data: Mapping[str, Any]
[docs]
class ServerMessage(BaseModel):
"""Generic server message.
Used for custom server-to-client messages.
"""
label: MessageLiteral = MESSAGE_LABEL
type: Literal["server-message"] = "server-message"
data: Any
[docs]
class AudioLevelMessageData(BaseModel):
"""Data format for sending audio levels."""
value: float
[docs]
class UserAudioLevelMessage(BaseModel):
"""Message indicating user audio level."""
label: MessageLiteral = MESSAGE_LABEL
type: Literal["user-audio-level"] = "user-audio-level"
data: AudioLevelMessageData
[docs]
class BotAudioLevelMessage(BaseModel):
"""Message indicating bot audio level."""
label: MessageLiteral = MESSAGE_LABEL
type: Literal["bot-audio-level"] = "bot-audio-level"
data: AudioLevelMessageData
[docs]
class SystemLogMessage(BaseModel):
"""Message including a system log."""
label: MessageLiteral = MESSAGE_LABEL
type: Literal["system-log"] = "system-log"
data: TextMessageData