Bus

AgentBus

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.

from pipecat_subagents.bus import AgentBus

Methods

start

async def start(self) -> None

Start dispatch tasks for all registered subscribers.

stop

async def stop(self) -> None

Stop all dispatch tasks.

async def subscribe(self, subscriber: BusSubscriber) -> None

Parameter	Type	Description
`subscriber`	`BusSubscriber`	The subscriber to register

unsubscribe

async def unsubscribe(self, subscriber: BusSubscriber) -> None

Remove a subscriber and cancel its dispatch task.

Parameter	Type	Description
`subscriber`	`BusSubscriber`	The subscriber to remove

send

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.

Parameter	Type	Description
`message`	`BusMessage`	The bus message to send

publish

@abstractmethod
async def publish(self, message: BusMessage) -> None

Publish a message to the transport. Subclasses implement this for the specific transport (e.g. in-process queue, Redis pub/sub).

Parameter	Type	Description
`message`	`BusMessage`	The bus message to publish

on_message_received

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

AsyncQueueBus

In-process bus that delivers messages via priority queues. This is the default bus used by AgentRunner when no bus is provided.

from pipecat_subagents.bus import AsyncQueueBus

bus = AsyncQueueBus()
runner = AgentRunner(bus=bus)

Suitable for single-process setups where all agents run in the same Python process.

RedisBus

Distributed agent bus backed by Redis pub/sub for cross-process communication.

Requires the redis extra: uv add "pipecat-ai-subagents[redis]"

from redis.asyncio import Redis
from pipecat_subagents.bus.network.redis import RedisBus

redis = Redis.from_url("redis://localhost:6379")
bus = RedisBus(redis=redis, channel="my-session")
runner = AgentRunner(bus=bus)

Configuration

Redis

required

A redis.asyncio.Redis client instance.

MessageSerializer | None

default:"None"

The MessageSerializer for encoding/decoding messages. Defaults to JSONMessageSerializer.

str

default:"pipecat:bus"

The Redis pub/sub channel name.

PgmqBus

Distributed agent bus backed by PGMQ (PostgreSQL Message Queue). Implements pub/sub fan-out on top of PGMQ’s point-to-point queue semantics by giving each bus instance its own queue and broadcasting on publish.

Requires the pgmq extra: uv add "pipecat-ai-subagents[pgmq]"

PgmqBus supports two backend modes:

DirectPgmqBackend: Calls pgmq.* directly with prefix-based peer discovery. Suitable when bus peers trust each other (single-tenant deployments, internal services).
IsolatedPgmqBackend: Uses SECURITY DEFINER Postgres wrappers for isolation. Suitable when bus peers should be isolated and the channel name is the bus capability.

DirectPgmqBackend example

from pgmq.async_queue import PGMQueue
from pipecat_subagents.bus.network.pgmq import PgmqBus

pgmq = PGMQueue(
    host="localhost",
    port="5432",
    database="postgres",
    username="postgres",
    password="...",
    pool_size=4,
)
await pgmq.init()
bus = PgmqBus(pgmq=pgmq, channel="pipecat_acme")

IsolatedPgmqBackend example

import asyncpg
from pipecat_subagents.bus.network.pgmq import PgmqBus
from pipecat_subagents.bus.network.pgmq_backends import IsolatedPgmqBackend

pool = await asyncpg.create_pool(
    host="localhost",
    port=5432,
    database="postgres",
    user="postgres",
    password="...",
    min_size=2,
    max_size=4,
)
backend = IsolatedPgmqBackend(pool=pool)
bus = PgmqBus(backend=backend, channel="pipecat_acme")

Configuration

PGMQueue | None

default:"None"

An initialized PGMQueue client (call await pgmq.init() before passing). Selects DirectPgmqBackend. Mutually exclusive with backend.

PgmqBackend | None

default:"None"

A backend instance (e.g. IsolatedPgmqBackend or custom). Mutually exclusive with pgmq.

MessageSerializer | None

default:"None"

The MessageSerializer for encoding/decoding messages. Defaults to JSONMessageSerializer.

str

default:"pipecat_bus"

Channel name. With DirectPgmqBackend this is sanitized into a queue-name prefix. With IsolatedPgmqBackend it is the bus capability passed to every wrapper call.

int

default:"30"

Seconds a read message stays invisible before redelivery.

int

default:"10"

Maximum messages to fetch per read.

int

default:"100"

Long-poll check interval in milliseconds. Backend may ignore if it doesn’t expose this knob.

int

default:"5"

Maximum seconds the reader blocks per poll cycle.

Prefer a session-mode pooler when available. Transaction-mode pooling works for direct pgmq.* calls but is fragile around the long-poll inside the SECURITY DEFINER bus_subscribe wrapper used by IsolatedPgmqBackend.The underlying connection pool must allow at least two concurrent connections (one for the reader’s long-poll, one for publishes).

BusBridgeProcessor

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

from pipecat_subagents.bus import BusBridgeProcessor

bridge = BusBridgeProcessor(
    bus=runner.bus,
    agent_name="session",
    bridge="voice",
)

pipeline = Pipeline([transport.input(), bridge, transport.output()])

Configuration

AgentBus

required

The AgentBus to exchange frames with.

str

required

Name of this agent, used as message source.

str | None

default:"None"

When set, only exchange frames with this agent.

str | None

default:"None"

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.

tuple[type[Frame], ...] | None

default:"None"

Extra frame types that should never cross the bus (on top of lifecycle frames which are always excluded).

BusSubscriber

Mixin for objects that receive messages from an AgentBus. Implementors override on_bus_message() to handle incoming messages. Concrete subscribers must provide a name property (typically inherited from BaseObject) that uniquely identifies the subscriber on the bus.

from pipecat_subagents.bus.subscriber import BusSubscriber

class MySubscriber(BusSubscriber):
    @property
    def name(self) -> str:
        return "my_subscriber"

    async def on_bus_message(self, message: BusMessage) -> None:
        ...

Properties

str

required

Unique name identifying this subscriber on the bus. Built-in subscribers inherit this from BaseObject; custom implementations that extend BusSubscriber directly must provide one.

Methods

async (message: BusMessage) -> None

Override to handle an incoming bus message.

Pipecat Server

Pipecat Subagents

Client SDKs

Pipecat Flows

Pipecat Cloud

CLI

AgentBus

Methods

start

stop

unsubscribe

send

publish

on_message_received

AsyncQueueBus

RedisBus

Configuration

PgmqBus

DirectPgmqBackend example

IsolatedPgmqBackend example

Configuration

BusBridgeProcessor

Configuration

BusSubscriber

Properties

Methods

​AgentBus

​Methods

​start

​stop

​subscribe

​unsubscribe

​send

​publish

​on_message_received

​AsyncQueueBus

​RedisBus

​Configuration

​PgmqBus

​DirectPgmqBackend example

​IsolatedPgmqBackend example

​Configuration

​BusBridgeProcessor

​Configuration

​BusSubscriber

​Properties

​Methods

AgentBus

Methods

start

stop

subscribe

unsubscribe

send

publish

on_message_received

AsyncQueueBus

RedisBus

Configuration

PgmqBus

DirectPgmqBackend example

IsolatedPgmqBackend example

Configuration

BusBridgeProcessor

Configuration

BusSubscriber

Properties

Methods