Messages

All bus message types for agent communication

Overview

Bus messages are the communication primitives for the multi-agent framework. They are organized into two priority levels:

• Data messages (BusDataMessage): Normal-priority messages for routine communication.
• System messages (BusSystemMessage): High-priority messages that preempt normal messages in subscriber queues (e.g. cancel, errors).

Some messages also implement BusLocalMessage, meaning they stay on the local bus and are never forwarded to remote buses.

from pipecat_subagents.bus import (
    BusActivateAgentMessage,
    BusDeactivateAgentMessage,
    BusEndMessage,
    BusCancelMessage,
    # ... etc
)

Base Types

Type	Parent	Description
BusMessage	-	Mixin carrying source and target metadata
BusLocalMessage	-	Mixin: message stays on the local bus only
BusDataMessage	BusMessage, DataFrame	Normal-priority bus message
BusSystemMessage	BusMessage, SystemFrame	High-priority bus message

Common Fields

All messages inherit these fields from BusDataMessage or BusSystemMessage:

| Field | Type | Default | Description | | -------- | ----- | ------- | ----------------------------------------------------- | ------------------------------------------------------- | | source | str | | Name of the agent or component that sent this message | | target | str | None | None | Name of the intended recipient, or None for broadcast |

Frame Transport

Type	Priority	Description
BusFrameMessage	Data	Wraps a Pipecat Frame for transport over the bus

BusFrameMessage

| Field | Type | Default | Description | | ----------- | ---------------- | ------- | ------------------------------------------------------------- | ---------------------------------------------- | | source | str | | Sending agent name | | target | str | None | None | Recipient agent name | | frame | Frame | | The Pipecat frame to transport | | direction | FrameDirection | | Direction the frame should travel in the recipient’s pipeline | | bridge | str | None | None | Bridge name for routing in multi-bridge setups |

Agent Lifecycle

Type	Priority	Local	Description
BusActivateAgentMessage	Data	No	Tells a targeted agent to become active
BusDeactivateAgentMessage	Data	No	Tells a targeted agent to become inactive
BusEndMessage	Data	No	Request a graceful end of the session
BusEndAgentMessage	Data	No	Tells a targeted agent to end its pipeline
BusCancelMessage	System	No	Request a hard cancel of the session
BusCancelAgentMessage	System	No	Tells a targeted agent to cancel its pipeline

BusActivateAgentMessage

| Field | Type | Default | Description | | -------- | ----- | ------- | ------------------ | ------------------------------------------------ | | source | str | | Sending agent name | | target | str | None | None | Agent to activate | | args | dict | None | None | Activation arguments forwarded to on_activated |

BusEndMessage / BusEndAgentMessage

| Field | Type | Default | Description | | -------- | ----- | ------- | ------------------ | -------------------------------- | | source | str | | Sending agent name | | target | str | None | None | Recipient agent name | | reason | str | None | None | Human-readable reason for ending |

BusCancelMessage / BusCancelAgentMessage

| Field | Type | Default | Description | | -------- | ----- | ------- | ------------------ | -------------------------------------- | | source | str | | Sending agent name | | target | str | None | None | Recipient agent name | | reason | str | None | None | Human-readable reason for cancellation |

Registry & Errors

Type	Priority	Local	Description
BusAddAgentMessage	System	Yes	Request to add an agent to the local runner
BusAgentRegistryMessage	System	No	Snapshot of agents managed by a runner
BusAgentReadyMessage	Data	No	Announces that an agent is ready
BusAgentErrorMessage	System	No	Reports an error from a root agent
BusAgentLocalErrorMessage	System	Yes	Reports an error from a child agent to its parent

BusAgentRegistryMessage

Field	Type	Default	Description
source	str		Sending runner name
runner	str		Name of the runner that owns these agents
agents	list[AgentRegistryEntry]		List of agent entries with their state

BusAgentReadyMessage

| Field | Type | Default | Description | | ------------ | ------ | ------- | -------------------------------------- | --------------------------------------------------- | | source | str | | Agent name | | runner | str | | Name of the runner managing this agent | | parent | str | None | None | Name of the parent agent, or None for root agents | | active | bool | False | Whether the agent started active | | bridged | bool | False | Whether the agent is bridged | | started_at | float | None | None | Unix timestamp when the agent became ready |

