RTVI Control Plane

Every NeMo Labs Voice Agent bot embeds a Pipecat RTVIProcessor in its pipeline. Alongside the audio stream, a connected client can send client messages over the same WebSocket to inspect or mutate bot state. The client can clear the conversation, swap the system prompt and tool surface, snapshot the large language model (LLM) context, or seed scenario fixtures. These operations form the control plane.

The handlers live in nemo_voice_agent/pipecat/processors/frameworks/rtvi_actions.py. Each is produced by a create_*_action factory that returns a (wire_name, handler) pair. The bot installs them all with a single call to register_client_message_handlers. For the wire envelope (client-message / server-response / error-response and the t / d argument encoding), refer to RTVI Message Reference.

Pipecat 1.0 removed the RTVIAction and RTVIProcessor.register_action() API. The wire names below survive from that era — the code still calls them “actions” — but they are now plain client-message types.

The Six Handlers

The shipped handlers cover runtime reset, prompt updates, state initialization and synchronization, and result retrieval.

examples/generic_voice_agent/server/server.py registers only reset — that is all the browser client needs. evaluation/bot_server.py registers all six, because the evaluation bridge drives the full scenario lifecycle over this channel.

reset

Takes no arguments. Resets the user and assistant aggregators and re-seeds both with a deep copy of original_messages. The handler captures original_messages by reference, so the deep copy reflects the last value that update_system_prompt wrote. It calls .reset() on each resettable_services entry that defines one and skips None entries. The browser client connects this action to its Reset button using sendClientRequest('reset', {}).

update_system_prompt

Arguments: prompt (required), tools (JSON string), add_suffix (default true), tool_domain (default "default").

Replaces the system message, optionally appends the configured system-prompt suffix, and then resets both aggregators. It is also the scenario-start gate. It clears shared_state in place using dict.clear(), so tools holding a reference continue to use the same object. The handler then stashes two bot-side runtime sentinels: __rtvi__ and __tool_domain__. Write tools use __rtvi__ to emit action-applied messages. When tool calling is enabled and a tools payload is present, the injected tool_factory builds each entry. register_schema_tools then replaces the previous LLM tool set with those entries.

It does not load scenario fixture data. db_path, DB contents, and custom shared-state keys all arrive through apply_initialization.

get_context_history

No arguments. Returns the assistant aggregator’s message list, stringified, as context. Before creating the snapshot, it polls has_function_calls_in_progress until in-flight function calls commit. A hard 3-second deadline turns a stuck tool into a warning instead of a deadlock. The snapshot then passes through sanitize_context_for_transport, which replaces raw audio, image, and file blobs with placeholder tags. Omni models keep audio inline, and serializing it verbatim overflows Pipecat’s WebSocket frame cap. Sanitization is non-mutating, so the live context keeps its bytes.

get_scenario_summary

Argument: include_db (default false).

Returns actions, the auto-aggregated write-tool records from shared_state, and db_hash. The hash is the SHA-256 of the canonicalized DB that get_dict_hash from nemo_voice_agent.evaluation.db_hash computes. The DB stays on the bot server. The runner computes the same hash from its in-process gold replay and compares strings, keeping the payload small. Setting include_db to true adds the inline db dict. The bridge requests it only for domains whose per-predicate db_state_assertions need real values and whose DB fits within the frame limit.

Each bot returns only its own DB. The db versus user_db distinction is applied bridge-side, based on which WebSocket the response came from.

apply_initialization

Arguments: domain, shared_state_init (JSON string), actions (list).

The single scenario-state initializer is called for each bot immediately after update_system_prompt. It does three things in order:

1. Merges the decoded shared_state_init into shared_state, preserving the runtime sentinels.
2. If the merged state carries a db_path, resolves it under the eval data root and loads it into db. Idempotent — skipped when db is already present, and the redundant db_path key is dropped either way.
3. Dispatches each func_name and arguments record in actions against the loaded db, using the registry in nemo_voice_agent/evaluation/initialization_functions.py.

