wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add task bus package

Browse Source 
Introduces `TaskBus`, the in-process `AsyncQueueBus`, the bus
message hierarchy (lifecycle, jobs, frames, registry), a
priority-aware bus queue, the `BusSubscriber` mixin, and the
`BusBridgeProcessor` / internal `_BusEdgeProcessor` used to
exchange frames between a local pipeline and the bus.
This commit is contained in:
Aleix Conchillo Flaqué
2026-05-13 19:13:53 -07:00
parent 30df8e4ca5
commit 5d94506265
17 changed files with 1692 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

88
src/pipecat/bus/__init__.py Normal file
View File

@@ -0,0 +1,88 @@
#
# Copyright (c) 2026, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Agent bus package -- pub/sub messaging between agents and the runner.
Provides the pub/sub infrastructure that connects agents to each other and to
the runner. Key components:
- `TaskBus` -- abstract base class defining the send/receive interface.
- `AsyncQueueBus` -- in-process implementation backed by ``asyncio.Queue``.
- `BusBridgeProcessor` -- bidirectional mid-pipeline bridge for
transport/session agents that exchanges frames with other agents
through the bus.
- `BusMessage` and its subclasses -- the typed message hierarchy used for
agent lifecycle events (activation, cancellation, shutdown), task
coordination, and frame transport.
"""
from pipecat.bus.bridge_processor import BusBridgeProcessor
from pipecat.bus.bus import TaskBus
from pipecat.bus.local import AsyncQueueBus
from pipecat.bus.messages import (
BusActivateTaskMessage,
BusAddTaskMessage,
BusCancelMessage,
BusCancelTaskMessage,
BusDataMessage,
BusDeactivateTaskMessage,
BusEndMessage,
BusEndTaskMessage,
BusFrameMessage,
BusJobCancelMessage,
BusJobRequestMessage,
BusJobResponseMessage,
BusJobResponseUrgentMessage,
BusJobStreamDataMessage,
BusJobStreamEndMessage,
BusJobStreamStartMessage,
BusJobUpdateMessage,
BusJobUpdateRequestMessage,
BusJobUpdateUrgentMessage,
BusLocalMessage,
BusMessage,
BusSystemMessage,
BusTaskErrorMessage,
BusTaskLocalErrorMessage,
BusTaskReadyMessage,
BusTaskRegistryMessage,
)
from pipecat.bus.subscriber import BusSubscriber
from pipecat.registry.types import TaskRegistryEntry
__all__ = [
"TaskBus",
"AsyncQueueBus",
"BusActivateTaskMessage",
"BusAddTaskMessage",
"BusTaskErrorMessage",
"BusTaskLocalErrorMessage",
"TaskRegistryEntry",
"BusTaskReadyMessage",
"BusTaskRegistryMessage",
"BusBridgeProcessor",
"BusCancelTaskMessage",
"BusCancelMessage",
"BusDeactivateTaskMessage",
"BusEndTaskMessage",
"BusEndMessage",
"BusFrameMessage",
"BusDataMessage",
"BusLocalMessage",
"BusMessage",
"BusSubscriber",
"BusSystemMessage",
"BusJobCancelMessage",
"BusJobRequestMessage",
"BusJobResponseMessage",
"BusJobResponseUrgentMessage",
"BusJobStreamDataMessage",
"BusJobStreamEndMessage",
"BusJobStreamStartMessage",
"BusJobUpdateMessage",
"BusJobUpdateRequestMessage",
"BusJobUpdateUrgentMessage",
]

21
src/pipecat/bus/adapters/__init__.py Normal file
View File

@@ -0,0 +1,21 @@
#
# Copyright (c) 2026, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Type adapters for bus message serialization.
Provides ready-made ``TypeAdapter`` implementations for common Pipecat types
(``LLMContext``, ``ToolsSchema``) used in bus messages.
"""
from pipecat.bus.adapters.base import TypeAdapter
from pipecat.bus.adapters.llm_context_adapter import LLMContextAdapter
from pipecat.bus.adapters.tools_schema_adapter import ToolsSchemaAdapter
__all__ = [
"LLMContextAdapter",
"ToolsSchemaAdapter",
"TypeAdapter",
]

59
src/pipecat/bus/adapters/base.py Normal file
View File

@@ -0,0 +1,59 @@
#
# Copyright (c) 2026, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Abstract base class for type adapters used by message serializers."""
from abc import ABC, abstractmethod
from collections.abc import Callable
from typing import Any
SerializeFunc = Callable[[Any], Any]
DeserializeFunc = Callable[[Any], Any]
class TypeAdapter(ABC):
"""Serialize and deserialize instances of a specific type for network transport.
Each adapter handles one or more types, converting them to/from a
JSON-compatible dict. Register adapters on a ``JSONMessageSerializer``
to handle non-JSON-native field values (e.g. ``LLMContext``, ``ToolsSchema``).
Adapters receive ``serialize_value`` and ``deserialize_value`` callbacks
from the serializer so they can recursively serialize nested fields
without importing the serializer itself.
"""
@abstractmethod
def serialize(self, obj: Any, serialize_value: SerializeFunc) -> dict[str, Any]:
"""Convert an object to a JSON-compatible dict.
Args:
obj: The object to serialize.
serialize_value: Callback to recursively serialize nested values.
Returns:
A dict representation of the object.
"""
pass
@abstractmethod
def deserialize(
self,
data: dict[str, Any],
deserialize_value: DeserializeFunc,
target_type: type | None = None,
) -> Any:
"""Reconstruct an object from a dict.
Args:
data: The dict representation produced by ``serialize()``.
deserialize_value: Callback to recursively deserialize nested values.
target_type: The resolved target class. Defaults to None.
Returns:
The reconstructed object.
"""
pass

108
src/pipecat/bus/adapters/llm_context_adapter.py Normal file
View File

