RTVIProcessor

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 PipelineWorker. Access it via worker.rtvi:

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

worker = PipelineWorker(pipeline)

# Access the processor
rtvi = worker.rtvi

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

from pipecat.services.google.rtvi import GoogleRTVIProcessor

worker = PipelineWorker(pipeline, rtvi_processor=GoogleRTVIProcessor())

To disable RTVI entirely, set enable_rtvi=False.

Properties

client_version

Returns the negotiated client protocol version as [major, minor, patch]. Defaults to [0, 0, 0] until the client sends a client-ready message.

version = rtvi.client_version
if version[0] == 2:
    # Client is using protocol v2
    pass

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 worker.queue_frames([...])

The processor validates the client’s RTVI protocol version during the handshake. The server protocol version is 2.1.0. Clients with major version 2 are fully compatible. Clients with major version 1 (deprecated) are still supported with the legacy bot-output format. 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:

LLM requests a function call
Processor notifies client with llm-function-call message
Client executes function and returns result
Result is passed back to LLM via FunctionCallResultFrame
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 uses the high-priority SystemFrame lane, remains ordered with other SystemFrames, and is not discarded by interruptions.

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.

DTMF Input

The processor handles dtmf client messages for keypad input (e.g., phone menu navigation). Each dtmf message carries one or more keypad entries (0-9, *, #) as a list. The processor pushes one InputDTMFFrame per key, in order, downstream to the bot’s DTMF handling (e.g., a DTMFAggregator) exactly as a telephony transport’s rapid keypresses would arrive.

# Client message format (JSON) — protocol 2.1.0
{
  "label": "rtvi-ai",
  "type": "dtmf",
  "id": "msg-123",
  "data": {
    "buttons": ["1", "2", "#"]
  }
}

When a bot uses a DTMFAggregator, the aggregator accumulates keys and flushes them (on # or its idle timeout) into a transcription like "DTMF: 123#" that the LLM can react to.

UI Messaging

The processor handles the User Interface messages that let a server-side agent observe and drive the client GUI. Inbound UI messages from the client are turned into pipeline frames and surfaced via the on_ui_message event handler:

Wire message	Frame pushed downstream	`on_ui_message` payload
`ui-event`	`RTVIUIEventFrame`	`UIEventMessage`
`ui-snapshot`	`RTVIUISnapshotFrame`	`UISnapshotMessage`
`ui-cancel-job-group`	`RTVIUICancelJobGroupFrame`	`UICancelJobGroupMessage`

@rtvi.event_handler("on_ui_message")
async def on_ui_message(processor, message):
    # message is a UIEventMessage / UISnapshotMessage / UICancelJobGroupMessage
    ...

When you use UIWorker, PipelineWorker wires this up automatically (RTVI is enabled by default): inbound UI messages are republished onto the bus for the worker, and outbound ui-command / ui-job-group frames are translated by the RTVI Observer. You only handle on_ui_message directly when targeting the wire format from a single-LLM app without a UIWorker.

Pipecat Server

Client SDKs

Pipecat Flows

Pipecat Cloud

CLI

Pipecat Context Hub

Initialization

Properties

client_version

Readiness Protocol

Client Ready State

Bot Ready State

Services

Registering Services

Option Types

Actions

Registering Actions

Action Arguments

Function Calls

Error Handling

Bot Control

Custom Messaging

Sending server messages

Client-side handling

DTMF Input

UI Messaging

​Initialization

​Properties

​client_version

​Readiness Protocol

​Client Ready State

​Bot Ready State

​Services

​Registering Services

​Option Types

​Actions

​Registering Actions

​Action Arguments

​Function Calls

​Error Handling

​Bot Control

​Custom Messaging

​Sending server messages

​Client-side handling

​DTMF Input

​UI Messaging

Initialization

Properties

client_version

Readiness Protocol

Client Ready State

Bot Ready State

Services

Registering Services

Option Types

Actions

Registering Actions

Action Arguments

Function Calls

Error Handling

Bot Control

Custom Messaging

Sending server messages

Client-side handling

DTMF Input

UI Messaging