Rtvi

class pipecat.processors.frameworks.rtvi.RTVIServiceOption(**data)

Bases: BaseModel

Parameters:

data (Any)

name: str

type: Literal['bool', 'number', 'string', 'array', 'object']

handler: Callable[[RTVIProcessor, str, RTVIServiceOptionConfig], Awaitable[None]]

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pipecat.processors.frameworks.rtvi.RTVIService(**data)

Bases: BaseModel

Parameters:

data (Any)

name: str

options: List[RTVIServiceOption]

model_post_init(context, /)

We need to both initialize private attributes and call the user-defined model_post_init method.

Parameters:

• self (BaseModel)

• context (Any)

Return type:

None

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pipecat.processors.frameworks.rtvi.RTVIActionArgumentData(*, name, value)

Bases: BaseModel

Parameters:

• name (str)

• value (Any)

name: str

value: Any

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pipecat.processors.frameworks.rtvi.RTVIActionArgument(*, name, type)

Bases: BaseModel

Parameters:

• name (str)

• type (Literal['bool', 'number', 'string', 'array', 'object'])

name: str

type: Literal['bool', 'number', 'string', 'array', 'object']

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pipecat.processors.frameworks.rtvi.RTVIAction(**data)

Bases: BaseModel

Parameters:

data (Any)

service: str

action: str

arguments: List[RTVIActionArgument]

result: Literal['bool', 'number', 'string', 'array', 'object']

handler: Callable[[RTVIProcessor, str, Dict[str, Any]], Awaitable[bool | int | float | str | list | dict]]

model_post_init(context, /)

We need to both initialize private attributes and call the user-defined model_post_init method.

Parameters:

• self (BaseModel)

• context (Any)

Return type:

None

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pipecat.processors.frameworks.rtvi.RTVIServiceOptionConfig(*, name, value)

Bases: BaseModel

Parameters:

• name (str)

• value (Any)

name: str

value: Any

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pipecat.processors.frameworks.rtvi.RTVIServiceConfig(*, service, options)

Bases: BaseModel

Parameters:

• service (str)

• options (List[RTVIServiceOptionConfig])

service: str

options: List[RTVIServiceOptionConfig]

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pipecat.processors.frameworks.rtvi.RTVIConfig(*, config)

Bases: BaseModel

Parameters:

config (List[RTVIServiceConfig])

config: List[RTVIServiceConfig]

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pipecat.processors.frameworks.rtvi.RTVIUpdateConfig(*, config, interrupt=False)

Bases: BaseModel

Parameters:

• config (List[RTVIServiceConfig])

• interrupt (bool)

config: List[RTVIServiceConfig]

interrupt: bool

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pipecat.processors.frameworks.rtvi.RTVIActionRunArgument(*, name, value)

Bases: BaseModel

Parameters:

• name (str)

• value (Any)

name: str

value: Any

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pipecat.processors.frameworks.rtvi.RTVIActionRun(*, service, action, arguments=None)

Bases: BaseModel

Parameters:

• service (str)

• action (str)

• arguments (List[RTVIActionRunArgument] | None)

service: str

action: str

arguments: List[RTVIActionRunArgument] | None

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pipecat.processors.frameworks.rtvi.RTVIActionFrame(rtvi_action_run: pipecat.processors.frameworks.rtvi.RTVIActionRun, message_id: str | None = None)

Bases: DataFrame

Parameters:

• rtvi_action_run (RTVIActionRun)

• message_id (str | None)

rtvi_action_run: RTVIActionRun

message_id: str | None = None

class pipecat.processors.frameworks.rtvi.RTVIMessage(*, label='rtvi-ai', type, id, data=None)

Bases: BaseModel

Parameters:

• label (Literal['rtvi-ai'])

• type (str)

• id (str)

• data (Dict[str, Any] | None)

label: Literal['rtvi-ai']

type: str

id: str

data: Dict[str, Any] | None

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pipecat.processors.frameworks.rtvi.RTVIErrorResponseData(*, error)

Bases: BaseModel

Parameters:

error (str)