@@ -0,0 +1,108 @@
#
# Copyright (c) 2026, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Type adapter for LLMContext serialization."""
from typing import Any
from openai import NOT_GIVEN as OPENAI_NOT_GIVEN
from pipecat.adapters.schemas.function_schema import FunctionSchema
from pipecat.adapters.schemas.tools_schema import ToolsSchema
from pipecat.bus.adapters.base import DeserializeFunc, SerializeFunc, TypeAdapter
from pipecat.processors.aggregators.llm_context import (
LLMContext,
LLMSpecificMessage,
NotGiven,
)
class LLMContextAdapter(TypeAdapter):
"""Serialize and deserialize ``LLMContext`` instances.
The ``NOT_GIVEN`` sentinel is preserved across serialization: missing
keys are restored as ``NOT_GIVEN`` on deserialization.
"""
def serialize(self, obj: Any, serialize_value: SerializeFunc) -> dict[str, Any]:
"""Serialize an ``LLMContext`` to a JSON-compatible dict.
Args:
obj: An ``LLMContext`` instance.
serialize_value: Callback to recursively serialize nested values.
Returns:
A dict with ``messages`` and, optionally, ``tools`` and
``tool_choice`` keys.
"""
result: dict[str, Any] = {
"messages": [self._serialize_message(m, serialize_value) for m in obj.messages],
}
if not isinstance(obj.tools, NotGiven):
result["tools"] = self._serialize_tools(obj.tools)
if not isinstance(obj.tool_choice, NotGiven):
result["tool_choice"] = serialize_value(obj.tool_choice)
return result
def deserialize(
self,
data: dict[str, Any],
deserialize_value: DeserializeFunc,
target_type: type | None = None,
) -> Any:
"""Reconstruct an ``LLMContext`` from a serialized dict.
Missing ``tools`` and ``tool_choice`` keys are restored as
OpenAI's ``NOT_GIVEN`` sentinel.
Args:
data: A dict produced by ``serialize()``.
deserialize_value: Callback to recursively deserialize nested values.
target_type: Unused. ``LLMContext`` is always the target.
Returns:
A new ``LLMContext`` instance.
"""
messages = [self._deserialize_message(m, deserialize_value) for m in data["messages"]]
tools = self._deserialize_tools(data["tools"]) if "tools" in data else OPENAI_NOT_GIVEN
tool_choice = (
deserialize_value(data["tool_choice"]) if "tool_choice" in data else OPENAI_NOT_GIVEN
)
return LLMContext(messages=messages, tools=tools, tool_choice=tool_choice)
def _serialize_message(self, msg: Any, serialize_value: SerializeFunc) -> dict[str, Any]:
if isinstance(msg, LLMSpecificMessage):
return {
"__specific__": True,
"llm": msg.llm,
"message": serialize_value(msg.message),
}
return serialize_value(msg)
def _deserialize_message(self, data: Any, deserialize_value: DeserializeFunc) -> Any:
if isinstance(data, dict) and data.get("__specific__"):
return LLMSpecificMessage(
llm=data["llm"],
message=deserialize_value(data["message"]),
)
return deserialize_value(data)
def _serialize_tools(self, tools: Any) -> list[dict[str, Any]]:
return [tool.to_default_dict() for tool in tools.standard_tools]
def _deserialize_tools(self, data: list[dict[str, Any]]) -> Any:
tools = []
for item in data:
params = item.get("parameters", {})
tools.append(
FunctionSchema(
name=item["name"],
description=item.get("description", ""),
properties=params.get("properties", {}),
required=params.get("required", []),
)
)
return ToolsSchema(standard_tools=tools)

58
src/pipecat/bus/adapters/tools_schema_adapter.py Normal file
View File

@@ -0,0 +1,58 @@
#
# Copyright (c) 2026, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Type adapter for ToolsSchema serialization."""
from typing import Any
from pipecat.adapters.schemas.function_schema import FunctionSchema
from pipecat.adapters.schemas.tools_schema import ToolsSchema
from pipecat.bus.adapters.base import DeserializeFunc, SerializeFunc, TypeAdapter
class ToolsSchemaAdapter(TypeAdapter):
"""Serialize and deserialize ``ToolsSchema`` instances for network transport."""
def serialize(self, obj: Any, serialize_value: SerializeFunc) -> dict[str, Any]:
"""Serialize a ``ToolsSchema`` to a JSON-compatible dict.
Args:
obj: A ``ToolsSchema`` instance.
serialize_value: Callback to recursively serialize nested values.
Returns:
A dict with a ``standard_tools`` list.
"""
return {"standard_tools": [tool.to_default_dict() for tool in obj.standard_tools]}
def deserialize(
self,
data: dict[str, Any],
deserialize_value: DeserializeFunc,
target_type: type | None = None,
) -> Any:
"""Reconstruct a ``ToolsSchema`` from a serialized dict.
Args:
data: A dict produced by ``serialize()``.
deserialize_value: Callback to recursively deserialize nested values.
target_type: Unused. ``ToolsSchema`` is always the target.
Returns:
A new ``ToolsSchema`` instance.
"""
tools = []
for item in data["standard_tools"]:
params = item.get("parameters", {})
tools.append(
FunctionSchema(
name=item["name"],
description=item.get("description", ""),
properties=params.get("properties", {}),
required=params.get("required", []),
)
)
return ToolsSchema(standard_tools=tools)

233
src/pipecat/bus/bridge_processor.py Normal file
View File

