Skip to content

webchat

webchat

WebChatChannel — in-memory message queue for webchat/testing.

Classes

WebChatChannel 

WebChatChannel(*, bus: Optional[EventBus] = None)

Bases: BaseChannel

In-memory webchat channel for testing and embedded web UIs.

Messages are stored in an internal list and can be retrieved via :meth:get_messages. No external dependencies are required.

PARAMETER DESCRIPTION
bus

Optional event bus for publishing channel events.

TYPE: Optional[EventBus] DEFAULT: None

Source code in src/openjarvis/channels/webchat.py 
35
36
37
38
39
40
41
42
43
def __init__(
    self,
    *,
    bus: Optional[EventBus] = None,
) -> None:
    self._bus = bus
    self._handlers: List[ChannelHandler] = []
    self._messages: List[ChannelMessage] = []
    self._status = ChannelStatus.DISCONNECTED
Functions
connect 
connect() -> None

Mark as connected (always succeeds for in-memory channel).

Source code in src/openjarvis/channels/webchat.py 
47
48
49
def connect(self) -> None:
    """Mark as connected (always succeeds for in-memory channel)."""
    self._status = ChannelStatus.CONNECTED
disconnect 
disconnect() -> None

Mark as disconnected.

Source code in src/openjarvis/channels/webchat.py 
51
52
53
def disconnect(self) -> None:
    """Mark as disconnected."""
    self._status = ChannelStatus.DISCONNECTED
send 
send(channel: str, content: str, *, conversation_id: str = '', metadata: Dict[str, Any] | None = None) -> bool

Append a message to the in-memory queue.

Source code in src/openjarvis/channels/webchat.py 
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
def send(
    self,
    channel: str,
    content: str,
    *,
    conversation_id: str = "",
    metadata: Dict[str, Any] | None = None,
) -> bool:
    """Append a message to the in-memory queue."""
    msg = ChannelMessage(
        channel=channel,
        sender="jarvis",
        content=content,
        conversation_id=conversation_id,
        metadata=metadata or {},
    )
    self._messages.append(msg)
    self._publish_sent(channel, content, conversation_id)
    return True
status 
status() -> ChannelStatus

Return the current connection status.

Source code in src/openjarvis/channels/webchat.py 
77
78
79
def status(self) -> ChannelStatus:
    """Return the current connection status."""
    return self._status
list_channels 
list_channels() -> List[str]

Return available channel identifiers.

Source code in src/openjarvis/channels/webchat.py 
81
82
83
def list_channels(self) -> List[str]:
    """Return available channel identifiers."""
    return ["webchat"]
on_message 
on_message(handler: ChannelHandler) -> None

Register a callback for incoming messages.

Source code in src/openjarvis/channels/webchat.py 
85
86
87
def on_message(self, handler: ChannelHandler) -> None:
    """Register a callback for incoming messages."""
    self._handlers.append(handler)
get_messages 
get_messages() -> List[ChannelMessage]

Return all messages stored in the in-memory queue.

Source code in src/openjarvis/channels/webchat.py 
91
92
93
def get_messages(self) -> List[ChannelMessage]:
    """Return all messages stored in the in-memory queue."""
    return self._messages
clear_messages 
clear_messages() -> None

Clear all messages from the in-memory queue.

Source code in src/openjarvis/channels/webchat.py 
95
96
97
def clear_messages(self) -> None:
    """Clear all messages from the in-memory queue."""
    self._messages.clear()