""" Rocket.Chat Platform Plugin for Hermes Agent. Registers a gateway platform adapter that connects to a self-hosted Rocket.Chat server via DDP/WebSocket (real-time incoming messages + presence) and REST API (outgoing replies). """ import logging import os logger = logging.getLogger(__name__) # ── Interaction Templates (source of truth — zero yaml dependency) ── _TEMPLATE_MAP = { "yes_no": { "label": "Ja/Nein", "description": "Binary confirmation", "attachment": {"color": "#1d74f5", "title": "Bitte bestätigen:"}, "buttons": [ {"text": "✅ Ja", "msg": "ja", "style": "primary"}, {"text": "❌ Nein", "msg": "nein", "style": "danger"}, ], }, "confirm_cancel": { "label": "Bestätigen/Abbrechen", "description": "Safe action with abort", "attachment": {"color": "#1d74f5", "title": "Aktion bestätigen:"}, "buttons": [ {"text": "✅ Bestätigen", "msg": "bestätigen", "style": "primary"}, {"text": "🚫 Abbrechen", "msg": "abbrechen", "style": "danger"}, ], }, "ok": { "label": "OK", "description": "Single acknowledgement", "attachment": {"color": "#1d74f5", "title": "Hinweis:"}, "buttons": [ {"text": "👍 OK", "msg": "ok", "style": "primary"}, ], }, "multi_choice_3": { "label": "3 Optionen", "description": "Three-option selection", "attachment": {"color": "#1d74f5", "title": "Wähle eine Option:"}, "buttons": [ {"text": "1._Links", "msg": "links", "style": "default"}, {"text": "2._Rechts", "msg": "rechts", "style": "default"}, {"text": "3._Zwei_Runden", "msg": "zwei_runden", "style": "default"}, ], }, } def _build_custom_buttons(labels: list, prefix: str = "") -> list: """Build Rocket.Chat action buttons from plain label strings. Rocket.Chat UI truncates button text after ~11-12 chars. Use short labels (emoji+text without spaces, or numbered refs). """ buttons = [] for idx, label in enumerate(labels): # Sanitize label for button text (keep short) display = label.strip() # Payload may carry question_id prefix for correlation msg_payload = display if prefix: msg_payload = f"{prefix}::{msg_payload}" buttons.append({ "type": "button", "text": display, "msg": msg_payload, "msg_in_chat_window": True, "style": "default", }) return buttons def check_requirements(): """Rocket.Chat adapter uses only stdlib (urllib, socket, ssl, threading).""" return True def validate_config(config) -> bool: """Check whether the platform is properly configured.""" extra = getattr(config, "extra", {}) or {} base_url = os.getenv("ROCKETCHAT_BASE_URL") or extra.get("base_url", "") user = os.getenv("ROCKETCHAT_USER") or extra.get("user", "") password = os.getenv("ROCKETCHAT_PASSWORD") or extra.get("password", "") return bool(base_url and user and password) def is_connected(config) -> bool: """Check whether Rocket.Chat is configured (env or config.yaml).""" return validate_config(config) def register(ctx): """Plugin entry point — called by the Hermes plugin system at startup.""" from .adapter import RocketChatAdapter # lazy import avoids circular deps ctx.register_platform( name="rocketchat", label="Rocket.Chat", adapter_factory=lambda cfg: RocketChatAdapter(cfg), check_fn=check_requirements, validate_config=validate_config, is_connected=is_connected, required_env=["ROCKETCHAT_BASE_URL", "ROCKETCHAT_USER", "ROCKETCHAT_PASSWORD"], install_hint="No extra packages needed (stdlib only)", allowed_users_env="ROCKETCHAT_ALLOWED_USERS", allow_all_env="ROCKETCHAT_ALLOW_ALL_USERS", max_message_length=4000, emoji="🚀", pii_safe=False, allow_update_command=True, platform_hint=( "You are chatting via Rocket.Chat. You support markdown formatting: " "**bold**, *italic*, `inline code`, ```code blocks```, and [links](url). " "Messages can be up to ~4000 characters. You have a green online dot " "via DDP presence. Use professional but friendly tone." ), )