@@ -0,0 +1,233 @@
#
# Copyright (c) 2026, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Bus bridge and edge processors for inter-task frame routing.
Provides:
- `BusBridgeProcessor`: a mid-pipeline processor that exchanges frames
with other tasks through the bus, consuming local frames.
- `_BusEdgeProcessor`: a pipeline-edge processor used internally by
`PipelineTask` when ``bridged`` is set. Tees frames between the local
pipeline and the bus (frames continue locally and are also forwarded
to the bus).
"""
from typing import TYPE_CHECKING
from pipecat.bus.bus import TaskBus
from pipecat.bus.messages import BusFrameMessage, BusMessage
from pipecat.bus.subscriber import BusSubscriber
from pipecat.frames.frames import (
CancelFrame,
EndFrame,
Frame,
OutputTransportMessageUrgentFrame,
StartFrame,
StopFrame,
)
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor, FrameProcessorSetup
if TYPE_CHECKING:
from pipecat.pipeline.base_task import BaseTask
_LIFECYCLE_FRAMES = (StartFrame, EndFrame, CancelFrame, StopFrame)
_PASSTHROUGH_FRAMES = (OutputTransportMessageUrgentFrame,)
class BusBridgeProcessor(FrameProcessor, BusSubscriber):
"""Bidirectional mid-pipeline bridge between a Pipecat pipeline and the bus.
Placed in a transport or session agent's pipeline to exchange frames
with other agents via the `TaskBus`. Lifecycle and excluded frames
pass through locally without crossing the bus.
"""
def __init__(
self,
*,
bus: TaskBus,
agent_name: str,
target_agent: str | None = None,
bridge: str | None = None,
exclude_frames: tuple[type[Frame], ...] | None = None,
**kwargs,
):
"""Initialize the BusBridgeProcessor.
Args:
bus: The ``TaskBus`` to exchange frames with.
agent_name: Name of this agent, used as message source.
target_agent: When set, only exchange frames with this agent.
bridge: Optional bridge name for routing. When set, outgoing
frames are tagged with this name and only incoming frames
with the same bridge name are accepted.
exclude_frames: Extra frame types that should never cross the bus
(on top of lifecycle frames which are always excluded).
**kwargs: Additional arguments passed to ``FrameProcessor``.
"""
super().__init__(**kwargs)
self._bus = bus
self._agent_name = agent_name
self._target_agent = target_agent
self._bridge = bridge
self._exclude_frames = exclude_frames or ()
async def setup(self, setup: FrameProcessorSetup):
"""Subscribe to the bus during processor setup."""
await super().setup(setup)
await self._bus.subscribe(self)
async def cleanup(self):
"""Unsubscribe from the bus on cleanup."""
await super().cleanup()
await self._bus.unsubscribe(self)
async def process_frame(self, frame: Frame, direction: FrameDirection):
"""Process a frame: send to bus, or pass through locally if excluded.
Args:
frame: The frame to process.
direction: The direction the frame is traveling.
"""
await super().process_frame(frame, direction)
# Lifecycle frames never cross the bus
if isinstance(frame, _LIFECYCLE_FRAMES):
await self.push_frame(frame, direction)
return
# Urgent transport frames pass through directly. They need to
# reach the transport even when no child agent is active yet.
if isinstance(frame, _PASSTHROUGH_FRAMES):
await self.push_frame(frame, direction)
return
# Excluded frames never cross the bus
if self._exclude_frames and isinstance(frame, self._exclude_frames):
await self.push_frame(frame, direction)
return
# Send to bus
msg = BusFrameMessage(
source=self._agent_name,
frame=frame,
direction=direction,
bridge=self._bridge,
)
await self._bus.send(msg)
async def on_bus_message(self, message: BusMessage) -> None:
"""Handle an incoming bus message by pushing its frame into the pipeline.
Args:
message: The bus message to handle.
"""
if not isinstance(message, BusFrameMessage):
return
# Skip own frames
if message.source == self._agent_name:
return
# Filter by bridge name
if self._bridge and message.bridge != self._bridge:
return
# If target_agent set, only accept from that agent
if self._target_agent and message.source != self._target_agent:
return
# If message targeted at someone else, skip
if message.target and message.target != self._agent_name:
return
await self.push_frame(message.frame, message.direction)
class _BusEdgeProcessor(FrameProcessor, BusSubscriber):
"""Pipeline-edge tee between a local pipeline and the bus.
Placed by `PipelineTask` at the source and sink of a bridged
pipeline. Frames always continue through the local pipeline; in
addition, frames travelling in ``direction`` are forwarded to the
bus, and frames received from the bus in the opposite direction
are injected into the pipeline.
"""
def __init__(
self,
*,
bus: TaskBus,
task: "BaseTask",
direction: FrameDirection,
bridges: tuple[str, ...] = (),
exclude_frames: tuple[type[Frame], ...] | None = None,
**kwargs,
):
"""Initialize the edge processor.
Args:
bus: The ``TaskBus`` to exchange frames with.
task: The owning task; ``task.name`` is the message source
and ``task.active`` gates inbound frames.
direction: Direction this edge captures and forwards to the
bus. Inbound frames from the bus travelling in the
opposite direction are injected here.
bridges: Bridge names this edge accepts. Empty tuple accepts
frames from all bridges.
exclude_frames: Extra frame types that should never cross
the bus (lifecycle frames are always excluded).
**kwargs: Additional arguments passed to ``FrameProcessor``.
"""
super().__init__(**kwargs)
self._bus = bus
self._task = task
self._direction = direction
self._bridges = bridges
self._exclude_frames = exclude_frames or ()
async def setup(self, setup: FrameProcessorSetup):
"""Subscribe to the bus during processor setup."""
await super().setup(setup)
await self._bus.subscribe(self)
async def cleanup(self):
"""Unsubscribe from the bus on cleanup."""
await super().cleanup()
await self._bus.unsubscribe(self)
async def process_frame(self, frame: Frame, direction: FrameDirection):
"""Pass the frame through locally and forward matching ones to the bus."""
await super().process_frame(frame, direction)
await self.push_frame(frame, direction)
if direction != self._direction:
return
if isinstance(frame, _LIFECYCLE_FRAMES):
return
if self._exclude_frames and isinstance(frame, self._exclude_frames):
return
await self._bus.send(
BusFrameMessage(source=self._task.name, frame=frame, direction=direction)
)
async def on_bus_message(self, message: BusMessage) -> None:
"""Inject incoming bus frames into the pipeline."""
if not isinstance(message, BusFrameMessage):
return
if message.source == self._task.name:
return
if message.direction == self._direction:
return
if not self._task.active:
return
if message.target and message.target != self._task.name:
return
if self._bridges and message.bridge not in self._bridges:
return
await self.push_frame(message.frame, message.direction)

183
src/pipecat/bus/bus.py Normal file
View File

@@ -0,0 +1,183 @@
#
# Copyright (c) 2026, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Abstract agent bus for inter-agent pub/sub messaging.
Provides the abstract `TaskBus` base class. Concrete implementations
(e.g. `AsyncQueueBus`) live in separate modules.
"""
import asyncio
from abc import abstractmethod
from dataclasses import dataclass, field
from pipecat.bus.messages import BusLocalMessage, BusMessage
from pipecat.bus.queue import BusMessageQueue
from pipecat.bus.subscriber import BusSubscriber
from pipecat.frames.frames import SystemFrame
from pipecat.utils.base_object import BaseObject
@dataclass
class BusSubscription:
"""A single subscriber's state on the bus.
Parameters:
subscriber: The subscriber receiving messages.
queue: Priority queue for incoming messages.
data_queue: Secondary queue for data messages dispatched by
the router task.
router_task: Task that reads from the priority queue, handles
system messages inline, and routes data messages to the
data queue.
data_task: Task that processes data messages sequentially from
the data queue.
"""
subscriber: BusSubscriber
queue: BusMessageQueue = field(default_factory=BusMessageQueue, repr=False)
data_queue: asyncio.Queue = field(default_factory=asyncio.Queue, repr=False)
router_task: asyncio.Task | None = field(default=None, repr=False)
data_task: asyncio.Task | None = field(default=None, repr=False)
class TaskBus(BaseObject):
"""Abstract base for inter-agent and runner-agent communication.
Provides pub/sub messaging where each subscriber receives messages
independently through its own priority queue. System messages
(e.g. cancel) are delivered before normal data messages.
Subclasses implement ``publish()`` for the specific transport.
``send()`` handles local-only messages automatically. For network
buses, override ``start()``/``stop()`` to manage connections and
call ``on_message_received()`` when messages arrive from the
network.
"""
def __init__(self, **kwargs):
"""Initialize the TaskBus.
Args:
**kwargs: Additional arguments passed to `BaseObject`.
"""
super().__init__(**kwargs)
self._subscriptions: dict[str, BusSubscription] = {}
self._running = False
async def start(self):
"""Start dispatch tasks for all registered subscribers."""
if self._running:
return
self._running = True
for sub in self._subscriptions.values():
self._start_dispatch_task(sub)
# Schedule tasks right away.
await asyncio.sleep(0)
async def stop(self):
"""Stop all dispatch tasks."""
if not self._running:
return
self._running = False
for sub in self._subscriptions.values():
if sub.router_task:
await self.cancel_task(sub.router_task)
sub.router_task = None
if sub.data_task:
await self.cancel_task(sub.data_task)
sub.data_task = None
async def subscribe(self, subscriber: BusSubscriber) -> None:
"""Register a subscriber to receive messages from the bus.
Args:
subscriber: The `BusSubscriber` to register.
"""
sub = BusSubscription(subscriber=subscriber)
if self._running:
self._start_dispatch_task(sub)
# Schedule task right away.
await asyncio.sleep(0)
if subscriber.name in self._subscriptions:
raise ValueError(f"Subscriber '{subscriber.name}' is already registered on the bus")
self._subscriptions[subscriber.name] = sub
async def unsubscribe(self, subscriber: BusSubscriber) -> None:
"""Remove a subscriber and cancel its dispatch tasks.
Args:
subscriber: The `BusSubscriber` to remove.
"""
sub = self._subscriptions.pop(subscriber.name, None)
if sub:
if sub.router_task:
await self.cancel_task(sub.router_task)
if sub.data_task:
await self.cancel_task(sub.data_task)
async def send(self, message: BusMessage) -> None:
"""Send a message through the bus.
Local-only messages are delivered directly to subscribers.
All other messages are passed to ``publish()`` for transport.
Args:
message: The bus message to send.
"""
if isinstance(message, BusLocalMessage):
self.on_message_received(message)
return
await self.publish(message)
@abstractmethod
async def publish(self, message: BusMessage) -> None:
"""Publish a message to the transport.
Subclasses implement this for the specific transport. Called
by ``send()`` after filtering local-only messages.
Args:
message: The bus message to publish.
"""
pass
def on_message_received(self, message: BusMessage) -> None:
"""Deliver a message to all local subscribers via their priority queues.
Called by bus implementations when a message arrives (either from
a local ``send()`` or from a network transport).
"""
for sub in self._subscriptions.values():
sub.queue.put_nowait(message)
def _start_dispatch_task(self, sub: BusSubscription) -> None:
"""Start the router and data dispatch tasks for a subscriber."""
sub.router_task = self.create_task(self._router_task(sub), f"bus_router_{sub.subscriber}")
sub.data_task = self.create_task(
self._data_dispatch_task(sub), f"bus_data_{sub.subscriber}"
)
async def _router_task(self, sub: BusSubscription):
"""Route system messages inline, data messages to the data queue."""
try:
while True:
message = await sub.queue.get()
if isinstance(message, SystemFrame):
await sub.subscriber.on_bus_message(message)
else:
sub.data_queue.put_nowait(message)
except asyncio.CancelledError:
pass
async def _data_dispatch_task(self, sub: BusSubscription):
"""Process data messages sequentially from the data queue."""
try:
while True:
message = await sub.data_queue.get()
await sub.subscriber.on_bus_message(message)
except asyncio.CancelledError:
pass