error: str

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pipecat.processors.frameworks.rtvi.RTVIErrorResponse(*, label='rtvi-ai', type='error-response', id, data)

Bases: BaseModel

Parameters:

• label (Literal['rtvi-ai'])

• type (Literal['error-response'])

• id (str)

• data (RTVIErrorResponseData)

label: Literal['rtvi-ai']

type: Literal['error-response']

id: str

data: RTVIErrorResponseData

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pipecat.processors.frameworks.rtvi.RTVIErrorData(*, error, fatal)

Bases: BaseModel

Parameters:

• error (str)

• fatal (bool)

error: str

fatal: bool

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pipecat.processors.frameworks.rtvi.RTVIError(*, label='rtvi-ai', type='error', data)

Bases: BaseModel

Parameters:

• label (Literal['rtvi-ai'])

• type (Literal['error'])

• data (RTVIErrorData)

label: Literal['rtvi-ai']

type: Literal['error']

data: RTVIErrorData

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pipecat.processors.frameworks.rtvi.RTVIDescribeConfigData(**data)

Bases: BaseModel

Parameters:

data (Any)

config: List[RTVIService]

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pipecat.processors.frameworks.rtvi.RTVIDescribeConfig(**data)

Bases: BaseModel

Parameters:

data (RTVIDescribeConfigData)

label: Literal['rtvi-ai']

type: Literal['config-available']

id: str

data: RTVIDescribeConfigData

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pipecat.processors.frameworks.rtvi.RTVIDescribeActionsData(**data)

Bases: BaseModel

Parameters:

data (Any)

actions: List[RTVIAction]

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pipecat.processors.frameworks.rtvi.RTVIDescribeActions(**data)

Bases: BaseModel

Parameters:

data (RTVIDescribeActionsData)

label: Literal['rtvi-ai']

type: Literal['actions-available']

id: str

data: RTVIDescribeActionsData

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pipecat.processors.frameworks.rtvi.RTVIConfigResponse(*, label='rtvi-ai', type='config', id, data)

Bases: BaseModel

Parameters:

• label (Literal['rtvi-ai'])

• type (Literal['config'])

• id (str)

• data (RTVIConfig)

label: Literal['rtvi-ai']

type: Literal['config']

id: str

data: RTVIConfig

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pipecat.processors.frameworks.rtvi.RTVIActionResponseData(*, result)

Bases: BaseModel

Parameters:

result (bool | int | float | str | list | dict)

result: bool | int | float | str | list | dict

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pipecat.processors.frameworks.rtvi.RTVIActionResponse(*, label='rtvi-ai', type='action-response', id, data)

Bases: BaseModel

Parameters:

• label (Literal['rtvi-ai'])

• type (Literal['action-response'])

• id (str)

• data (RTVIActionResponseData)

label: Literal['rtvi-ai']

type: Literal['action-response']

id: str

data: RTVIActionResponseData

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pipecat.processors.frameworks.rtvi.RTVIBotReadyData(*, version, config)

Bases: BaseModel

Parameters:

• version (str)

• config (List[RTVIServiceConfig])

version: str

config: List[RTVIServiceConfig]

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pipecat.processors.frameworks.rtvi.RTVIBotReady(*, label='rtvi-ai', type='bot-ready', id, data)

Bases: BaseModel

Parameters:

• label (Literal['rtvi-ai'])

• type (Literal['bot-ready'])

• id (str)

• data (RTVIBotReadyData)

label: Literal['rtvi-ai']

type: Literal['bot-ready']

id: str

data: RTVIBotReadyData

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pipecat.processors.frameworks.rtvi.RTVILLMFunctionCallMessageData(*, function_name, tool_call_id, args)

Bases: BaseModel

Parameters:

• function_name (str)

• tool_call_id (str)

• args (Mapping[str, Any])

function_name: str

tool_call_id: str

args: Mapping[str, Any]

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pipecat.processors.frameworks.rtvi.RTVILLMFunctionCallMessage(*, label='rtvi-ai', type='llm-function-call', data)

Bases: BaseModel