BusAgentErrorMessage / BusAgentLocalErrorMessage

Field	Type	Description
source	str	Agent name
error	str	Description of the error

Task Messages

Type	Priority	Description
BusTaskRequestMessage	Data	Requests a task agent to start work
BusTaskResponseMessage	Data	Response from a task agent
BusTaskResponseUrgentMessage	System	High-priority response from a task agent
BusTaskUpdateMessage	Data	Progress update from a task agent
BusTaskUpdateUrgentMessage	System	High-priority progress update
BusTaskUpdateRequestMessage	Data	Request a progress update from a task agent
BusTaskCancelMessage	System	Cancel a running task

BusTaskRequestMessage

| Field | Type | Default | Description | | ----------- | ----- | ------- | ---------------------- | --------------------------------------- | | source | str | | Requester agent name | | target | str | None | None | Worker agent name | | task_id | str | | Unique task identifier | | task_name | str | None | None | Task name for routing to named handlers | | payload | dict | None | None | Structured data describing the work |

BusTaskResponseMessage / BusTaskResponseUrgentMessage

| Field | Type | Default | Description | | ---------- | ----------------------------------------------------------------- | ------- | ------------------- | -------------------- | | source | str | | Worker agent name | | target | str | None | None | Requester agent name | | task_id | str | | The task identifier | | status | TaskStatus | | Completion status | | response | dict | None | None | Result data |

BusTaskUpdateMessage / BusTaskUpdateUrgentMessage

| Field | Type | Default | Description | | --------- | ----- | ------- | ------------------- | -------------------- | | source | str | | Worker agent name | | target | str | None | None | Requester agent name | | task_id | str | | The task identifier | | update | dict | None | None | Progress data |

BusTaskCancelMessage

| Field | Type | Default | Description | | --------- | ----- | ------- | -------------------- | -------------------------------------- | | source | str | | Requester agent name | | target | str | None | None | Worker agent name | | task_id | str | | The task identifier | | reason | str | None | None | Human-readable reason for cancellation |

Task Streaming

Type	Priority	Description
BusTaskStreamStartMessage	Data	Signals the start of a streaming task response
BusTaskStreamDataMessage	Data	A chunk of streaming task data
BusTaskStreamEndMessage	Data	Signals the end of a streaming task response

BusTaskStreamStartMessage / BusTaskStreamDataMessage / BusTaskStreamEndMessage

| Field | Type | Default | Description | | --------- | ----- | ------- | ------------------- | ---------------------------------------------------------------------- | | source | str | | Worker agent name | | target | str | None | None | Requester agent name | | task_id | str | | The task identifier | | data | dict | None | None | Stream metadata (start), chunk payload (data), or final metadata (end) |

UI Messages

Type	Priority	Description
BusUIEventMessage	Data	UI event from the client to a server-side UIAgent
BusUICommandMessage	Data	UI command from a UIAgent to the client
BusUITaskGroupStartedMessage	Data	A user-facing task group has been dispatched (forwarded to the client as ui-task group_started)
BusUITaskUpdateMessage	Data	Per-worker progress (ui-task task_update)
BusUITaskCompletedMessage	Data	A worker in a user-facing task group has completed (ui-task task_completed)
BusUITaskGroupCompletedMessage	Data	A user-facing task group has finished (ui-task group_completed)

The bus messages are subagents-internal carriers that the bridge installed by attach_ui_bridge translates to/from the on-the-wire RTVI types. They live in pipecat_subagents.agents.ui.ui_messages.

BusUIEventMessage

Emitted by attach_ui_bridge when the client dispatches an event via PipecatClient.sendUIEvent(event, payload). UIAgent subclasses dispatch these to @on_ui_event(name) handlers.

| Field | Type | Default | Description | | ------------ | ----- | ------- | -------------------------------- | -------------------------------------- | | source | str | | Sending component (the bridge) | | target | str | None | None | Recipient agent name (None broadcasts) | | event_name | str | "" | App-defined event name | | payload | Any | None | App-defined payload. Schemaless. |

BusUICommandMessage

Published by UIAgent.send_command. The bridge installed by attach_ui_bridge turns this into an RTVIUICommandFrame on the root agent’s pipeline; the RTVIObserver wraps the frame into a ui-command envelope on the wire and the client’s PipecatClient emits RTVIEvent.UICommand and invokes the onUICommand callback, where apps can dispatch by command name.