11
src/pipecat/bus/local/__init__.py Normal file
View File

@@ -0,0 +1,11 @@
#
# Copyright (c) 2026, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Local (in-process) bus implementations."""
from pipecat.bus.local.async_queue import AsyncQueueBus
__all__ = ["AsyncQueueBus"]

25
src/pipecat/bus/local/async_queue.py Normal file
View File

@@ -0,0 +1,25 @@
#
# Copyright (c) 2026, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
"""In-process agent bus backed by asyncio queues."""
from loguru import logger
from pipecat.bus.bus import TaskBus
from pipecat.bus.messages import BusMessage
class AsyncQueueBus(TaskBus):
"""In-process bus that delivers messages via priority queues."""
async def publish(self, message: BusMessage) -> None:
"""Deliver a message to all local subscriber queues.
Args:
message: The bus message to deliver.
"""
logger.trace(f"{self}: sending {message}")
self.on_message_received(message)

407
src/pipecat/bus/messages.py Normal file
View File

@@ -0,0 +1,407 @@
#
# Copyright (c) 2026, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Bus message types for inter-agent communication.
Defines the message hierarchy used by the `TaskBus` for pub/sub messaging
between agents, the session, and the runner.
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import TYPE_CHECKING
from pipecat.frames.frames import DataFrame, Frame, SystemFrame
from pipecat.processors.frame_processor import FrameDirection
from pipecat.registry.types import TaskRegistryEntry
if TYPE_CHECKING:
from pipecat.pipeline.base_task import BaseTask
from pipecat.pipeline.job_context import JobStatus
# ---------------------------------------------------------------------------
# Base types and mixins
# ---------------------------------------------------------------------------
class BusMessage:
"""Mixin carrying source/target metadata for bus messages.
Not a frame itself. Combined with ``DataFrame`` or ``SystemFrame``
to create concrete message types with appropriate priority.
"""
source: str
target: str | None = None
def __str__(self):
return f"{type(self).__name__} (source={self.source}, target={self.target})"
class BusLocalMessage:
"""Mixin: message stays on the local bus, never forwarded to remote buses."""
pass
@dataclass(kw_only=True)
class BusDataMessage(BusMessage, DataFrame):
"""Normal-priority bus message.
Parameters:
source: Name of the agent or component that sent this message.
target: Name of the intended recipient agent, or None for broadcast.
"""
source: str
target: str | None = None
@dataclass(kw_only=True)
class BusSystemMessage(BusMessage, SystemFrame):
"""High-priority bus message that preempts normal messages in subscriber queues.
Parameters:
source: Name of the agent or component that sent this message.
target: Name of the intended recipient agent, or None for broadcast.
"""
source: str
target: str | None = None
# ---------------------------------------------------------------------------
# Frame transport
# ---------------------------------------------------------------------------
@dataclass
class BusFrameMessage(BusDataMessage):
"""Wraps a Pipecat `Frame` for transport over the bus.
Parameters:
frame: The Pipecat frame to transport.
direction: Direction the frame should travel in the recipient's pipeline.
bridge: Optional bridge name for routing in multi-bridge setups.
"""
frame: Frame
direction: FrameDirection
bridge: str | None = None
# ---------------------------------------------------------------------------
# Agent lifecycle
# ---------------------------------------------------------------------------
@dataclass
class BusActivateTaskMessage(BusDataMessage):
"""Tells a targeted agent to become active and start processing.
Parameters:
args: Optional activation arguments forwarded to ``on_activated``.
"""
args: dict | None = None
@dataclass
class BusDeactivateTaskMessage(BusDataMessage):
"""Tells a targeted agent to become inactive and stop processing."""
pass
@dataclass
class BusEndMessage(BusDataMessage):
"""Request a graceful end of the session.
Sent by an agent to the runner, which responds by sending
`BusEndTaskMessage` to each agent.
Parameters:
reason: Optional human-readable reason for ending.
"""
reason: str | None = None
@dataclass
class BusEndTaskMessage(BusDataMessage):
"""Tells a targeted agent to end its pipeline gracefully.
Sent by the runner to individual agents during shutdown.
Parameters:
reason: Optional human-readable reason for ending.
"""
reason: str | None = None
@dataclass
class BusCancelMessage(BusSystemMessage):
"""Request a hard cancel of the session.
Sent by an agent to the runner, which responds by sending
`BusCancelTaskMessage` to each agent.
Parameters:
reason: Optional human-readable reason for the cancellation.
"""
reason: str | None = None
@dataclass
class BusCancelTaskMessage(BusSystemMessage):
"""Tells a targeted agent to cancel its pipeline task.
Sent by the runner to individual agents during cancellation.
Parameters:
reason: Optional human-readable reason for the cancellation.
"""
reason: str | None = None
# ---------------------------------------------------------------------------
# Agent registry and errors
# ---------------------------------------------------------------------------
@dataclass
class BusAddTaskMessage(BusSystemMessage, BusLocalMessage):
"""Request to add a task to the local runner.
Local-only: carries an in-memory task reference that cannot be
serialized over the network.
Parameters:
task: The task instance to add.
"""
task: BaseTask
@dataclass
class BusTaskRegistryMessage(BusSystemMessage):
"""Snapshot of tasks managed by a runner.
Sent by the runner on startup and when new runners connect,
so that remote runners can discover each other's tasks.
Parameters:
runner: Name of the runner that owns these tasks.
tasks: List of task entries with their state.
"""
runner: str
tasks: list[TaskRegistryEntry]
@dataclass
class BusTaskReadyMessage(BusDataMessage):
"""Announces that an agent is ready.
Sent when any agent (root or child) becomes ready. Carries the
agent's parent name so observers can reconstruct the full hierarchy.
Parameters:
runner: Name of the runner managing this agent.
parent: Name of the parent agent, or None for root agents.
active: Whether the agent started active.
bridged: Whether the agent is bridged (receives pipeline frames
from the bus).
started_at: Unix timestamp when the agent became ready.
"""
runner: str
parent: str | None = None
active: bool = False
bridged: bool = False
started_at: float | None = None
@dataclass
class BusTaskErrorMessage(BusSystemMessage):
"""Reports an error from a root agent.
Sent over the network so remote agents can react. For child agent
errors, see ``BusTaskLocalErrorMessage``.
Parameters:
error: Description of the error.
"""
error: str
@dataclass
class BusTaskLocalErrorMessage(BusSystemMessage, BusLocalMessage):
"""Reports an error from a child agent to its parent.
Local-only: never crosses the network. The parent receives it
via ``on_task_failed()``.
Parameters:
error: Description of the error.
"""
error: str
# ---------------------------------------------------------------------------
# Tasks
# ---------------------------------------------------------------------------
@dataclass
class BusJobRequestMessage(BusDataMessage):
"""Requests a task agent to start work.
Parameters:
job_id: Unique identifier for this task.
job_name: Optional task name for routing to named handlers.
payload: Optional structured data describing the work.
"""
job_id: str
job_name: str | None = None
payload: dict | None = None
@dataclass
class BusJobResponseMessage(BusDataMessage):
"""Response from a task agent when it completes.
Parameters:
job_id: The task identifier.
status: Completion status.
response: Optional result data.
"""
job_id: str
status: JobStatus
response: dict | None = None
@dataclass
class BusJobResponseUrgentMessage(BusSystemMessage):
"""High-priority response from a task agent.
Same semantics as ``BusJobResponseMessage`` but delivered with
system priority, preempting queued data messages.
Parameters:
job_id: The task identifier.
status: Completion status.
response: Optional result data.
"""
job_id: str
status: JobStatus
response: dict | None = None
@dataclass
class BusJobUpdateMessage(BusDataMessage):
"""Progress update from a task agent.
Parameters:
job_id: The task identifier.
update: Optional progress data.
"""
job_id: str
update: dict | None = None
@dataclass
class BusJobUpdateUrgentMessage(BusSystemMessage):
"""High-priority progress update from a task agent.
Same semantics as ``BusJobUpdateMessage`` but delivered with
system priority, preempting queued data messages.
Parameters:
job_id: The task identifier.
update: Optional progress data.
"""
job_id: str
update: dict | None = None
@dataclass
class BusJobUpdateRequestMessage(BusDataMessage):
"""Request a progress update from a task agent.
Parameters:
job_id: The task identifier.
"""
job_id: str
@dataclass
class BusJobCancelMessage(BusSystemMessage):
"""Cancel a running task.
Parameters:
job_id: The task identifier.
reason: Optional human-readable reason for cancellation.
"""
job_id: str
reason: str | None = None
# ---------------------------------------------------------------------------
# Task streaming
# ---------------------------------------------------------------------------
@dataclass
class BusJobStreamStartMessage(BusDataMessage):
"""Signals the start of a streaming task response.
Parameters:
job_id: The task identifier.
data: Optional metadata (e.g. content type).
"""
job_id: str
data: dict | None = None
@dataclass
class BusJobStreamDataMessage(BusDataMessage):
"""A chunk of streaming task data.
Parameters:
job_id: The task identifier.
data: The chunk payload.
"""
job_id: str
data: dict | None = None
@dataclass
class BusJobStreamEndMessage(BusDataMessage):
"""Signals the end of a streaming task response.
Parameters:
job_id: The task identifier.
data: Optional final metadata.
"""
job_id: str
data: dict | None = None

