Messages - Pipecat

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.bus import (
    BusActivateWorkerMessage,
    BusDeactivateWorkerMessage,
    BusEndMessage,
    BusCancelMessage,
    # ... etc
)

Base Types

Type	Parent	Description
`BusMessage`	-	Base class carrying `source` and `target` metadata
`BusLocalMessage`	-	Mixin: message stays on the local bus only
`BusDataMessage`	`BusMessage`	Normal-priority bus message (FIFO)
`BusSystemMessage`	`BusMessage`	High-priority bus message

Common Fields

All messages inherit these fields from BusMessage:

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

Pipeline Commands

Type	Priority	Description
`BusTTSSpeakMessage`	Data	Asks a `PipelineWorker` to speak text via its TTS

BusTTSSpeakMessage

Field	Type	Default	Description
`source`	`str`		Sending agent name
`target`	`str \| None`	`None`	Recipient agent name
`text`	`str`		The text to be spoken
`append_to_context`	`bool \| None`	`None`	Whether the spoken text should be appended to the context

Agent Lifecycle

Type	Priority	Local	Description
`BusActivateWorkerMessage`	Data	No	Tells a targeted agent to become active
`BusDeactivateWorkerMessage`	Data	No	Tells a targeted agent to become inactive
`BusEndMessage`	Data	No	Request a graceful end of the session
`BusEndWorkerMessage`	Data	No	Tells a targeted agent to end its pipeline
`BusCancelMessage`	System	No	Request a hard cancel of the session
`BusCancelWorkerMessage`	System	No	Tells a targeted agent to cancel its pipeline

BusActivateWorkerMessage

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 / BusEndWorkerMessage

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 / BusCancelWorkerMessage

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
`BusAddWorkerMessage`	System	Yes	Request to add an agent to the local runner
`BusWorkerRegistryMessage`	System	No	Snapshot of agents managed by a runner
`BusWorkerReadyMessage`	Data	No	Announces that an agent is ready
`BusWorkerErrorMessage`	System	No	Reports an error from a root agent
`BusWorkerLocalErrorMessage`	System	Yes	Reports an error from a child agent to its parent

BusAddWorkerMessage

Field	Type	Default	Description
`source`	`str`		Sending agent name
`target`	`str \| None`	`None`	Recipient name
`worker`	`BaseWorker`		The agent instance to add (in-memory)

BusWorkerRegistryMessage

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

BusWorkerReadyMessage

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

BusWorkerErrorMessage / BusWorkerLocalErrorMessage

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

Job Messages

Type	Priority	Description
`BusJobRequestMessage`	Data	Requests a worker agent to start work
`BusJobResponseMessage`	Data	Response from a worker agent
`BusJobResponseUrgentMessage`	System	High-priority response from a worker agent
`BusJobUpdateMessage`	Data	Progress update from a worker agent
`BusJobUpdateUrgentMessage`	System	High-priority progress update
`BusJobUpdateRequestMessage`	Data	Request a progress update from a worker agent
`BusJobCancelMessage`	System	Cancel a running job

BusJobRequestMessage

Field	Type	Default	Description
`source`	`str`		Requester agent name
`target`	`str \| None`	`None`	Worker agent name
`job_id`	`str`		Unique job identifier
`job_name`	`str \| None`	`None`	Job name for routing to named `@job` handlers
`payload`	`dict \| None`	`None`	Structured data describing the work

BusJobResponseMessage / BusJobResponseUrgentMessage

Field	Type	Default	Description
`source`	`str`		Worker agent name
`target`	`str \| None`	`None`	Requester agent name
`job_id`	`str`		The job identifier
`status`	`JobStatus`		Completion status
`response`	`dict \| None`	`None`	Result data

BusJobUpdateMessage / BusJobUpdateUrgentMessage

Field	Type	Default	Description
`source`	`str`		Worker agent name
`target`	`str \| None`	`None`	Requester agent name
`job_id`	`str`		The job identifier
`update`	`dict \| None`	`None`	Progress data

BusJobUpdateRequestMessage

Field	Type	Default	Description
`source`	`str`		Requester agent name
`target`	`str \| None`	`None`	Worker agent name
`job_id`	`str`		The job identifier

BusJobCancelMessage

Field	Type	Default	Description
`source`	`str`		Requester agent name
`target`	`str \| None`	`None`	Worker agent name
`job_id`	`str`		The job identifier
`reason`	`str \| None`	`None`	Human-readable reason for cancellation

Job Streaming

