Skip to content

channel_bridge

channel_bridge

ChannelBridge — unified orchestrator for multi-channel messaging.

Classes

ChannelBridge 

ChannelBridge(channels: Dict[str, BaseChannel], session_store: SessionStore, bus: EventBus, system: Any = None, agent_manager: Any = None, deep_research_agent: Any = None)

Orchestrates incoming messages across multiple channel adapters.

Provides backward-compatible send()/status()/list_channels() so it can replace the old single-channel bridge in app.state.

Source code in src/openjarvis/server/channel_bridge.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
def __init__(
    self,
    channels: Dict[str, BaseChannel],
    session_store: SessionStore,
    bus: EventBus,
    system: Any = None,
    agent_manager: Any = None,
    deep_research_agent: Any = None,
) -> None:
    self._channels = channels
    self._session_store = session_store
    self._bus = bus
    self._system = system
    self._agent_manager = agent_manager
    self._deep_research_agent = deep_research_agent
    self._notification_timestamps: Dict[str, float] = {}
    self._subscribe_notifications()