Skip to content

slack

slack

SlackChannel — native Slack Web API adapter.

Classes

SlackChannel 

SlackChannel(
    bot_token: str = "",
    *,
    app_token: str = "",
    bus: Optional[EventBus] = None,
)

Bases: BaseChannel

Native Slack channel adapter using the Slack Web API.

PARAMETER DESCRIPTION
bot_token

Slack Bot User OAuth Token. Falls back to SLACK_BOT_TOKEN env var.

TYPE: str DEFAULT: ''

app_token

Slack App-Level Token for Socket Mode. Falls back to SLACK_APP_TOKEN env var.

TYPE: str DEFAULT: ''

bus

Optional event bus for publishing channel events.

TYPE: Optional[EventBus] DEFAULT: None

Source code in src/diapason/channels/slack.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
def __init__(
    self,
    bot_token: str = "",
    *,
    app_token: str = "",
    bus: Optional[EventBus] = None,
) -> None:
    self._token = bot_token or os.environ.get("SLACK_BOT_TOKEN", "")
    self._app_token = app_token or os.environ.get("SLACK_APP_TOKEN", "")
    self._bus = bus
    self._handlers: List[ChannelHandler] = []
    self._status = ChannelStatus.DISCONNECTED
    self._listener_thread: Optional[threading.Thread] = None
    self._stop_event = threading.Event()
Methods:
connect 
connect() -> None

Start listening for incoming messages via Slack Socket Mode.

Source code in src/diapason/channels/slack.py 
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
def connect(self) -> None:
    """Start listening for incoming messages via Slack Socket Mode."""
    if not self._token:
        logger.warning("No Slack bot token configured")
        self._status = ChannelStatus.ERROR
        return

    self._stop_event.clear()
    self._status = ChannelStatus.CONNECTING

    try:
        from slack_sdk.socket_mode import SocketModeClient  # noqa: F401

        if not self._app_token:
            logger.info("No app token for Socket Mode; send-only mode")
            self._status = ChannelStatus.CONNECTED
            return

        self._listener_thread = threading.Thread(
            target=self._socket_mode_loop,
            daemon=True,
        )
        self._listener_thread.start()
        self._status = ChannelStatus.CONNECTED
        logger.info("Slack channel connected (Socket Mode)")
    except ImportError:
        logger.info("slack-sdk not installed; send-only mode")
        self._status = ChannelStatus.CONNECTED
disconnect 
disconnect() -> None

Stop the listener thread.

Source code in src/diapason/channels/slack.py 
85
86
87
88
89
90
91
def disconnect(self) -> None:
    """Stop the listener thread."""
    self._stop_event.set()
    if self._listener_thread is not None:
        self._listener_thread.join(timeout=5.0)
        self._listener_thread = None
    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 Slack channel via the Web API.

Source code in src/diapason/channels/slack.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
def send(
    self,
    channel: str,
    content: str,
    *,
    conversation_id: str = "",
    metadata: Dict[str, Any] | None = None,
) -> bool:
    """Send a message to a Slack channel via the Web API."""
    if not self._token:
        logger.warning("Cannot send: no Slack bot token")
        return False

    try:
        import httpx

        url = "https://slack.com/api/chat.postMessage"
        headers = {
            "Authorization": f"Bearer {self._token}",
            "Content-Type": "application/json",
        }
        payload: Dict[str, Any] = {
            "channel": channel,
            "text": content,
        }
        if conversation_id:
            payload["thread_ts"] = conversation_id

        resp = httpx.post(
            url,
            json=payload,
            headers=headers,
            timeout=10.0,
        )
        if resp.status_code < 300:
            data = resp.json()
            if data.get("ok"):
                self._publish_sent(channel, content, conversation_id)
                return True
            logger.warning("Slack API error: %s", data.get("error"))
            return False
        logger.warning(
            "Slack API returned status %d",
            resp.status_code,
        )
        return False
    except Exception:
        logger.debug("Slack send failed", exc_info=True)
        return False
status 
status() -> ChannelStatus

Return the current connection status.

Source code in src/diapason/channels/slack.py 
145
146
147
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/slack.py 
149
150
151
def list_channels(self) -> List[str]:
    """Return available channel identifiers."""
    return ["slack"]
on_message 
on_message(handler: ChannelHandler) -> None

Register a callback for incoming messages.

Source code in src/diapason/channels/slack.py 
153
154
155
def on_message(self, handler: ChannelHandler) -> None:
    """Register a callback for incoming messages."""
    self._handlers.append(handler)