rocketchat_channel

rocketchat_channel

RocketChatChannel — Rocket.Chat adapter via rocketchat_API.

Classes

RocketChatChannel

RocketChatChannel(
    url: str = "",
    *,
    user: str = "",
    password: str = "",
    auth_token: str = "",
    user_id: str = "",
    bus: Optional[EventBus] = None,
)

Bases: BaseChannel

Rocket.Chat messaging channel adapter.

Uses the Rocket.Chat REST API via rocketchat_API.

PARAMETER	DESCRIPTION
url	Rocket.Chat server URL. Falls back to ROCKETCHAT_URL env var. TYPE: str DEFAULT: ''
user	Rocket.Chat username. Falls back to ROCKETCHAT_USER env var. TYPE: str DEFAULT: ''
password	Rocket.Chat password. Falls back to ROCKETCHAT_PASSWORD env var. TYPE: str DEFAULT: ''
auth_token	Rocket.Chat auth token (alternative to password). Falls back to ROCKETCHAT_AUTH_TOKEN env var. TYPE: str DEFAULT: ''
user_id	Rocket.Chat user ID (used with auth_token). Falls back to ROCKETCHAT_USER_ID env var. TYPE: str DEFAULT: ''
bus	Optional event bus for publishing channel events. TYPE: Optional[EventBus] DEFAULT: None

Source code in src/diapason/channels/rocketchat_channel.py

def __init__(
    self,
    url: str = "",
    *,
    user: str = "",
    password: str = "",
    auth_token: str = "",
    user_id: str = "",
    bus: Optional[EventBus] = None,
) -> None:
    self._url = url or os.environ.get("ROCKETCHAT_URL", "")
    self._user = user or os.environ.get("ROCKETCHAT_USER", "")
    self._password = password or os.environ.get("ROCKETCHAT_PASSWORD", "")
    self._auth_token = auth_token or os.environ.get("ROCKETCHAT_AUTH_TOKEN", "")
    self._user_id = user_id or os.environ.get("ROCKETCHAT_USER_ID", "")
    self._bus = bus
    self._handlers: List[ChannelHandler] = []
    self._status = ChannelStatus.DISCONNECTED

Methods:

connect

connect() -> None

Validate credentials and mark as connected.

Source code in src/diapason/channels/rocketchat_channel.py

def connect(self) -> None:
    """Validate credentials and mark as connected."""
    if not self._url:
        logger.warning("No Rocket.Chat URL configured")
        self._status = ChannelStatus.ERROR
        return
    has_password_auth = self._user and self._password
    has_token_auth = self._auth_token and self._user_id
    if not has_password_auth and not has_token_auth:
        logger.warning("No Rocket.Chat credentials configured")
        self._status = ChannelStatus.ERROR
        return
    try:
        import rocketchat_API  # noqa: F401
    except ImportError:
        raise ImportError(
            "rocketchat_API not installed. Install with: "
            "uv sync --extra channel-rocketchat"
        )
    self._status = ChannelStatus.CONNECTED

disconnect

disconnect() -> None

Mark as disconnected.

Source code in src/diapason/channels/rocketchat_channel.py

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

Send a message to a Rocket.Chat channel or DM.

PARAMETER	DESCRIPTION
channel	Rocket.Chat channel name or room ID. TYPE: str
content	Text message content. TYPE: str

Source code in src/diapason/channels/rocketchat_channel.py

def send(
    self,
    channel: str,
    content: str,
    *,
    conversation_id: str = "",
    metadata: Dict[str, Any] | None = None,
) -> bool:
    """Send a message to a Rocket.Chat channel or DM.

    Parameters
    ----------
    channel:
        Rocket.Chat channel name or room ID.
    content:
        Text message content.
    """
    if not self._url:
        logger.warning("Cannot send: no Rocket.Chat URL configured")
        return False

    try:
        from rocketchat_API import RocketChat

        if self._auth_token and self._user_id:
            rocket = RocketChat(
                server_url=self._url,
                auth_token=self._auth_token,
                user_id=self._user_id,
            )
        else:
            rocket = RocketChat(
                user=self._user,
                password=self._password,
                server_url=self._url,
            )

        rocket.chat_post_message(content, channel=channel)

        self._publish_sent(channel, content, conversation_id)
        return True
    except ImportError:
        logger.debug("rocketchat_API not installed")
        return False
    except Exception:
        logger.debug("Rocket.Chat send failed", exc_info=True)
        return False

status

status() -> ChannelStatus

Return the current connection status.

Source code in src/diapason/channels/rocketchat_channel.py

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/diapason/channels/rocketchat_channel.py

def list_channels(self) -> List[str]:
    """Return available channel identifiers."""
    return ["rocketchat"]

on_message

on_message(handler: ChannelHandler) -> None

Register a callback for incoming messages.

Source code in src/diapason/channels/rocketchat_channel.py

def on_message(self, handler: ChannelHandler) -> None:
    """Register a callback for incoming messages."""
    self._handlers.append(handler)