> For clean Markdown of any page, append .md to the page URL.
> For a complete documentation index, see https://docs.nvidia.com/nemo/labs-voice-agent/llms.txt.
> For AI client integration (Claude Code, Cursor, etc.), connect to the MCP server at https://docs.nvidia.com/nemo/labs-voice-agent/_mcp/server.

# nemo_voice_agent.pipecat.processors.frameworks.rtvi_actions

Factory helpers for the custom RTVI messages used by voice-agent bots.

Each factory returns a `(name, handler)` pair. Register them all at once with
`register_client_message_handlers`, which installs a single
`on_client_message` dispatcher on the RTVI processor.

Pipecat 1.0 removed the `RTVIAction` / `RTVIProcessor.register_action()`
API in favor of the client-message / server-response pattern, so what used to
be an "action" is now just a message type. The wire names are unchanged
(`reset`, `update_system_prompt`, `get_context_history`,
`get_scenario_summary`, `apply_initialization`, `apply_sync_delta`) so the
evaluation bridge's vocabulary is stable across the upgrade; the argument
encoding changed from `arguments: [{name, value}]` to a plain `d` object.

Note there is no schema or defaults layer on the 1.x path — every handler here
already reads its arguments as `arguments.get(name, default)`, so nothing is
lost.

The handlers are parameterized so the same factory works for bots with different
pipeline shapes: pass in whichever aggregators, services, and handlers the bot
actually has. `None` entries in `resettable_services` are silently skipped.

`TaskRef` is a tiny holder the bot populates after constructing the pipeline
worker, which happens *after* the RTVI processor (the worker needs `rtvi` in
its observer list). Handlers use it to reach the live worker.

