Coverage for agentos/channels/adapters/slack.py: 0%

1""" 

2Slack Channel Adapter — Slack Events API (Bolt for Python). 

3 

4OAuth 2.0 flow → Bolt App → Socket Mode / HTTP Webhook → ChannelMessage. 

5""" 

6 

7from __future__ import annotations 

8 

9import json 

10import hmac 

11import hashlib 

12import time 

13from typing import Optional 

14 

15from agentos.channels.base import BaseChannelAdapter, ChannelConfig, ReplyResult 

16from agentos.channels.message import ChannelMessage, ChannelType, MessageType 

17 

18 

19class SlackAdapter(BaseChannelAdapter): 

20 """Slack Events API adapter. 

21 

22 Config fields: 

23 bot_token: Slack Bot User OAuth Token (xoxb-...) 

24 signing_secret: Slack Signing Secret for request verification 

25 app_token: Socket Mode app-level token (xapp-...) — optional 

26 socket_mode: bool — use Socket Mode instead of HTTP webhooks 

27 """ 

28 

29 CHANNEL_TYPE = ChannelType.SLACK 

30 

31 def __init__(self, config: ChannelConfig): 

32 super().__init__(config) 

33 self._bot_token = config.extra.get("bot_token", "") 

34 self._signing_secret = config.extra.get("signing_secret", "") 

35 self._app_token = config.extra.get("app_token", "") 

36 self._socket_mode = config.extra.get("socket_mode", False) 

37 

38 # ── Webhook verification ── 

39 

40 def verify_signature(self, body: bytes, headers: dict) -> bool: 

41 """Verify Slack request signature (HMAC-SHA256).""" 

42 timestamp = headers.get("x-slack-request-timestamp", "") 

43 slack_sig = headers.get("x-slack-signature", "") 

44 

45 if abs(time.time() - int(timestamp)) > 300: 

46 return False 

47 

48 sig_basestring = f"v0:{timestamp}:{body.decode()}" 

49 computed = "v0=" + hmac.new( 

50 self._signing_secret.encode(), 

51 sig_basestring.encode(), 

52 hashlib.sha256, 

53 ).hexdigest() 

54 

55 return hmac.compare_digest(computed, slack_sig) 

56 

57 # ── Message parsing ── 

58 

59 async def parse_incoming(self, payload: dict) -> Optional[ChannelMessage]: 

60 """Parse a Slack event payload into ChannelMessage.""" 

61 event_type = payload.get("type", "") 

62 

63 # URL verification challenge 

64 if event_type == "url_verification": 

65 return ChannelMessage( 

66 channel_type=ChannelType.SLACK, 

67 channel_id=self.config.channel_id, 

68 user_id="system", 

69 content=payload.get("challenge", ""), 

70 message_type=MessageType.SYSTEM, 

71 raw=payload, 

72 reply_token=payload.get("challenge"), 

73 ) 

74 

75 # Event callback 

76 if event_type == "event_callback": 

77 event = payload.get("event", {}) 

78 return self._parse_event(event) 

79 

80 return None 

81 

82 def _parse_event(self, event: dict) -> Optional[ChannelMessage]: 

83 """Parse a Slack event (message, app_mention, etc.).""" 

84 event_type = event.get("type", "") 

85 user = event.get("user", "") 

86 channel = event.get("channel", "") 

87 text = event.get("text", "") 

88 ts = event.get("ts", "") 

89 

90 # Strip bot mention prefix 

91 if event_type == "app_mention" and text: 

92 text = self._strip_mention(text) 

93 

94 if not text: 

95 return None 

96 

97 msg_type = MessageType.TEXT 

98 return ChannelMessage( 

99 channel_type=ChannelType.SLACK, 

100 channel_id=channel, 

101 user_id=user, 

102 content=text, 

103 message_type=msg_type, 

104 raw=event, 

105 reply_token=ts, 

106 ) 

107 

108 def _strip_mention(self, text: str) -> str: 

109 """Remove <@BOT_ID> prefix from message text.""" 

110 import re 

111 return re.sub(r"^<@U[A-Z0-9]+>\s*", "", text).strip() 

112 

113 # ── Reply ── 

114 

115 async def reply(self, channel_id: str, content: str, **kwargs) -> ReplyResult: 

116 """Send a message to Slack channel via chat.postMessage.""" 

117 url = "https://slack.com/api/chat.postMessage" 

118 headers = { 

119 "Authorization": f"Bearer {self._bot_token}", 

120 "Content-Type": "application/json", 

121 } 

122 body = { 

123 "channel": channel_id, 

124 "text": content, 

125 } 

126 

127 thread_ts = kwargs.get("thread_ts") or kwargs.get("reply_token") 

128 if thread_ts: 

129 body["thread_ts"] = thread_ts 

130 

131 try: 

132 import aiohttp 

133 async with aiohttp.ClientSession() as session: 

134 async with session.post(url, headers=headers, json=body) as resp: 

135 data = await resp.json() 

136 if data.get("ok"): 

137 return ReplyResult(success=True, message_id=data.get("ts", "")) 

138 return ReplyResult(success=False, error=data.get("error", "unknown")) 

139 except ImportError: 

140 import urllib.request 

141 req = urllib.request.Request( 

142 url, data=json.dumps(body).encode(), headers=headers 

143 ) 

144 with urllib.request.urlopen(req) as resp: 

145 data = json.loads(resp.read()) 

146 return ReplyResult(success=data.get("ok", False), message_id=data.get("ts", "")) 

147 

148 async def reply_blocks( 

149 self, channel_id: str, blocks: list[dict], **kwargs 

150 ) -> ReplyResult: 

151 """Send Slack Block Kit message.""" 

152 url = "https://slack.com/api/chat.postMessage" 

153 headers = { 

154 "Authorization": f"Bearer {self._bot_token}", 

155 "Content-Type": "application/json", 

156 } 

157 body = {"channel": channel_id, "blocks": blocks} 

158 

159 try: 

160 import aiohttp 

161 async with aiohttp.ClientSession() as session: 

162 async with session.post(url, headers=headers, json=body) as resp: 

163 data = await resp.json() 

164 return ReplyResult(success=data.get("ok", False), message_id=data.get("ts", "")) 

165 except ImportError: 

166 import urllib.request 

167 req = urllib.request.Request( 

168 url, data=json.dumps(body).encode(), headers=headers 

169 ) 

170 with urllib.request.urlopen(req) as resp: 

171 data = json.loads(resp.read()) 

172 return ReplyResult(success=data.get("ok", False), message_id=data.get("ts", ""))