Skip to content

Index

channels

Channel abstraction for multi-platform messaging.

Classes

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 
46
47
48
@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 
50
51
52
@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 
54
55
56
57
58
59
60
61
62
63
@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 
65
66
67
@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 
69
70
71
@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 
73
74
75
@abstractmethod
def on_message(self, handler: ChannelHandler) -> None:
    """Register a callback for incoming messages."""

ChannelMessage dataclass 

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

A message received from or sent to a channel.

ChannelStatus

Bases: str, Enum

Channel connection status.