26
src/pipecat/bus/network/__init__.py Normal file
View File

@@ -0,0 +1,26 @@
#
# Copyright (c) 2026, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Network bus implementations for distributed agents.
Each adapter has its own optional dependency. Imports are lazy so the
package can be loaded with only the extras you need; importing a specific
bus without its extra raises a clear error from that submodule.
"""
__all__ = ["PgmqBus", "RedisBus"]
def __getattr__(name: str):
if name == "PgmqBus":
from pipecat.bus.network.pgmq import PgmqBus
return PgmqBus
if name == "RedisBus":
from pipecat.bus.network.redis import RedisBus
return RedisBus
raise AttributeError(f"module {__name__!r} has no attribute {name!r}")

107
src/pipecat/bus/network/redis.py Normal file
View File

@@ -0,0 +1,107 @@
#
# Copyright (c) 2026, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Redis pub/sub agent bus for distributed agents."""
import asyncio
from loguru import logger
from pipecat.bus.bus import TaskBus
from pipecat.bus.messages import BusMessage
from pipecat.bus.serializers import JSONMessageSerializer
from pipecat.bus.serializers.base import MessageSerializer
try:
from redis.asyncio import Redis
from redis.asyncio.client import PubSub
except ModuleNotFoundError as e:
logger.error(f"Exception: {e}")
logger.error("In order to use RedisBus, you need to `pip install pipecat-ai-subagents[redis]`.")
raise Exception(f"Missing module: {e}")
class RedisBus(TaskBus):
"""Distributed agent bus backed by Redis pub/sub.
Publishes serialized messages to a Redis channel for cross-process
communication. ``BusLocalMessage`` messages bypass Redis and are
delivered directly to local subscribers.
Requires the ``redis[hiredis]`` package (``redis.asyncio``).
Example::
from redis.asyncio import Redis
redis = Redis.from_url("redis://localhost:6379")
bus = RedisBus(redis=redis, channel="my-session")
"""
def __init__(
self,
*,
redis: Redis,
serializer: MessageSerializer | None = None,
channel: str = "pipecat:bus",
**kwargs,
):
"""Initialize the RedisBus.
Args:
redis: A ``redis.asyncio.Redis`` client instance.
serializer: The `MessageSerializer` for encoding/decoding messages.
Defaults to `JSONMessageSerializer`.
channel: The Redis pub/sub channel name. Defaults to ``"pipecat:bus"``.
**kwargs: Additional arguments passed to `TaskBus`.
"""
super().__init__(**kwargs)
self._redis = redis
self._serializer = serializer or JSONMessageSerializer()
self._channel = channel
self._pubsub: PubSub | None = None
self._reader_task: asyncio.Task | None = None
async def start(self):
"""Subscribe to Redis channel and start the reader task."""
await super().start()
self._pubsub = self._redis.pubsub()
await self._pubsub.subscribe(self._channel)
self._reader_task = self.create_task(self._reader_loop(), f"{self}::redis_reader")
await asyncio.sleep(0)
async def stop(self):
"""Stop the reader task and unsubscribe from Redis."""
await super().stop()
if self._reader_task:
await self.cancel_task(self._reader_task)
self._reader_task = None
if self._pubsub:
await self._pubsub.unsubscribe(self._channel)
await self._pubsub.close()
self._pubsub = None
async def publish(self, message: BusMessage) -> None:
"""Publish a message to the Redis channel.
Args:
message: The bus message to publish.
"""
logger.trace(f"{self}: publishing {message} to {self._channel}")
data = self._serializer.serialize(message)
await self._redis.publish(self._channel, data)
async def _reader_loop(self) -> None:
"""Read messages from Redis pub/sub and deliver to subscribers."""
async for raw_message in self._pubsub.listen():
if raw_message["type"] != "message":
continue
try:
message = self._serializer.deserialize(raw_message["data"])
if message:
self.on_message_received(message)
except Exception:
logger.exception(f"{self}: failed to deserialize message")

