Skip to content

API Reference: Channels

The openjarvis.channels package provides the channel messaging abstraction and the OpenClaw gateway bridge. All public classes and types are documented below.

For usage examples, CLI commands, and configuration, see the Channels user guide. For the architectural design and listener loop internals, see Channels architecture.

Types and Enums

ChannelStatus

Connection status values for a channel. Returned by BaseChannel.status().

ChannelStatus

Bases: str, Enum

Channel connection status.

ChannelMessage

Dataclass representing a message received from or sent to a channel. All fields correspond to the JSON payload exchanged with the gateway.

ChannelMessage dataclass 

ChannelMessage(channel: str, sender: str, content: str, message_id: str = '', conversation_id: str = '', metadata: Dict[str, Any] = dict())

A message received from or sent to a channel.

ChannelHandler

Type alias for message handler callbacks:

ChannelHandler = Callable[[ChannelMessage], Optional[str]]

Handlers are called synchronously from the listener thread when a message arrives. The optional str return value is reserved for future auto-reply routing and has no effect in the current implementation.

Thread safety

Handlers run on the listener thread, not the caller thread. Protect shared state with locks or queue.Queue.

ChannelHandler module-attribute 

ChannelHandler = Callable[[ChannelMessage], Optional[str]]

BaseChannel

Abstract base class for all channel implementations. Subclasses must implement all six abstract methods and register via @ChannelRegistry.register("name").

custom_channel.py
from openjarvis.channels._stubs import BaseChannel, ChannelMessage, ChannelStatus
from openjarvis.core.registry import ChannelRegistry


@ChannelRegistry.register("my-channel")
class MyChannel(BaseChannel):
    channel_id = "my-channel"

    def connect(self) -> None: ...
    def disconnect(self) -> None: ...
    def send(self, channel, content, *, conversation_id="", metadata=None) -> bool: ...
    def status(self) -> ChannelStatus: ...
    def list_channels(self) -> list[str]: ...
    def on_message(self, handler) -> None: ...

BaseChannel

Bases: ABC

Base class for all channel implementations.

Subclasses must be registered via @ChannelRegistry.register("name") to become discoverable.

Functions

connect abstractmethod 
connect() -> None

Establish connection to the channel gateway.

Source code in src/openjarvis/channels/_stubs.py 
45
46
47
@abstractmethod
def connect(self) -> None:
    """Establish connection to the channel gateway."""
disconnect abstractmethod 
disconnect() -> None

Close connection to the channel gateway.

Source code in src/openjarvis/channels/_stubs.py 
49
50
51
@abstractmethod
def disconnect(self) -> None:
    """Close connection to the channel gateway."""
send abstractmethod 
send(channel: str, content: str, *, conversation_id: str = '', metadata: Dict[str, Any] | None = None) -> bool

Send a message to a specific channel. Returns True on success.

Source code in src/openjarvis/channels/_stubs.py 
53
54
55
56
57
58
59
60
61
62
@abstractmethod
def send(
    self,
    channel: str,
    content: str,
    *,
    conversation_id: str = "",
    metadata: Dict[str, Any] | None = None,
) -> bool:
    """Send a message to a specific channel. Returns True on success."""
status abstractmethod 
status() -> ChannelStatus

Return the current connection status.

Source code in src/openjarvis/channels/_stubs.py 
64
65
66
@abstractmethod
def status(self) -> ChannelStatus:
    """Return the current connection status."""
list_channels abstractmethod 
list_channels() -> List[str]

Return list of available channel names.

Source code in src/openjarvis/channels/_stubs.py 
68
69
70
@abstractmethod
def list_channels(self) -> List[str]:
    """Return list of available channel names."""
on_message abstractmethod 
on_message(handler: ChannelHandler) -> None

Register a callback for incoming messages.

Source code in src/openjarvis/channels/_stubs.py 
72
73
74
@abstractmethod
def on_message(self, handler: ChannelHandler) -> None:
    """Register a callback for incoming messages."""

OpenClawChannelBridge

OpenClawChannelBridge connects to the OpenClaw gateway over WebSocket, with automatic HTTP fallback when the websockets package is not installed or a WebSocket send fails. It is registered as "openclaw" in ChannelRegistry.

bridge_example.py
from openjarvis.channels.openclaw_bridge import OpenClawChannelBridge
from openjarvis.channels._stubs import ChannelMessage
from openjarvis.core.events import EventBus

bus = EventBus()
bridge = OpenClawChannelBridge(
    gateway_url="ws://127.0.0.1:18789/ws",
    reconnect_interval=5.0,
    bus=bus,
)


def on_message(msg: ChannelMessage) -> None:
    print(f"[{msg.channel}] {msg.sender}: {msg.content}")


bridge.on_message(on_message)
bridge.connect()

bridge.send("notifications", "Hello!")
channels = bridge.list_channels()

bridge.disconnect()

Constructor Parameters

Parameter Type Default Description
gateway_url str ws://127.0.0.1:18789/ws WebSocket URL of the OpenClaw gateway
reconnect_interval float 5.0 Seconds to wait before reconnecting after a disconnect
bus EventBus None Event bus for publishing CHANNEL_MESSAGE_RECEIVED and CHANNEL_MESSAGE_SENT events

Events Published

Event When
CHANNEL_MESSAGE_RECEIVED Message received from gateway WebSocket
CHANNEL_MESSAGE_SENT Message successfully delivered via WebSocket or HTTP

Optional dependency

OpenClawChannelBridge requires the openjarvis[openclaw] extra. See the Channels architecture for design details.