Parameters:

• label (Literal['rtvi-ai'])

• type (Literal['llm-function-call'])

• data (RTVILLMFunctionCallMessageData)

label: Literal['rtvi-ai']

type: Literal['llm-function-call']

data: RTVILLMFunctionCallMessageData

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pipecat.processors.frameworks.rtvi.RTVILLMFunctionCallStartMessageData(*, function_name)

Bases: BaseModel

Parameters:

function_name (str)

function_name: str

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pipecat.processors.frameworks.rtvi.RTVILLMFunctionCallStartMessage(*, label='rtvi-ai', type='llm-function-call-start', data)

Bases: BaseModel

Parameters:

• label (Literal['rtvi-ai'])

• type (Literal['llm-function-call-start'])

• data (RTVILLMFunctionCallStartMessageData)

label: Literal['rtvi-ai']

type: Literal['llm-function-call-start']

data: RTVILLMFunctionCallStartMessageData

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pipecat.processors.frameworks.rtvi.RTVILLMFunctionCallResultData(*, function_name, tool_call_id, arguments, result)

Bases: BaseModel

Parameters:

• function_name (str)

• tool_call_id (str)

• arguments (dict)

• result (dict | str)

function_name: str

tool_call_id: str

arguments: dict

result: dict | str

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pipecat.processors.frameworks.rtvi.RTVIBotLLMStartedMessage(*, label='rtvi-ai', type='bot-llm-started')

Bases: BaseModel

Parameters:

• label (Literal['rtvi-ai'])

• type (Literal['bot-llm-started'])

label: Literal['rtvi-ai']

type: Literal['bot-llm-started']

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pipecat.processors.frameworks.rtvi.RTVIBotLLMStoppedMessage(*, label='rtvi-ai', type='bot-llm-stopped')

Bases: BaseModel

Parameters:

• label (Literal['rtvi-ai'])

• type (Literal['bot-llm-stopped'])

label: Literal['rtvi-ai']

type: Literal['bot-llm-stopped']

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pipecat.processors.frameworks.rtvi.RTVIBotTTSStartedMessage(*, label='rtvi-ai', type='bot-tts-started')

Bases: BaseModel

Parameters:

• label (Literal['rtvi-ai'])

• type (Literal['bot-tts-started'])

label: Literal['rtvi-ai']

type: Literal['bot-tts-started']

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pipecat.processors.frameworks.rtvi.RTVIBotTTSStoppedMessage(*, label='rtvi-ai', type='bot-tts-stopped')

Bases: BaseModel

Parameters:

• label (Literal['rtvi-ai'])

• type (Literal['bot-tts-stopped'])

label: Literal['rtvi-ai']

type: Literal['bot-tts-stopped']

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pipecat.processors.frameworks.rtvi.RTVITextMessageData(*, text)

Bases: BaseModel

Parameters:

text (str)

text: str

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pipecat.processors.frameworks.rtvi.RTVIBotTranscriptionMessage(*, label='rtvi-ai', type='bot-transcription', data)

Bases: BaseModel

Parameters:

• label (Literal['rtvi-ai'])

• type (Literal['bot-transcription'])

• data (RTVITextMessageData)

label: Literal['rtvi-ai']

type: Literal['bot-transcription']

data: RTVITextMessageData

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pipecat.processors.frameworks.rtvi.RTVIBotLLMTextMessage(*, label='rtvi-ai', type='bot-llm-text', data)

Bases: BaseModel

Parameters:

• label (Literal['rtvi-ai'])

• type (Literal['bot-llm-text'])

• data (RTVITextMessageData)

label: Literal['rtvi-ai']

type: Literal['bot-llm-text']

data: RTVITextMessageData

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pipecat.processors.frameworks.rtvi.RTVIBotTTSTextMessage(*, label='rtvi-ai', type='bot-tts-text', data)

Bases: BaseModel

Parameters:

• label (Literal['rtvi-ai'])

• type (Literal['bot-tts-text'])

• data (RTVITextMessageData)

label: Literal['rtvi-ai']

