The RTVIProcessor is the main coordinator for RTVI protocol communication between clients and your Pipecat application. It manages service configuration, handles client actions, and coordinates function calls.

​
Basic Usage

The RTVIProcessor needs to be initialized with a configuration and added to your pipeline. The configuration defines what services and configuration options are available to clients.

from pipecat.processors.rtvi import RTVIProcessor, RTVIConfig, RTVIServiceConfig

# Create initial configuration
initial_config = RTVIConfig(
    config=[
        RTVIServiceConfig(
            service="voice",
            options=[
                RTVIServiceOptionConfig(
                    name="voice_id",
                    value="default"
                )
            ]
        )
    ]
)

# Initialize processor
rtvi = RTVIProcessor(config=initial_config)

# Add to pipeline
pipeline = Pipeline([
    transport.input(),
    rtvi,
    # ... other processors ...
    transport.output()
])

​
Bot Ready State

The processor manages the bot’s ready state. The bot must be marked as ready before it can receive client messages:

await processor.set_bot_ready()

When the bot is marked ready, it sends a bot-ready message to the client containing:

  • The RTVI protocol version
  • Current service configuration
  • Available actions

​
Client Ready State

Clients must also indicate they are ready to receive messages:

await processor.set_client_ready()

After both bot and client are ready:

  • Services can be configured
  • Actions can be executed
  • Function calls can be processed

​
Services

Services represent configurable components of your application. Each service can have multiple options that clients can control.

​
Registering Services

To make a service available to clients, you need to:

  1. Define the option handler function(s)
  2. Create an RTVIService instance
  3. Register it with the RTVIProcessor using register_service()
# 1. Define option handler
async def handle_voice_option(processor, service, option):
    voice_id = option.value
    # Update voice configuration...
    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_processor = RTVIProcessor(config=RTVIConfig(config=[]))
rtvi_processor.register_service(voice_service)

​
Defining Services

Create a service with configurable options:

from pipecat.processors.rtvi import RTVIService, RTVIServiceOption

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

The service handler is called whenever the option changes:

  • Receives the processor instance
  • Gets the service name
  • Contains the new option configuration
  • Can validate and apply changes
  • Can send errors if needed

​
Option Types

Services support multiple option types for different configuration needs:

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

Available option types:

  • bool - True/false flags
  • number - Numeric values
  • string - Text values
  • array - Lists of values
  • object - Complex structured data

​
Option Handlers

Option handlers process configuration changes. A handler receives:

  • The processor instance
  • The service name
  • The option configuration with new value
async def handle_voice_option(processor, service, option):
    voice_id = option.value
    # Update voice configuration...

​
Actions

Actions are functions that clients can trigger. They can accept arguments and return results.

Learn how to use RTVI Actions in the Client SDK documentation.

​
Registering Actions

To make an action available to clients, you need to:

  1. Define the action handler function
  2. Create an RTVIAction instance
  3. Register it with the RTVIProcessor using register_action()
# 1. Define handler function
async def handle_print_message(processor, service, arguments):
    """Handler for print message action"""
    message = arguments.get("message", "Default message")
    logger.info(f"Print action triggered with message: {message}")
    return True

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

# 3. Register with processor
rtvi_processor = RTVIProcessor(config=RTVIConfig(config=[]))
rtvi_processor.register_action(print_action)

​
Basic Actions

Register a simple action:

reset_action = RTVIAction(
    service="conversation",
    action="reset",
    result="bool",
    handler=handle_reset
)

The action handler receives:

  • The processor instance
  • The service name
  • Any provided arguments

​
Actions with 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,
    llm=llm,
    context=context,
    result_callback=result_callback
)

The function call process:

  1. LLM requests a function call
  2. Processor notifies client
  3. Client executes function
  4. Result is returned to LLM
  5. Conversation continues

​
Error Handling

Send error messages to clients in different ways:

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

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

Error types include:

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

​
Bot Control

Control bot state and handle interruptions:

# Start the bot
await processor.set_bot_ready()

# Handle interruptions
await processor.interrupt_bot()

Bot control features:

  • Ready state management
  • Client synchronization
  • Graceful interruption
  • Error recovery

​
Next Steps

Frame Processors

Learn about the specialized processors that work alongside RTVIProcessor
OverviewRTVI Frame Processors