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 hashlib 

10import hmac 

11import json 

12import time 

13 

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

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

16 

17 

18class SlackAdapter(BaseChannelAdapter): 

19 """Slack Events API adapter. 

20 

21 Config fields: 

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

23 signing_secret: Slack Signing Secret for request verification 

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

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

26 """ 

27 

28 CHANNEL_TYPE = ChannelType.SLACK 

29 

30 def __init__(self, config: ChannelConfig): 

31 super().__init__(config) 

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

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

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

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

36 

37 # ── Webhook verification ── 

38 

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

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

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

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

43 

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

45 return False 

46 

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

48 computed = ( 

49 "v0=" 

50 + hmac.new( 

51 self._signing_secret.encode(), 

52 sig_basestring.encode(), 

53 hashlib.sha256, 

54 ).hexdigest() 

55 ) 

56 

57 return hmac.compare_digest(computed, slack_sig) 

58 

59 # ── Message parsing ── 

60 

61 async def parse_incoming(self, payload: dict) -> ChannelMessage | None: 

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

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

64 

65 # URL verification challenge 

66 if event_type == "url_verification": 

67 return ChannelMessage( 

68 channel_type=ChannelType.SLACK, 

69 channel_id=self.config.channel_id, 

70 user_id="system", 

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

72 message_type=MessageType.SYSTEM, 

73 raw=payload, 

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

75 ) 

76 

77 # Event callback 

78 if event_type == "event_callback": 

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

80 return self._parse_event(event) 

81 

82 return None 

83 

84 def _parse_event(self, event: dict) -> ChannelMessage | None: 

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

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

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

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

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

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

91 

92 # Strip bot mention prefix 

93 if event_type == "app_mention" and text: 

94 text = self._strip_mention(text) 

95 

96 if not text: 

97 return None 

98 

99 msg_type = MessageType.TEXT 

100 return ChannelMessage( 

101 channel_type=ChannelType.SLACK, 

102 channel_id=channel, 

103 user_id=user, 

104 content=text, 

105 message_type=msg_type, 

106 raw=event, 

107 reply_token=ts, 

108 ) 

109 

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

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

112 import re 

113 

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

115 

116 # ── Reply ── 

117 

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

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

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

121 headers = { 

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

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

124 } 

125 body = { 

126 "channel": channel_id, 

127 "text": content, 

128 } 

129 

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

131 if thread_ts: 

132 body["thread_ts"] = thread_ts 

133 

134 try: 

135 import aiohttp 

136 

137 async with aiohttp.ClientSession() as session: 

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

139 data = await resp.json() 

140 if data.get("ok"): 

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

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

143 except ImportError: 

144 import urllib.request 

145 

146 req = urllib.request.Request(url, data=json.dumps(body).encode(), headers=headers) 

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

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

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

150 

151 async def reply_blocks(self, channel_id: str, blocks: list[dict], **kwargs) -> ReplyResult: 

152 """Send Slack Block Kit message.""" 

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

154 headers = { 

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

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

157 } 

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

159 

160 try: 

161 import aiohttp 

162 

163 async with aiohttp.ClientSession() as session: 

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

165 data = await resp.json() 

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

167 except ImportError: 

168 import urllib.request 

169 

170 req = urllib.request.Request(url, data=json.dumps(body).encode(), headers=headers) 

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

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

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