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.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	Default	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