type: Literal['bot-tts-text']

data: RTVITextMessageData

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pipecat.processors.frameworks.rtvi.RTVIAudioMessageData(*, audio, sample_rate, num_channels)

Bases: BaseModel

Parameters:

• audio (str)

• sample_rate (int)

• num_channels (int)

audio: str

sample_rate: int

num_channels: int

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pipecat.processors.frameworks.rtvi.RTVIBotTTSAudioMessage(*, label='rtvi-ai', type='bot-tts-audio', data)

Bases: BaseModel

Parameters:

• label (Literal['rtvi-ai'])

• type (Literal['bot-tts-audio'])

• data (RTVIAudioMessageData)

label: Literal['rtvi-ai']

type: Literal['bot-tts-audio']

data: RTVIAudioMessageData

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pipecat.processors.frameworks.rtvi.RTVIUserTranscriptionMessageData(*, text, user_id, timestamp, final)

Bases: BaseModel

Parameters:

• text (str)

• user_id (str)

• timestamp (str)

• final (bool)

text: str

user_id: str

timestamp: str

final: bool

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pipecat.processors.frameworks.rtvi.RTVIUserTranscriptionMessage(*, label='rtvi-ai', type='user-transcription', data)

Bases: BaseModel

Parameters:

• label (Literal['rtvi-ai'])

• type (Literal['user-transcription'])

• data (RTVIUserTranscriptionMessageData)

label: Literal['rtvi-ai']

type: Literal['user-transcription']

data: RTVIUserTranscriptionMessageData

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pipecat.processors.frameworks.rtvi.RTVIUserLLMTextMessage(*, label='rtvi-ai', type='user-llm-text', data)

Bases: BaseModel

Parameters:

• label (Literal['rtvi-ai'])

• type (Literal['user-llm-text'])

• data (RTVITextMessageData)

label: Literal['rtvi-ai']

type: Literal['user-llm-text']

data: RTVITextMessageData

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pipecat.processors.frameworks.rtvi.RTVIUserStartedSpeakingMessage(*, label='rtvi-ai', type='user-started-speaking')

Bases: BaseModel

Parameters:

• label (Literal['rtvi-ai'])

• type (Literal['user-started-speaking'])

label: Literal['rtvi-ai']

type: Literal['user-started-speaking']

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pipecat.processors.frameworks.rtvi.RTVIUserStoppedSpeakingMessage(*, label='rtvi-ai', type='user-stopped-speaking')

Bases: BaseModel

Parameters:

• label (Literal['rtvi-ai'])

• type (Literal['user-stopped-speaking'])

label: Literal['rtvi-ai']

type: Literal['user-stopped-speaking']

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pipecat.processors.frameworks.rtvi.RTVIBotStartedSpeakingMessage(*, label='rtvi-ai', type='bot-started-speaking')

Bases: BaseModel

Parameters:

• label (Literal['rtvi-ai'])

• type (Literal['bot-started-speaking'])

label: Literal['rtvi-ai']

type: Literal['bot-started-speaking']

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pipecat.processors.frameworks.rtvi.RTVIBotStoppedSpeakingMessage(*, label='rtvi-ai', type='bot-stopped-speaking')

Bases: BaseModel

Parameters:

• label (Literal['rtvi-ai'])

• type (Literal['bot-stopped-speaking'])

label: Literal['rtvi-ai']

type: Literal['bot-stopped-speaking']

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pipecat.processors.frameworks.rtvi.RTVIMetricsMessage(*, label='rtvi-ai', type='metrics', data)

Bases: BaseModel

Parameters:

• label (Literal['rtvi-ai'])

• type (Literal['metrics'])

• data (Mapping[str, Any])

label: Literal['rtvi-ai']

type: Literal['metrics']

data: Mapping[str, Any]

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pipecat.processors.frameworks.rtvi.RTVIServerMessage(*, label='rtvi-ai', type='server-message', data)

Bases: BaseModel

Parameters:

• label (Literal['rtvi-ai'])

• type (Literal['server-message'])

• data (Any)