Returns success and errors. The bridge calls it for every scenario even when there are no init actions, because steps 1 and 2 must still run. Any success: false aborts the scenario rather than scoring partially seeded state.

apply_sync_delta

Arguments: domain, delta (dict).

Bot-side endpoint of the cross-side state-sync pipeline. After a write tool emits action-applied, the bridge replays the action onto its shadow DBs and runs the scenario’s sync_state. It pushes any non-empty per-side delta to this endpoint. The handler dispatches through the per-domain applier registry in nemo_voice_agent/evaluation/sync_appliers.py, which mutates shared_state["db"] in place. Unregistered domains use a generic dotted-path setter. The returned success and errors values are informational, so a malformed delta does not stall the conversation. Only scenarios that override Scenario.sync_state trigger the handler. Currently, telecom scenarios use it, but the registration is domain-agnostic.

register_client_message_handlers

Register message types with the RTVIProcessor before the pipeline starts processing client messages.

1
from nemo_voice_agent.pipecat.processors.frameworks.rtvi_actions import (
2
    TaskRef,
3
    create_reset_context_action,
4
    register_client_message_handlers,
5
)
6
7
task_ref = TaskRef()
8
register_client_message_handlers(
9
    rtvi,
10
    [create_reset_context_action(task_ref, user_agg, assistant_agg, original_messages, resettable)],
11
)

The function builds a dispatch table from the (name, handler) pairs and installs one on_client_message event handler on the processor. A handler’s return value becomes the d payload of the server-response keyed to the request id. An unknown message type or a raised exception produces an error-response instead, so a caller fails loudly rather than blocking until its read timeout expires.

Import path matters. The nemo_voice_agent.pipecat.processors.frameworks package re-exports a subset of the module: TaskRef, register_client_message_handlers, sanitize_context_for_transport, and the reset / update_system_prompt / get_context_history factories. SharedStateRef, ClientMessageHandler, and the three evaluation-specific factories must be imported from the rtvi_actions module directly.

TaskRef and SharedStateRef

Both are small mutable dataclasses that exist to break a construction-order cycle — the handlers are built before the objects they need.

TaskRef exists because the PipelineWorker cannot be constructed until after the RTVI processor. The worker takes rtvi in its observer list. Create the ref early and give it to the factories. run_bot_websocket_server sets task and flips running to true. The runner flips it back during shutdown, so a handler can avoid queuing frames onto a stopped task.

SharedStateRef.state is the same dict passed to tool constructors, which is why handlers can read actions and db without holding tool references. Its identity is never reassigned, so tools registered for an earlier scenario stay valid.

Adding Your Own Handler

Write a factory that closes over whatever pipeline objects it needs and returns a (name, coroutine) pair, then append it to the list you pass to register_client_message_handlers:

1
from typing import Any
2
3
from pipecat.processors.frameworks.rtvi import RTVIProcessor
4
5
from nemo_voice_agent.pipecat.processors.frameworks.rtvi_actions import ClientMessageHandler, TaskRef
6
7
8
def create_ping_action(task_ref: TaskRef) -> ClientMessageHandler:
9
    async def handler(rtvi_processor: RTVIProcessor, arguments: dict[str, Any]) -> dict:
10
        return {
11
            "echo": arguments.get("payload", ""),
12
            "running": task_ref.running,
13
        }
14
15
    return ("ping", handler)

Two rules:

• Read arguments defensively. There is no schema or defaults layer on this path. Every shipped handler uses arguments.get(name, default).
• Never end the pipeline from a handler. The WebSocket server lives inside the input transport, so tearing down the pipeline also kills the server and nothing can reconnect. reset and update_system_prompt run at the start of every evaluation scenario. Ending the pipeline there would kill the bot before its first turn.

If you need bot-to-client notifications instead of request/response, push an RTVI.ServerMessage with rtvi.push_transport_message(...) — that is how write tools emit action-applied.

Related Pages

Use these pages for wire-level message fields and end-to-end client connection behavior:

• RTVI Message Reference — wire format and per-message payloads
• Client Protocol — connecting a client to the bot
• Custom Pipeline — assembling a bot that registers its own handlers