61
src/pipecat/bus/queue.py Normal file
View File

@@ -0,0 +1,61 @@
#
# Copyright (c) 2026, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Priority queue for bus messages."""
import asyncio
from typing import Any
from pipecat.frames.frames import SystemFrame
HIGH_PRIORITY = 1
LOW_PRIORITY = 2
class BusMessageQueue(asyncio.PriorityQueue):
"""Priority queue that delivers system messages before normal messages.
Messages that extend ``SystemFrame`` (e.g. cancel messages) get high
priority. All other messages are delivered in FIFO order at normal
priority.
"""
def __init__(self):
"""Initialize the BusMessageQueue."""
super().__init__()
self._high_counter = 0
self._low_counter = 0
def put_nowait(self, item) -> None:
"""Add a message to the queue with automatic priority assignment.
Args:
item: The bus message to enqueue.
"""
if isinstance(item, SystemFrame):
self._high_counter += 1
super().put_nowait((HIGH_PRIORITY, self._high_counter, item))
else:
self._low_counter += 1
super().put_nowait((LOW_PRIORITY, self._low_counter, item))
async def put(self, item) -> None:
"""Add a message to the queue with automatic priority assignment.
Args:
item: The bus message to enqueue.
"""
if isinstance(item, SystemFrame):
self._high_counter += 1
await super().put((HIGH_PRIORITY, self._high_counter, item))
else:
self._low_counter += 1
await super().put((LOW_PRIORITY, self._low_counter, item))
async def get(self) -> Any:
"""Get the next message, with system messages prioritized."""
_, _, message = await super().get()
return message