label: Literal['rtvi-ai']

type: Literal['server-message']

data: Any

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pipecat.processors.frameworks.rtvi.RTVIServerMessageFrame(data)

Bases: SystemFrame

A frame for sending server messages to the client.

Parameters:

data (Any)

data: Any

class pipecat.processors.frameworks.rtvi.RTVIObserverParams(bot_llm_enabled=True, bot_tts_enabled=True, bot_speaking_enabled=True, user_llm_enabled=True, user_speaking_enabled=True, user_transcription_enabled=True, metrics_enabled=True, errors_enabled=True)

Bases: object

Parameters for configuring RTVI Observer behavior.

Parameters:

• bot_llm_enabled (bool)

• bot_tts_enabled (bool)

• bot_speaking_enabled (bool)

• user_llm_enabled (bool)

• user_speaking_enabled (bool)

• user_transcription_enabled (bool)

• metrics_enabled (bool)

• errors_enabled (bool)

bot_llm_enabled

Indicates if the bot’s LLM messages should be sent.

Type:

bool

bot_tts_enabled

Indicates if the bot’s TTS messages should be sent.

Type:

bool

bot_speaking_enabled

Indicates if the bot’s started/stopped speaking messages should be sent.

Type:

bool

user_llm_enabled

Indicates if the user’s LLM input messages should be sent.

Type:

bool

user_speaking_enabled

Indicates if the user’s started/stopped speaking messages should be sent.

Type:

bool

user_transcription_enabled

Indicates if user’s transcription messages should be sent.

Type:

bool

metrics_enabled

Indicates if metrics messages should be sent.

Type:

bool

errors_enabled

Indicates if errors messages should be sent.

Type:

bool

bot_llm_enabled: bool = True

bot_tts_enabled: bool = True

bot_speaking_enabled: bool = True

user_llm_enabled: bool = True

user_speaking_enabled: bool = True

user_transcription_enabled: bool = True

metrics_enabled: bool = True

errors_enabled: bool = True

class pipecat.processors.frameworks.rtvi.RTVIObserver(rtvi, *, params=None, **kwargs)

Bases: BaseObserver

Pipeline frame observer for RTVI server message handling.

This observer monitors pipeline frames and converts them into appropriate RTVI messages for client communication. It handles various frame types including speech events, transcriptions, LLM responses, and TTS events.

Note

This observer only handles outgoing messages. Incoming RTVI client messages are handled by the RTVIProcessor.

Parameters:

• rtvi (RTVIProcessor) – The RTVI processor to push frames to.

• params (RTVIObserverParams) – Settings to enable/disable specific messages.

async on_push_frame(data)

Process a frame being pushed through the pipeline.

Parameters:

• src – Source processor pushing the frame

• dst – Destination processor receiving the frame

• frame – The frame being pushed

• direction – Direction of frame flow in pipeline

• timestamp – Time when frame was pushed

• data (FramePushed)

async push_transport_message_urgent(model, exclude_none=True)

Push an urgent transport message to the RTVI processor.

Parameters:

• model (BaseModel) – The message model to send

• exclude_none (bool) – Whether to exclude None values from the model dump

class pipecat.processors.frameworks.rtvi.RTVIProcessor(*, config=None, transport=None, **kwargs)

Bases: FrameProcessor

Parameters:

• config (RTVIConfig | None)

• transport (BaseTransport | None)

register_action(action)

Parameters:

action (RTVIAction)

register_service(service)

Parameters:

service (RTVIService)

async set_client_ready()

async set_bot_ready()

set_errors_enabled(enabled)

Parameters:

enabled (bool)

async interrupt_bot()

async send_error(error)

Parameters:

error (str)

async handle_message(message)

Parameters:

message (RTVIMessage)

async handle_function_call(params)

Parameters:

params (FunctionCallParams)

async handle_function_call_start(function_name, llm, context)

Parameters:

• function_name (str)

• llm (FrameProcessor)

• context (OpenAILLMContext)

async process_frame(frame, direction)

Parameters:

• frame (Frame)

• direction (FrameDirection)