Skip to content

google_chat

google_chat

GoogleChatChannel — Google Chat webhook adapter.

Classes

GoogleChatChannel 

GoogleChatChannel(
    webhook_url: str = "", *, bus: Optional[EventBus] = None
)

Bases: BaseChannel

Google Chat webhook channel adapter (send-only).

PARAMETER DESCRIPTION
webhook_url

Google Chat incoming webhook URL. Falls back to GOOGLE_CHAT_WEBHOOK_URL env var.

TYPE: str DEFAULT: ''

bus

Optional event bus for publishing channel events.

TYPE: Optional[EventBus] DEFAULT: None

Source code in src/diapason/channels/google_chat.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
def __init__(
    self,
    webhook_url: str = "",
    *,
    bus: Optional[EventBus] = None,
) -> None:
    self._webhook_url = webhook_url or os.environ.get(
        "GOOGLE_CHAT_WEBHOOK_URL",
        "",
    )
    self._bus = bus
    self._handlers: List[ChannelHandler] = []
    self._status = ChannelStatus.DISCONNECTED
Methods:
connect 
connect() -> None

Mark as connected (send-only — no persistent connection).

Source code in src/diapason/channels/google_chat.py 
51
52
53
54
55
56
57
def connect(self) -> None:
    """Mark as connected (send-only — no persistent connection)."""
    if not self._webhook_url:
        logger.warning("No Google Chat webhook URL configured")
        self._status = ChannelStatus.ERROR
        return
    self._status = ChannelStatus.CONNECTED
disconnect 
disconnect() -> None

Mark as disconnected.

Source code in src/diapason/channels/google_chat.py 
59
60
61
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 Google Chat via the configured webhook URL.

Source code in src/diapason/channels/google_chat.py 
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
def send(
    self,
    channel: str,
    content: str,
    *,
    conversation_id: str = "",
    metadata: Dict[str, Any] | None = None,
) -> bool:
    """Send a message to Google Chat via the configured webhook URL."""
    if not self._webhook_url:
        logger.warning("Cannot send: no Google Chat webhook URL configured")
        return False

    try:
        import httpx

        payload: Dict[str, Any] = {"text": content}

        resp = httpx.post(
            self._webhook_url,
            json=payload,
            timeout=10.0,
        )
        if resp.status_code < 300:
            self._publish_sent(channel, content, conversation_id)
            return True
        logger.warning(
            "Google Chat API returned status %d",
            resp.status_code,
        )
        return False
    except Exception:
        logger.debug("Google Chat send failed", exc_info=True)
        return False
status 
status() -> ChannelStatus

Return the current connection status.

Source code in src/diapason/channels/google_chat.py 
100
101
102
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/google_chat.py 
104
105
106
def list_channels(self) -> List[str]:
    """Return available channel identifiers."""
    return ["google_chat"]
on_message 
on_message(handler: ChannelHandler) -> None

Register a callback for incoming messages.

Source code in src/diapason/channels/google_chat.py 
108
109
110
def on_message(self, handler: ChannelHandler) -> None:
    """Register a callback for incoming messages."""
    self._handlers.append(handler)