19
src/pipecat/bus/serializers/__init__.py Normal file
View File

@@ -0,0 +1,19 @@
#
# Copyright (c) 2026, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Bus message serialization for network transport.
Provides the abstract `MessageSerializer` interface and a default
`JSONMessageSerializer` implementation.
"""
from pipecat.bus.serializers.base import MessageSerializer
from pipecat.bus.serializers.json import JSONMessageSerializer
__all__ = [
"JSONMessageSerializer",
"MessageSerializer",
]

43
src/pipecat/bus/serializers/base.py Normal file
View File

@@ -0,0 +1,43 @@
#
# Copyright (c) 2026, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Abstract base classes for bus message serialization."""
from abc import ABC, abstractmethod
from pipecat.bus.messages import BusMessage
class MessageSerializer(ABC):
"""Serialize and deserialize `BusMessage` instances for network transport.
Network bus implementations use a `MessageSerializer` to convert messages
to bytes for transmission and reconstruct them on the receiving end.
"""
@abstractmethod
def serialize(self, message: BusMessage) -> bytes:
"""Convert a bus message to bytes.
Args:
message: The bus message to serialize.
Returns:
The serialized bytes.
"""
pass
@abstractmethod
def deserialize(self, data: bytes) -> BusMessage | None:
"""Reconstruct a bus message from bytes.
Args:
data: The serialized bytes produced by `serialize()`.
Returns:
The reconstructed `BusMessage`.
"""
pass

212
src/pipecat/bus/serializers/json.py Normal file
View File

