Skip to main content

​
What is the bus bridge?

The BusBridgeProcessor is a pipeline processor that connects a transport pipeline to the agent bus. It sits in the main agent’s pipeline where an LLM would normally go, routing frames to whichever agent is currently active. Without the bridge, your main agent’s pipeline would look like: 
transport.input → STT → context_agg → LLM → TTS → transport.output
With the bridge, the LLM is replaced: 
transport.input → STT → context_agg → BusBridge → TTS → transport.output
The bridge sends outgoing frames (like transcribed text) to the bus, and receives incoming frames (like LLM-generated text) from the active agent.
The examples above show a voice pipeline, but BusBridgeProcessor can be used in any type of pipeline. It works with any frames, not just audio.

​
How it works

The bridge operates in two directions: Outgoing (pipeline to bus): Frames flowing downstream through the pipeline are captured by the bridge and published as BusFrameMessage on the bus. The active agent receives these frames. Incoming (bus to pipeline): When an active agent sends frames back through the bus, the bridge pushes them into the pipeline. These frames continue downstream to TTS and the transport output. Certain frames are never sent across the bus:
  • Lifecycle frames (StartFrame, EndFrame, CancelFrame)
  • Transport-urgent frames (these pass through the bridge locally)

​
Basic usage

from pipecat_subagents.bus import BusBridgeProcessor

class MainAgent(BaseAgent):
    async def build_pipeline(self) -> Pipeline:
        bridge = BusBridgeProcessor(
            bus=self.bus,
            agent_name=self.name,
        )

        return Pipeline([
            self._transport.input(),
            stt,
            context_aggregator.user(),
            bridge,
            tts,
            self._transport.output(),
            context_aggregator.assistant(),
        ])

​
Named bridges

When you have multiple bridges in a system, you can name them to control which agents receive frames from which bridge. This is useful with parallel pipelines where each branch has its own bridge (for example, separate audio and video branches): 
voice_bridge = BusBridgeProcessor(
    bus=self.bus,
    agent_name=self.name,
    bridge="voice",
)
Agents can then filter which bridges they listen to: 
# Receives frames from all bridges
class AllBridgesAgent(LLMAgent):
    def __init__(self, name, *, bus):
        super().__init__(name, bus=bus, bridged=())

# Receives frames only from the "voice" bridge
class VoiceOnlyAgent(LLMAgent):
    def __init__(self, name, *, bus):
        super().__init__(name, bus=bus, bridged=("voice",))

​
The bridged agent side

When an agent sets bridged=() (or a tuple of bridge names), the framework automatically wraps its pipeline with edge processors that handle bus frame conversion: 
class MyLLMAgent(LLMAgent):
    def __init__(self, name, *, bus):
        super().__init__(name, bus=bus, bridged=())
The resulting pipeline looks like: 
BusEdgeProcessor (upstream) → [your pipeline] → BusEdgeProcessor (downstream)
The upstream edge processor receives BusFrameMessage from the bus and converts them back to regular Pipecat frames. The downstream edge processor captures output frames and sends them back through the bus. You don’t need to manage this — it happens automatically when bridged is set.