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

1""" 

2Discord Channel Adapter — Discord.py Gateway. 

3 

4Bot Token + Intents → Gateway connection → on_message → ChannelMessage. 

5""" 

6 

7from __future__ import annotations 

8 

9import json 

10 

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

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

13 

14 

15class DiscordAdapter(BaseChannelAdapter): 

16 """Discord Bot adapter (Gateway Intents). 

17 

18 Config fields: 

19 bot_token: Discord bot token 

20 guild_ids: list of guild/servers to monitor (empty = all) 

21 command_prefix: bot command prefix (default "!") 

22 dm_enabled: allow DM messages (default True) 

23 """ 

24 

25 CHANNEL_TYPE = ChannelType.DISCORD 

26 

27 def __init__(self, config: ChannelConfig): 

28 super().__init__(config) 

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

30 self._guild_ids = config.extra.get("guild_ids", []) 

31 self._command_prefix = config.extra.get("command_prefix", "!") 

32 self._dm_enabled = config.extra.get("dm_enabled", True) 

33 

34 # ── Message parsing ── 

35 

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

37 """Parse Discord gateway message event into ChannelMessage.""" 

38 event_type = payload.get("t", "") # Gateway event type 

39 data = payload.get("d", {}) 

40 

41 if event_type == "MESSAGE_CREATE": 

42 return self._parse_message(data) 

43 elif event_type == "INTERACTION_CREATE": 

44 return self._parse_interaction(data) 

45 elif event_type == "READY": 

46 return ChannelMessage( 

47 channel_type=ChannelType.DISCORD, 

48 channel_id="system", 

49 user_id="system", 

50 content=f"Bot ready (guilds: {len(data.get('guilds', []))})", 

51 message_type=MessageType.SYSTEM, 

52 raw=payload, 

53 ) 

54 

55 return None 

56 

57 def _parse_message(self, data: dict) -> ChannelMessage | None: 

58 """Parse a Discord Message Create event.""" 

59 author = data.get("author", {}) 

60 if author.get("bot", False): 

61 return None # Ignore other bots 

62 

63 content = data.get("content", "") 

64 if not content.strip(): 

65 return None 

66 

67 user_id = author.get("id", "") 

68 channel_id = data.get("channel_id", "") 

69 guild_id = data.get("guild_id", "") 

70 

71 # DM check 

72 if not guild_id and not self._dm_enabled: 

73 return None 

74 

75 # Guild filter 

76 if guild_id and self._guild_ids and guild_id not in self._guild_ids: 

77 return None 

78 

79 # Strip command prefix 

80 stripped = content 

81 if content.startswith(self._command_prefix): 

82 stripped = content[len(self._command_prefix) :] 

83 msg_type = MessageType.COMMAND 

84 else: 

85 msg_type = MessageType.TEXT 

86 

87 return ChannelMessage( 

88 channel_type=ChannelType.DISCORD, 

89 channel_id=channel_id, 

90 user_id=user_id, 

91 content=stripped.strip(), 

92 message_type=msg_type, 

93 raw=data, 

94 reply_token=data.get("id", ""), 

95 metadata={ 

96 "guild_id": guild_id, 

97 "username": author.get("username", ""), 

98 "display_name": data.get("member", {}).get("nick", author.get("username", "")), 

99 "attachments": [a.get("url") for a in data.get("attachments", [])], 

100 }, 

101 ) 

102 

103 def _parse_interaction(self, data: dict) -> ChannelMessage | None: 

104 """Parse Discord slash command interaction.""" 

105 interaction_data = data.get("data", {}) 

106 command_name = interaction_data.get("name", "") 

107 

108 user = data.get("user", {}) or data.get("member", {}).get("user", {}) 

109 user_id = user.get("id", "") 

110 channel_id = data.get("channel_id", "") 

111 

112 return ChannelMessage( 

113 channel_type=ChannelType.DISCORD, 

114 channel_id=channel_id, 

115 user_id=user_id, 

116 content=f"/{command_name} " 

117 + " ".join( 

118 f"{o.get('name')}:{o.get('value')}" for o in interaction_data.get("options", []) 

119 ), 

120 message_type=MessageType.COMMAND, 

121 raw=data, 

122 reply_token=data.get("token", ""), 

123 ) 

124 

125 # ── Reply ── 

126 

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

128 """Send message to Discord channel via REST API.""" 

129 url = f"https://discord.com/api/v10/channels/{channel_id}/messages" 

130 headers = { 

131 "Authorization": f"Bot {self._bot_token}", 

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

133 } 

134 body = {"content": content[:2000]} 

135 

136 if kwargs.get("embed"): 

137 body["embeds"] = [kwargs["embed"]] 

138 

139 # Interaction follow-up 

140 interaction_token = kwargs.get("interaction_token") or kwargs.get("reply_token") 

141 if interaction_token: 

142 url = f"https://discord.com/api/v10/webhooks/{self._bot_token}/{interaction_token}" 

143 

144 try: 

145 import aiohttp 

146 

147 async with aiohttp.ClientSession() as session: 

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

149 data = await resp.json() 

150 return ReplyResult(success=True, message_id=data.get("id", "")) 

151 except ImportError: 

152 import urllib.request 

153 

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

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

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

157 return ReplyResult(success=True, message_id=data.get("id", "")) 

158 

159 async def reply_embed( 

160 self, 

161 channel_id: str, 

162 title: str, 

163 description: str, 

164 color: int = 0x5865F2, 

165 fields: list = None, 

166 **kwargs, 

167 ) -> ReplyResult: 

168 """Send a Discord embed message.""" 

169 embed = { 

170 "title": title, 

171 "description": description, 

172 "color": color, 

173 } 

174 if fields: 

175 embed["fields"] = fields 

176 return await self.reply(channel_id, "", embed=embed, **kwargs)