@@ -0,0 +1,212 @@
#
# Copyright (c) 2026, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
"""JSON-based bus message serializer with pluggable type adapters."""
import base64
import dataclasses
import importlib
import json
from enum import Enum
from functools import cache
from typing import Any
from loguru import logger
from pydantic import BaseModel
from pipecat.bus.adapters.base import TypeAdapter
from pipecat.bus.messages import BusMessage
from pipecat.bus.serializers.base import MessageSerializer
# JSON-native types that don't need an adapter.
_JSON_NATIVE = (str, int, float, bool, type(None))
class JSONMessageSerializer(MessageSerializer):
"""Serialize bus messages as JSON with pluggable type adapters.
Handles JSON-native types, enums, bytes, dataclasses, and any type
with a registered ``TypeAdapter`` (e.g. ``LLMContext``, ``ToolsSchema``).
Adapters for common Pipecat types are registered by default.
Additional type adapters can be registered via ``register_adapter()``.
Example::
serializer = JSONMessageSerializer()
data = serializer.serialize(message)
restored = serializer.deserialize(data)
"""
def __init__(self):
"""Create a serializer with default adapters for `LLMContext` and `ToolsSchema`."""
from pipecat.adapters.schemas.tools_schema import ToolsSchema
from pipecat.bus.adapters import LLMContextAdapter, ToolsSchemaAdapter
from pipecat.processors.aggregators.llm_context import LLMContext
self._adapters: dict[type, TypeAdapter] = {
LLMContext: LLMContextAdapter(),
ToolsSchema: ToolsSchemaAdapter(),
}
def register_adapter(self, type_: type, adapter: TypeAdapter) -> None:
"""Register a type adapter.
Args:
type_: The type to handle.
adapter: The adapter that serializes/deserializes instances of this type.
"""
self._adapters[type_] = adapter
def serialize(self, message: BusMessage) -> bytes:
"""Convert a bus message to JSON bytes.
Args:
message: The bus message to serialize.
Returns:
UTF-8 encoded JSON bytes.
"""
data = self._serialize_value(message)
return json.dumps(data, separators=(",", ":")).encode("utf-8")
def deserialize(self, data: bytes) -> BusMessage | None:
"""Reconstruct a bus message from JSON bytes.
Args:
data: The JSON bytes produced by `serialize()`.
Returns:
The reconstructed `BusMessage`, or None if deserialization fails.
"""
payload = json.loads(data)
return self._deserialize_value(payload)
def _serialize_value(self, value: Any) -> Any:
"""Recursively serialize a value to a JSON-compatible representation."""
if isinstance(value, _JSON_NATIVE):
return value
if isinstance(value, Enum):
return {
"__type__": f"{type(value).__module__}.{type(value).__name__}",
"__data__": value.name,
}
if isinstance(value, dict):
return {k: self._serialize_value(v) for k, v in value.items()}
if isinstance(value, list):
return [self._serialize_value(v) for v in value]
if isinstance(value, bytes):
return {"__type__": "bytes", "__data__": base64.b64encode(value).decode("ascii")}
if isinstance(value, BaseModel):
return {
"__type__": f"{type(value).__module__}.{type(value).__name__}",
"__data__": {
k: self._serialize_value(v) for k, v in value.__dict__.items() if v is not None
},
}
if callable(value):
return None
adapter = self._find_adapter(type(value))
if adapter is not None:
return {
"__type__": f"{type(value).__module__}.{type(value).__name__}",
"__data__": adapter.serialize(value, self._serialize_value),
}
if dataclasses.is_dataclass(value) and not isinstance(value, type):
fields = {}
for f in dataclasses.fields(value):
v = getattr(value, f.name)
if v is None:
continue
serialized = self._serialize_value(v)
if serialized is not None:
fields[f.name] = serialized
return {
"__type__": f"{type(value).__module__}.{type(value).__name__}",
"__data__": fields,
}
logger.warning(
f"JSONMessageSerializer: skipping field with unserializable type {type(value).__name__}"
)
return None
def _deserialize_value(self, value: Any) -> Any:
"""Recursively deserialize a value from its JSON representation."""
if isinstance(value, _JSON_NATIVE):
return value
if isinstance(value, list):
return [self._deserialize_value(v) for v in value]
if isinstance(value, dict):
if "__type__" in value and "__data__" in value:
return self._deserialize_typed(value["__type__"], value["__data__"])
return {k: self._deserialize_value(v) for k, v in value.items()}
return value
def _deserialize_typed(self, type_name: str, data: Any) -> Any:
"""Deserialize a tagged value using its fully qualified type name."""
if type_name == "bytes":
return base64.b64decode(data)
cls = _resolve_type(type_name)
if cls is None:
logger.warning(f"JSONMessageSerializer: could not resolve type {type_name}")
return None
if issubclass(cls, Enum):
return cls[data]
adapter = self._find_adapter(cls)
if adapter is not None:
return adapter.deserialize(data, self._deserialize_value, target_type=cls)
if isinstance(data, dict) and issubclass(cls, BaseModel):
return cls.model_validate({k: self._deserialize_value(v) for k, v in data.items()})
if dataclasses.is_dataclass(cls) and isinstance(data, dict):
init_fields = {f.name: f for f in dataclasses.fields(cls) if f.init}
init_kwargs = {}
post_init = {}
for key, value in data.items():
deserialized = self._deserialize_value(value)
if key in init_fields:
init_kwargs[key] = deserialized
else:
post_init[key] = deserialized
for name, f in init_fields.items():
if name not in init_kwargs:
if (
f.default is dataclasses.MISSING
and f.default_factory is dataclasses.MISSING
):
init_kwargs[name] = None
obj = cls(**init_kwargs)
for key, value in post_init.items():
setattr(obj, key, value)
return obj
logger.warning(f"JSONMessageSerializer: no adapter registered for type {type_name}")
return None
def _find_adapter(self, type_: type) -> TypeAdapter | None:
"""Find an adapter for a type, checking parent classes via MRO."""
for cls in type_.__mro__:
if cls in self._adapters:
return self._adapters[cls]
return None
@cache
def _resolve_type(qualified_name: str) -> type | None:
"""Resolve a fully qualified type name to its class.
Args:
qualified_name: Dotted path like ``"pipecat.frames.frames.TextFrame"``.
Returns:
The resolved class, or None if it cannot be found.
"""
module_path, _, class_name = qualified_name.rpartition(".")
if not module_path:
return None
try:
module = importlib.import_module(module_path)
return getattr(module, class_name, None)
except ImportError:
return None

31
src/pipecat/bus/subscriber.py Normal file
View File

@@ -0,0 +1,31 @@
#
# Copyright (c) 2026, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Bus subscriber mixin for receiving messages from an TaskBus."""
from pipecat.bus.messages import BusMessage
class BusSubscriber:
"""Mixin for objects that receive messages from an `TaskBus`.
Implementors override `on_bus_message()` to handle incoming messages.
Concrete subscribers must provide a ``name`` property (typically
inherited from ``BaseObject``).
"""
@property
def name(self) -> str:
"""Unique name identifying this subscriber on the bus."""
raise NotImplementedError
async def on_bus_message(self, message: BusMessage) -> None:
"""Handle an incoming bus message.
Args:
message: The bus message to handle.
"""
...