| Field | Type | Default | Description | | -------------- | ----- | ------- | --------------------------------------------------------------- | -------------------------------------- | | source | str | | Sending agent name | | target | str | None | None | Recipient agent name (None broadcasts) | | command_name | str | "" | App-defined command name | | payload | Any | None | Plain dict by the time it lands on the bus (see send_command) |

BusUITaskGroupStartedMessage

Published by UIAgent.user_task_group(...) on entry. The bridge forwards it to the client as a ui-task envelope with kind = "group_started".

Field	Type	Default	Description
task_id	str	""	Shared task identifier for the group.
agents	list[str] | None	None	Names of the agents the work was dispatched to.
label	str | None	None	Optional human-readable label for the group.
cancellable	bool	True	Whether the client may request cancellation.
at	int	0	Epoch milliseconds when the group started.

BusUITaskUpdateMessage

Forwarded by UIAgent whenever a worker emits a BusTaskUpdateMessage whose task_id matches a registered user task group. The bridge forwards to the client as a ui-task envelope with kind = "task_update".

Field	Type	Default	Description
task_id	str	""	The shared task identifier.
agent_name	str	""	The worker that produced the update.
data	Any	None	The worker’s update payload, forwarded verbatim.
at	int	0	Epoch milliseconds when the update was emitted.

BusUITaskCompletedMessage

Forwarded by UIAgent whenever a worker’s BusTaskResponseMessage arrives for a registered user task group. The bridge forwards to the client as a ui-task envelope with kind = "task_completed".

Field	Type	Default	Description
task_id	str	""	The shared task identifier.
agent_name	str	""	The worker that produced the response.
status	str	""	Completion status as a string (TaskStatus value).
response	Any	None	The worker’s response payload.
at	int	0	Epoch milliseconds when the response was received.

BusUITaskGroupCompletedMessage

Published when UIAgent.user_task_group(...) exits, after every worker has responded (or the group has been cancelled). The bridge forwards to the client as a ui-task envelope with kind = "group_completed".

Field	Type	Default	Description
task_id	str	""	The shared task identifier.
at	int	0	Epoch milliseconds when the group completed.

Wire-format type strings

The on-the-wire type strings ("ui-event", "ui-command", "ui-snapshot", "ui-cancel-task", "ui-task") live exclusively as Literal[...] defaults on the matching pydantic envelope models in pipecat.processors.frameworks.rtvi.models. There are no standalone UI_*_MESSAGE_TYPE constants. Client-side, the strings are members of the RTVIMessageType enum (e.g. RTVIMessageType.UI_EVENT).

attach_ui_bridge

def attach_ui_bridge(agent: BaseAgent, *, target: str | None = None) -> None

Wire the root agent’s pipeline to the UI Agent Protocol. Call this from the root agent’s on_ready hook. The root agent’s pipeline task must be built with enable_rtvi=True.

from pipecat_subagents.agents import attach_ui_bridge

class RootAgent(BaseAgent):
    async def on_ready(self) -> None:
        await super().on_ready()
        attach_ui_bridge(self, target="ui")

Side effects:

• Inbound: subscribes to the RTVIProcessor’s on_ui_message handler. Every ui-event and ui-snapshot message is republished onto the bus as a BusUIEventMessage (snapshots are routed to UIAgent via a reserved internal event name; ui-cancel-task is routed similarly to cancel_task).
• Outbound: subscribes to the bus. When a BusUICommandMessage arrives, the bridge pushes an RTVIUICommandFrame downstream; when one of the BusUITask* messages arrives, it pushes an RTVIUITaskFrame carrying the matching UITask*Data. The RTVIObserver wraps each frame into the ui-command / ui-task envelope on the wire.

| Parameter | Type | Default | Description | | --------- | ----------- | ------- | ----------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | | agent | BaseAgent | | The root agent owning the pipeline task and RTVI processor. | | target | str | None | None | Optional target agent name for republished BusUIEventMessage. When None, the message is broadcast to every UIAgent on the bus. Pass the UI agent’s name (e.g. "ui") for direct delivery. |

Raises: RuntimeError if the agent’s pipeline task has no RTVI processor.