RTVIProcessor

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.

​

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.

​

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