RTVIProcessor

Core coordinator for RTVI protocol communication

The RTVIProcessor manages bidirectional communication between clients and your Pipecat application. It processes client messages, handles service configuration, executes actions, and coordinates function calls.

Initialization

RTVIProcessor is automatically added to your pipeline when you create a PipelineTask. Access it via task.rtvi:

pipeline = Pipeline([
    transport.input(),
    stt,
    # ... other processors ...
    transport.output()
])

task = PipelineTask(pipeline)

# Access the processor
rtvi = task.rtvi

To provide a custom processor (e.g., for Google RTVI):

from pipecat.services.google.rtvi import GoogleRTVIProcessor

task = PipelineTask(pipeline, rtvi_processor=GoogleRTVIProcessor())

To disable RTVI entirely, set enable_rtvi=False.

Readiness Protocol

Client Ready State

Clients indicate readiness by sending a client-ready message, triggering the on_client_ready event in the processor:

@rtvi.event_handler("on_client_ready")
async def on_client_ready(rtvi):
    # Handle client ready state
    await rtvi.set_bot_ready()
    # Initialize conversation
    await task.queue_frames([...])

The processor validates the client’s RTVI protocol version during the handshake. If the client’s major version doesn’t match the server’s protocol version, an error response is sent but the connection continues. Check server logs for version compatibility warnings.

Bot Ready State

The server must mark the bot as ready before it can process client messages:

await rtvi.set_bot_ready()

When marked ready, the bot sends a response containing:

• RTVI protocol version
• Current service configuration
• Available actions

Services

Services represent configurable components of your application that clients can interact with.

Registering Services

# 1. Define option handler
async def handle_voice_option(processor, service, option):
    voice_id = option.value
    # Apply configuration change
    logger.info(f"Voice ID updated to: {voice_id}")

# 2. Create RTVIService
voice_service = RTVIService(
    name="voice",
    options=[
        RTVIServiceOption(
            name="voice_id",
            type="string",
            handler=handle_voice_option
        )
    ]
)

# 3. Register with processor
rtvi.register_service(voice_service)

Option Types

Services support multiple data types for configuration:

RTVIServiceOption(
    name="temperature",
    type="number",  # number, string, bool, array, object
    handler=handle_temperature
)

Option handlers receive:

• The processor instance
• The service name
• The option configuration with new value

Actions

Actions are server-side functions that clients can trigger with arguments.

Registering Actions

# 1. Define handler function
async def handle_print_message(processor, service, arguments):
    message = arguments.get("message", "Default message")
    logger.info(f"Print action triggered with message: {message}")
    return True

# 2. Create and register RTVIAction
print_action = RTVIAction(
    service="conversation",
    action="print_message",
    arguments=[
        RTVIActionArgument(name="message", type="string")
    ],
    result="bool",
    handler=handle_print_message
)
rtvi.register_action(print_action)

Action Arguments

Actions can accept typed arguments from clients:

search_action = RTVIAction(
    service="knowledge",
    action="search",
    arguments=[
        RTVIActionArgument(name="query", type="string"),
        RTVIActionArgument(name="limit", type="number")
    ],
    result="array",
    handler=handle_search
)

Function Calls

Handle LLM function calls with client interaction:

await processor.handle_function_call(
    function_name=function_name,
    tool_call_id=tool_call_id,
    arguments=arguments,
)

await processor.handle_function_call(params)

The function call process:

1. LLM requests a function call
2. Processor notifies client with llm-function-call message
3. Client executes function and returns result
4. Result is passed back to LLM via FunctionCallResultFrame
5. Conversation continues

Error Handling

Send error messages to clients:

# General error
await processor.send_error("Invalid configuration")

# Request-specific error
await processor._send_error_response(request_id, "Invalid action arguments")

Error categories:

• Configuration errors
• Action execution errors
• Function call errors
• Protocol errors
• Fatal and non-fatal errors

Bot Control

Manage bot state and handle interruptions:

# Set bot as ready
await processor.set_bot_ready()

# Handle interruptions
await processor.interrupt_bot()

Custom Messaging

Server messages let you push unsolicited data from the server to the client at any time — notifications, status updates, real-time results, etc. They are distinct from server responses, which reply to a specific client request (see Requesting Information from the Server).

Sending server messages

Any FrameProcessor in the pipeline can push an RTVIServerMessageFrame. The RTVIObserver picks it up and delivers it to the client:

from pipecat.processors.frameworks.rtvi import RTVIServerMessageFrame

class MyProcessor(FrameProcessor):
    async def process_frame(self, frame, direction):
        await super().process_frame(frame, direction)
        if isinstance(frame, SomeEventFrame):
            await self.push_frame(
                RTVIServerMessageFrame(
                    data={"type": "event", "value": frame.value}
                )
            )
        await self.push_frame(frame, direction)

RTVIServerMessageFrame is a SystemFrame, so it propagates immediately through the pipeline and is not affected by interruptions or queuing.

Client-side handling

The message arrives at the client with the wire format { label: "rtvi-ai", type: "server-message", data: ... }. Handle it with the onServerMessage callback:

pcClient.onServerMessage((message) => {
  console.log("Server message:", message);
  // message.data contains whatever you passed on the server
});

See Handling Custom Messages from the Server for more details and examples.

UI Agent Protocol

The processor recognizes five first-class RTVI message types for UI Agent traffic — ui-event, ui-snapshot, ui-cancel-task (inbound from client) and ui-command, ui-task (outbound to client). The full wire format is documented in the RTVI standard.

For the recommended higher-level SDK flow, see the UI Agent guide. This page focuses on direct RTVIProcessor integrations.

Receiving UI messages

Inbound ui-event, ui-snapshot, and ui-cancel-task messages fire the on_ui_message event handler with the parsed envelope. The same dispatch also pushes a typed pipeline frame (RTVIUIEventFrame, RTVIUISnapshotFrame, or RTVIUICancelTaskFrame) so processors and observers downstream can react at the pipeline level — pick whichever shape fits your code path.

from pipecat.processors.frameworks.rtvi.models import (
    UIEventMessage, UISnapshotMessage, UICancelTaskMessage,
)

@rtvi.event_handler("on_ui_message")
async def on_ui_message(rtvi, message):
    if isinstance(message, UIEventMessage):
        # message.data.event, message.data.payload
        ...
    elif isinstance(message, UISnapshotMessage):
        # message.data.tree
        ...
    elif isinstance(message, UICancelTaskMessage):
        # message.data.task_id, message.data.reason
        ...

The handler is the canonical extension point for direct integrations. Higher-level frameworks like pipecat-subagents subscribe to it from their UI bridge and route the parsed envelopes onto an agent bus.

Sending UI messages

Outbound ui-command and ui-task messages flow through dedicated pipeline frames. Push RTVIUICommandFrame from any FrameProcessor and the RTVIObserver wraps it into the ui-command envelope on the wire:

from pipecat.processors.frameworks.rtvi import RTVIUICommandFrame
from pipecat.processors.frameworks.rtvi.models import Toast

await self.push_frame(
    RTVIUICommandFrame(
        command="toast",
        payload=Toast(title="Saved").model_dump(),
    )
)

For task-lifecycle traffic, push RTVIUITaskFrame carrying one of the four UITask*Data discriminated-union members (UITaskGroupStartedData, UITaskUpdateData, UITaskCompletedData, UITaskGroupCompletedData). See RTVI Frames for the frame definitions and the RTVI standard for the wire format.