None of these handlers end the pipeline. Before pipecat 1.0 they queued an
`EndTaskFrame`, but that was inert: `queue_frames` injects downstream while
the frame was only handled upstream, so it drifted out of the sink and did
nothing. Pipecat 1.6 also handles it downstream — the sink reflects it upstream
where it becomes an `EndFrame` — which tears down the pipeline, and with it
the WebSocket server that lives inside the input transport. Since `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.

## Module Contents

### Classes

| Name                                                                                            | Description                                              |
| ----------------------------------------------------------------------------------------------- | -------------------------------------------------------- |
| [`SharedStateRef`](#nemo_voice_agent-pipecat-processors-frameworks-rtvi_actions-SharedStateRef) | Mutable handle to the per-scenario `shared_state` dict.  |
| [`TaskRef`](#nemo_voice_agent-pipecat-processors-frameworks-rtvi_actions-TaskRef)               | Mutable handle to a PipelineWorker and its running flag. |

### Functions

| Name                                                                                                                                    | Description                                                                  |
| --------------------------------------------------------------------------------------------------------------------------------------- | ---------------------------------------------------------------------------- |
| [`_is_media_key`](#nemo_voice_agent-pipecat-processors-frameworks-rtvi_actions-_is_media_key)                                           | True if `key` names an audio/image/file container (e.g. `input_audio`,       |
| [`_reset_services`](#nemo_voice_agent-pipecat-processors-frameworks-rtvi_actions-_reset_services)                                       | -                                                                            |
| [`_tag_for_key`](#nemo_voice_agent-pipecat-processors-frameworks-rtvi_actions-_tag_for_key)                                             | Pick a human-readable placeholder tag for a stripped media/binary blob.      |
| [`create_apply_initialization_action`](#nemo_voice_agent-pipecat-processors-frameworks-rtvi_actions-create_apply_initialization_action) | Build the `context.apply_initialization` action.                             |
| [`create_apply_sync_delta_action`](#nemo_voice_agent-pipecat-processors-frameworks-rtvi_actions-create_apply_sync_delta_action)         | Build the `context.apply_sync_delta` action.                                 |
| [`create_get_context_history_action`](#nemo_voice_agent-pipecat-processors-frameworks-rtvi_actions-create_get_context_history_action)   | Build the `context.get_context_history` action.                              |
| [`create_get_scenario_summary_action`](#nemo_voice_agent-pipecat-processors-frameworks-rtvi_actions-create_get_scenario_summary_action) | Build the `context.get_scenario_summary` action.                             |
| [`create_reset_context_action`](#nemo_voice_agent-pipecat-processors-frameworks-rtvi_actions-create_reset_context_action)               | Build the `context.reset` action.                                            |
| [`create_update_system_prompt_action`](#nemo_voice_agent-pipecat-processors-frameworks-rtvi_actions-create_update_system_prompt_action) | Build the `context.update_system_prompt` action.                             |
| [`register_client_message_handlers`](#nemo_voice_agent-pipecat-processors-frameworks-rtvi_actions-register_client_message_handlers)     | Install one `on_client_message` dispatcher for `handlers`.                   |
| [`sanitize_context_for_transport`](#nemo_voice_agent-pipecat-processors-frameworks-rtvi_actions-sanitize_context_for_transport)         | Return a deep copy of an LLM-context structure with raw media blobs dropped. |

### Data

[`ClientMessageHandler`](#nemo_voice_agent-pipecat-processors-frameworks-rtvi_actions-ClientMessageHandler)

[`_BASE64_LEAF_KEYS`](#nemo_voice_agent-pipecat-processors-frameworks-rtvi_actions-_BASE64_LEAF_KEYS)

### API

```python
class nemo_voice_agent.pipecat.processors.frameworks.rtvi_actions.SharedStateRef(
    state: dict = dict()
)
```

Dataclass

Mutable handle to the per-scenario `shared_state` dict.

The same dict that's passed to tool constructors is also published here, so
other RTVI action handlers (specifically `get_scenario_summary`) can read
`shared_state["actions"]` and `shared_state["db"]` without needing tool
references. `state` is reset (re-pointed at a new dict) every time
`update_system_prompt` runs.

**`state`**

---

```python
class nemo_voice_agent.pipecat.processors.frameworks.rtvi_actions.TaskRef(
    task: typing.Optional[pipecat.pipeline.worker.PipelineWorker] = None,
    running: bool = False
)
```

Dataclass

Mutable handle to a PipelineWorker and its running flag.

Construct early, hand to RTVI action factories, then populate once the task
exists. `running` is flipped by the bot runner during shutdown so handlers
can avoid queueing frames onto a dead task.

**`running`**

---

**`task`**

---

```python
nemo_voice_agent.pipecat.processors.frameworks.rtvi_actions._is_media_key(
    key: typing.Any
) -> bool
```

True if `key` names an audio/image/file container (e.g. `input_audio`,
`image_url`, `file`, assistant `audio`).

```python
nemo_voice_agent.pipecat.processors.frameworks.rtvi_actions._reset_services(
    services: typing.List[pipecat.services.ai_service.AIService]
) -> None
```

```python
nemo_voice_agent.pipecat.processors.frameworks.rtvi_actions._tag_for_key(
    key: typing.Optional[str]
) -> str
```

Pick a human-readable placeholder tag for a stripped media/binary blob.

Keyed by the nearest media-container field name so audio/image/file payloads
read naturally in the saved context; anything else → generic `<binary>`.

```python
nemo_voice_agent.pipecat.processors.frameworks.rtvi_actions.create_apply_initialization_action(
    shared_state_ref: nemo_voice_agent.pipecat.processors.frameworks.rtvi_actions.SharedStateRef
) -> nemo_voice_agent.pipecat.processors.frameworks.rtvi_actions.ClientMessageHandler
```

Build the `context.apply_initialization` action.

Bot-side scenario-state initializer. The bridge calls this once per
scenario, immediately after `update_system_prompt`, to populate
the empty `shared_state` dict with everything tools need at call
time. Does three things in order:

1. **Parse** `shared_state_init` JSON into `shared_state_ref.state`
   (merge, not overwrite — preserves bot-side sentinels stashed by
   `update_system_prompt` like `__rtvi__` and `__tool_domain__`).
   Custom keys from `Scenario.setup_shared_state` flow through here.
2. **Load DB**: if the merged state has a `db_path` key, resolve it
   against `EVAL_DATA_ROOT` and replace with the loaded `db` dict.
   Idempotent — skipped when `db` is already present (e.g., when
   the bridge sent an inline DB instead of a path).
3. **Apply init functions**: dispatch each `{func_name, arguments}`
   record against the now-loaded `db`. Side-agnostic — the bridge
   pre-filters by `side` before sending.

Defensive create-if-missing: if `shared_state_ref.state` is
`None` (e.g., in tests that skip `update_system_prompt`), the
handler creates an empty dict and proceeds. In production the bridge
always calls `update_system_prompt` first, which creates + clears
the dict and stashes runtime sentinels.

**Payload shape:**

.. code-block:: json

\{
"domain": "tau2\_telecom",
"shared\_state\_init": "\{"db\_path": "tau2\_telecom/db.json"}",
"actions": \[
\{"side": "user",  "func\_name": "set\_user\_info",
"arguments": \{"name": "John Smith", "phone\_number": "555-..."}},
\{"side": "agent", "func\_name": "enable\_roaming",
"arguments": \{"customer\_id": "C1001", "line\_id": "L1002"}}
]
}

The bridge sends this action per-bot. `shared_state_init` carries
per-side scenario fixture data (path or inline DB, optional initial
actions list, etc.); `actions` carries the per-side filtered subset
of upstream init mutations. Both are independently optional — a
scenario with no init functions and an inline DB still benefits
from a single `apply_initialization` call that does only the
DB-load step.

**Returns** `{"success": bool, "errors": list[str]}`. `success`
is `True` only when DB load and every action dispatch completed
cleanly. The bridge treats any `success=False` as a
framework-class failure and aborts the scenario without scoring it
— partial seeding produces noise, not signal.

```python
nemo_voice_agent.pipecat.processors.frameworks.rtvi_actions.create_apply_sync_delta_action(
    shared_state_ref: nemo_voice_agent.pipecat.processors.frameworks.rtvi_actions.SharedStateRef
) -> nemo_voice_agent.pipecat.processors.frameworks.rtvi_actions.ClientMessageHandler
```

Build the `context.apply_sync_delta` action.

Bot-side endpoint of the cross-side state-propagation pipeline.
The bridge calls this after an `action-applied` event from the
*other* bot — it runs `scenario.sync_state(agent_db, user_db)` on
its in-process shadow DBs, then pushes any non-empty per-side delta
to the corresponding bot via this action.

**Payload shape:**

.. code-block:: json

\{
"domain": "tau2\_telecom",
"delta": \{
"surroundings.payment\_request": \{"bill\_id": "B1002", ...},
"surroundings.line\_active": true
}
}

The bot's handler dispatches via
`nemo_voice_agent.evaluation.sync_appliers.apply_sync_delta` —
looks up the domain-specific applier (falls back to the generic
dotted-path setter) and mutates `shared_state["db"]` in place.

For single-side domains (eva / airline / retail) this action would
never be called by the bridge — only scenarios overriding
`Scenario.sync_state` trigger the pipeline. But the action is
registered domain-agnostically so the same bot binary can run any
scenario.

**Returns** `{"success": bool, "errors": list[str]}`. `success`
is `True` when the delta applied cleanly. Errors are
informational — the bridge logs them and continues (a malformed
delta should not crash the bot or stall the conversation).

```python
nemo_voice_agent.pipecat.processors.frameworks.rtvi_actions.create_get_context_history_action(
    task_ref: nemo_voice_agent.pipecat.processors.frameworks.rtvi_actions.TaskRef,
    assistant_aggregator
) -> nemo_voice_agent.pipecat.processors.frameworks.rtvi_actions.ClientMessageHandler
```

Build the `context.get_context_history` action.

Returns the assistant aggregator's full message list, stringified to match
the shape evaluation clients expect.

**Race-safety against in-flight function calls.** When the bridge fetches
the context immediately after receiving the `<exit>` server message (the
`SendExitMessageTool` / `EndConversationTool` flow), the agent's final
`tool_call` may still be in-flight in the aggregator's pipeline — the
exit message arrives via the `bot_server_message` channel while the
tool\_call frame is still being committed via the
`FunctionCallInProgressFrame` → `FunctionCallResultFrame` cycle. If we
read the context before those frames commit, the captured message list is
missing the final assistant turn. The judge then sees a stale context
that doesn't contain the EndConversationTool tool\_call and (incorrectly)
deducts points for "didn't call EndConversationTool".

Fix: poll the aggregator until its in-progress map drains, with a hard
deadline so a stuck tool can never deadlock scenario cleanup. The common
case (no pending calls) returns immediately — zero added latency.

```python
nemo_voice_agent.pipecat.processors.frameworks.rtvi_actions.create_get_scenario_summary_action(
    task_ref: nemo_voice_agent.pipecat.processors.frameworks.rtvi_actions.TaskRef,
    shared_state_ref: nemo_voice_agent.pipecat.processors.frameworks.rtvi_actions.SharedStateRef
) -> nemo_voice_agent.pipecat.processors.frameworks.rtvi_actions.ClientMessageHandler
```

Build the `context.get_scenario_summary` action.

Returns `{"actions": [...], "db_hash": "<sha>"}` from the per-scenario
shared state by default. With `include_db=true` in the request payload,
also returns the inline `db` dict alongside the hash — used by the
runner's `db_state_assertions` aggregation (predicates need the actual
DB values, not just the hash). Auto-aggregating tools (e.g.
`WriteScenarioTool` subclasses) populate `shared_state["actions"]` on
each successful mutation; the inbound fixture-loading flow populates
`shared_state["db"]`. The bridge calls this action after `<exit>`
(or scenario timeout) to retrieve the final artifacts without depending
on any LLM-callable summary tool.

**One DB per bot.** Each bot's `shared_state["db"]` IS this bot's DB —
the agent bot's `db` is the agent-facing DB; the user bot's `db` is
the user-facing DB. The naming distinction (`db` vs `user_db`) lives
at the bridge/runner boundary: the bridge calls `get_scenario_summary`
once per bot and labels the responses by which WS it pulled from
(the user-side pull is added when the first telecom scenario is
ported). The bot itself doesn't know its own side and doesn't need to.

**Hash-only outbound by default (not inline DB).** The DB itself stays on
the bot server; only the SHA-256 of the canonicalized DB travels through
the WebSocket. This keeps the response payload under a few KB regardless
of DB size (tau2's airline DB is 7 MB inline; serialized via the previous
inline-DB scheme it exceeded pipecat's 1 MB WebSocket frame limit and
closed the connection with code 1009). Both the bot and the runner
import the same `get_dict_hash` from `nemo_voice_agent.evaluation.db_hash`
so the canonical hashing rule (float normalization, order-independent
list fields, excluded keys) is identical on both sides.

**Inline DB opt-in (`include_db=true`).** Telecom's `db_state_assertions`
surface needs the runner to invoke predicate functions on the actual DB
state, not just compare hashes. The bridge sets `include_db=True` when
`scenario.db_state_assertions` is truthy. Telecom's per-bot DBs are
small (\~5 KB MockPhone state on the user side, modest customer/line data
on the agent side) so the WS frame limit is not a concern; for retail
(7 MB DB) the bridge leaves the flag at the default `false` and the
existing hash-out behavior is preserved.

Trade-off (when `include_db=false`): the runner can no longer compute a
per-field `compute_db_diff` on mismatch since it never sees the actual
DB. For debugging hash mismatches in non-telecom domains, set
`include_db=true` temporarily on the bridge call.

Mirrors how `get_context_history` is consumed by the bridge.

```python
nemo_voice_agent.pipecat.processors.frameworks.rtvi_actions.create_reset_context_action(
    task_ref: nemo_voice_agent.pipecat.processors.frameworks.rtvi_actions.TaskRef,
    user_aggregator,
    assistant_aggregator,
    original_messages: typing.List[dict],
    resettable_services: typing.List[pipecat.services.ai_service.AIService]
) -> nemo_voice_agent.pipecat.processors.frameworks.rtvi_actions.ClientMessageHandler
```

Build the `context.reset` action.

`original_messages` is captured by reference so the action always resets to
whatever `update_system_prompt` last wrote.

```python
nemo_voice_agent.pipecat.processors.frameworks.rtvi_actions.create_update_system_prompt_action(
    task_ref: nemo_voice_agent.pipecat.processors.frameworks.rtvi_actions.TaskRef,
    user_aggregator,
    assistant_aggregator,
    original_messages: typing.List[dict],
    resettable_services: typing.List[typing.Any],
    system_role: str,
    system_prompt_suffix: str,
    enable_tool_calling: bool = False,
    llm = None,
    context = None,
    rtvi: typing.Optional[pipecat.processors.frameworks.rtvi.RTVIProcessor] = None,
    tool_factory: typing.Optional[typing.Callable[..., typing.Any]] = None,
    register_schema_tools: typing.Optional[typing.Callable[..., typing.Any]] = None,
    shared_state_ref: typing.Optional[nemo_voice_agent.pipecat.processors.frameworks.rtvi_actions.SharedStateRef] = None
) -> nemo_voice_agent.pipecat.processors.frameworks.rtvi_actions.ClientMessageHandler
```

Build the `context.update_system_prompt` action.

Tool registration is optional. When `enable_tool_calling` is True and a
`tools` JSON string is supplied by the caller, `tool_factory` is invoked
per tool to produce schema tools, then `register_schema_tools` swaps them
onto `llm` / `context`. This keeps the factory decoupled from
evaluation-specific tool registries.

**Scenario-start lifecycle gate.** This handler is the bot-side signal
that a new scenario is starting. In addition to swapping the prompt + tool
surface, it RESETS `shared_state_ref.state` so any prior scenario's
data (`db`, `actions` log, etc.) doesn't bleed into the new one. The
reset is done via `dict.clear()` rather than reassignment so the dict
identity is preserved — any tool that already holds a reference to
`shared_state` continues to see the same object after the clear, and
subsequent mutations by `apply_initialization` propagate correctly.

Scenario fixture data (`db`, `db_path`, `actions` list, custom
keys from `Scenario.setup_shared_state`) is NOT loaded here — that
moves to `create_apply_initialization_action` which the bridge calls
immediately after this one. This handler only stashes bot-side runtime
sentinels (`__rtvi__`, `__tool_domain__`) into the freshly-cleared
dict so write tools have them available at call time.

```python
nemo_voice_agent.pipecat.processors.frameworks.rtvi_actions.register_client_message_handlers(
    rtvi: pipecat.processors.frameworks.rtvi.RTVIProcessor,
    handlers: typing.List[nemo_voice_agent.pipecat.processors.frameworks.rtvi_actions.ClientMessageHandler]
) -> None
```

Install one `on_client_message` dispatcher for `handlers`.

This is the pipecat 1.x replacement for calling
`rtvi.register_action(...)` once per action. Handler return values are
sent back as the `d` payload of a `server-response` keyed to the
request's message id; an unknown type or a raised exception produces an
`error-response` instead, so callers fail loudly rather than blocking
until their read timeout expires.

```python
nemo_voice_agent.pipecat.processors.frameworks.rtvi_actions.sanitize_context_for_transport(
    obj: typing.Any,
    _media_key: typing.Optional[str] = None
) -> typing.Any
```

Return a deep copy of an LLM-context structure with raw media blobs dropped.

Omni models keep audio (and, in future, image/file) payloads inline in their
message list. Serializing those verbatim produces a multi-MB payload that
overflows pipecat's WebSocket frame cap, so `get_context_history` silently
fails and the saved agent context ends up empty.

Only the **raw encoded data** is dropped (replaced with an `<audio>` /
`<image>` / `<file>` / `<binary>` tag) — small metadata is kept so the
snapshot stays useful for debugging:

* dropped: `bytes` / `bytearray` values; `data:` base64 URI strings
  (e.g. `image_url.url`); `file_data` / `b64_json` leaves; and the
  `data` leaf when nested inside a media container (`input_audio.data`,
  assistant `audio.data`).
* kept: `text` / `refusal` strings, `format`, `filename`, `file_id`,
  `detail`, audio `transcript`, `id`, and plain `url` / file-path
  strings (only base64 `data:` URIs are stripped, not real URLs/paths).

`_media_key` carries the nearest enclosing media-container key name down the
recursion so a stripped leaf gets the right tag. Pure and non-mutating — the
caller's live context is never altered; a fresh structure is returned.

```python
nemo_voice_agent.pipecat.processors.frameworks.rtvi_actions.ClientMessageHandler = tuple[str, Callable[[RTVIProcessor, dict[str, Any]], Any]]
```

```python
nemo_voice_agent.pipecat.processors.frameworks.rtvi_actions._BASE64_LEAF_KEYS = frozenset({'file_data', 'b64_json'})
```