Type	Priority	Description
`BusJobStreamStartMessage`	Data	Signals the start of a streaming job response
`BusJobStreamDataMessage`	Data	A chunk of streaming job data
`BusJobStreamEndMessage`	Data	Signals the end of a streaming job response

BusJobStreamStartMessage / BusJobStreamDataMessage / BusJobStreamEndMessage

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

UI Messages

These carriers are the on-the-bus shape of the UI Worker protocol, exchanged between a UIWorker and the PipelineWorker that owns the transport. They are not the on-the-wire format the client sees — that is the RTVI User Interface protocol. When RTVI is enabled, PipelineWorker translates between the two: its on_ui_message handler republishes inbound client messages onto the bus, and its on_bus_message handler turns outbound carriers into RTVI frames. They live in the bus layer (rather than alongside UIWorker) because both PipelineWorker (in pipecat.pipeline) and UIWorker (in pipecat.workers) reference them, and pipeline must not import from workers. All subclass BusUIDataMessage.

Type	Direction	Description
`BusUIEventMessage`	client → worker	A UI event from the client; dispatched to `@ui_event(name)` handlers
`BusUICommandMessage`	worker → client	A UI command; translated to `RTVIUICommandFrame`
`BusUIJobGroupStartedMessage`	worker → client	A user-facing job group was dispatched (`kind = "group_started"`)
`BusUIJobUpdateMessage`	worker → client	Per-worker progress for a user-facing job group (`kind = "job_update"`)
`BusUIJobCompletedMessage`	worker → client	A worker in the group completed (`kind = "job_completed"`)
`BusUIJobGroupCompletedMessage`	worker → client	The job group completed (`kind = "group_completed"`)

BusUIEventMessage

| Field | Type | Default | Description | | ------------ | ----- | ------- | -------------------------------- | -------------------------- | | source | str | | Publisher name | | target | str | None | None | Target worker name, or all | | event_name | str | "" | App-defined event name | | payload | Any | None | App-defined payload (schemaless) |

BusUICommandMessage

| Field | Type | Default | Description | | -------------- | ----- | ------- | ------------------------------ | -------------- | | source | str | | Publishing worker name | | target | str | None | None | Target, or all | | command_name | str | "" | App-defined command name | | payload | Any | None | Command payload (a plain dict) |

BusUIJobGroupStartedMessage

| Field | Type | Default | Description | | ------------- | ---------- | ------- | ------------------------------------------- | ------------------------------------------- | | job_id | str | "" | Shared job-group identifier | | workers | list[str] | None | None | Names of the workers the work was sent 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 |

BusUIJobUpdateMessage

Field	Type	Default	Description
`job_id`	`str`	`""`	The shared job-group identifier
`worker_name`	`str`	`""`	The worker that produced the update
`data`	`Any`	`None`	The worker’s update payload, verbatim
`at`	`int`	`0`	Epoch milliseconds when the update was sent

BusUIJobCompletedMessage

Field	Type	Default	Description
`job_id`	`str`	`""`	The shared job-group identifier
`worker_name`	`str`	`""`	The worker that produced the response
`status`	`str`	`""`	Completion status (`JobStatus` value)
`response`	`Any`	`None`	The worker’s response payload
`at`	`int`	`0`	Epoch milliseconds when received

BusUIJobGroupCompletedMessage

Field	Type	Default	Description
`job_id`	`str`	`""`	The shared job-group identifier
`at`	`int`	`0`	Epoch milliseconds when the group ended

Documentation Index

​Overview

​Base Types

​Common Fields

​Frame Transport

​BusFrameMessage

​Pipeline Commands

​BusTTSSpeakMessage

​Agent Lifecycle

​BusActivateWorkerMessage

​BusEndMessage / BusEndWorkerMessage

​BusCancelMessage / BusCancelWorkerMessage

​Registry & Errors

​BusAddWorkerMessage

​BusWorkerRegistryMessage

​BusWorkerReadyMessage

​BusWorkerErrorMessage / BusWorkerLocalErrorMessage

​Job Messages

​BusJobRequestMessage

​BusJobResponseMessage / BusJobResponseUrgentMessage

​BusJobUpdateMessage / BusJobUpdateUrgentMessage

​BusJobUpdateRequestMessage

​BusJobCancelMessage

​Job Streaming

​BusJobStreamStartMessage / BusJobStreamDataMessage / BusJobStreamEndMessage

​UI Messages

​BusUIEventMessage

​BusUICommandMessage

​BusUIJobGroupStartedMessage

​BusUIJobUpdateMessage

​BusUIJobCompletedMessage

​BusUIJobGroupCompletedMessage