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

1""" 

2Telegram Channel Adapter — Telegram Bot API. 

3 

4BotFather token → long-polling / webhook → 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 TelegramAdapter(BaseChannelAdapter): 

16 """Telegram Bot API adapter. 

17 

18 Config fields: 

19 bot_token: Telegram bot token from @BotFather 

20 webhook_url: HTTPS webhook URL (leave empty for long-polling) 

21 allowed_chats: list of chat IDs (empty = all) 

22 parse_mode: "HTML" | "MarkdownV2" | None 

23 """ 

24 

25 CHANNEL_TYPE = ChannelType.TELEGRAM 

26 

27 # Telegram API base URL 

28 API_BASE = "https://api.telegram.org" 

29 

30 def __init__(self, config: ChannelConfig): 

31 super().__init__(config) 

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

33 self._webhook_url = config.extra.get("webhook_url", "") 

34 self._allowed_chats = config.extra.get("allowed_chats", []) 

35 self._parse_mode = config.extra.get("parse_mode", "") 

36 

37 @property 

38 def _api(self) -> str: 

39 return f"{self.API_BASE}/bot{self._bot_token}" 

40 

41 # ── Message parsing ── 

42 

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

44 """Parse Telegram Update into ChannelMessage.""" 

45 if "message" in payload: 

46 return self._parse_message(payload["message"]) 

47 elif "callback_query" in payload: 

48 return self._parse_callback(payload["callback_query"]) 

49 elif "inline_query" in payload: 

50 return self._parse_inline(payload["inline_query"]) 

51 return None 

52 

53 def _parse_message(self, msg: dict) -> ChannelMessage | None: 

54 """Parse a Telegram Message object.""" 

55 chat = msg.get("chat", {}) 

56 user = msg.get("from", {}) 

57 chat_id = str(chat.get("id", "")) 

58 user_id = str(user.get("id", "")) 

59 

60 if self._allowed_chats and chat_id not in self._allowed_chats: 

61 return None 

62 

63 content = "" 

64 msg_type = MessageType.TEXT 

65 metadata = { 

66 "chat_type": chat.get("type", "private"), 

67 "chat_title": chat.get("title", ""), 

68 "username": user.get("username", ""), 

69 "first_name": user.get("first_name", ""), 

70 } 

71 

72 if "text" in msg: 

73 content = msg["text"] 

74 elif "photo" in msg: 

75 content = msg.get("caption", "[Photo]") 

76 msg_type = MessageType.IMAGE 

77 metadata["file_id"] = msg["photo"][-1]["file_id"] 

78 elif "document" in msg: 

79 content = msg.get("caption", "[Document]") 

80 msg_type = MessageType.FILE 

81 metadata["file_id"] = msg["document"]["file_id"] 

82 metadata["file_name"] = msg["document"].get("file_name", "") 

83 elif "voice" in msg: 

84 content = "[Voice message]" 

85 msg_type = MessageType.VOICE 

86 elif "video" in msg: 

87 content = msg.get("caption", "[Video]") 

88 msg_type = MessageType.VIDEO 

89 elif "sticker" in msg: 

90 content = f"[Sticker: {msg['sticker'].get('emoji', '')}]" 

91 msg_type = MessageType.TEXT 

92 else: 

93 return None 

94 

95 return ChannelMessage( 

96 channel_type=ChannelType.TELEGRAM, 

97 channel_id=chat_id, 

98 user_id=user_id, 

99 content=content, 

100 message_type=msg_type, 

101 raw=msg, 

102 reply_token=str(msg.get("message_id", "")), 

103 metadata=metadata, 

104 ) 

105 

106 def _parse_callback(self, cb: dict) -> ChannelMessage | None: 

107 """Parse callback query (inline button press).""" 

108 user = cb.get("from", {}) 

109 msg = cb.get("message", {}) 

110 return ChannelMessage( 

111 channel_type=ChannelType.TELEGRAM, 

112 channel_id=str(msg.get("chat", {}).get("id", "")), 

113 user_id=str(user.get("id", "")), 

114 content=cb.get("data", ""), 

115 message_type=MessageType.INTERACTIVE, 

116 raw=cb, 

117 reply_token=cb.get("id", ""), 

118 ) 

119 

120 def _parse_inline(self, inline: dict) -> ChannelMessage | None: 

121 """Parse inline query.""" 

122 user = inline.get("from", {}) 

123 return ChannelMessage( 

124 channel_type=ChannelType.TELEGRAM, 

125 channel_id="inline", 

126 user_id=str(user.get("id", "")), 

127 content=inline.get("query", ""), 

128 message_type=MessageType.COMMAND, 

129 raw=inline, 

130 reply_token=inline.get("id", ""), 

131 ) 

132 

133 # ── Reply ── 

134 

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

136 """Send message via sendMessage.""" 

137 return await self._api_call( 

138 "sendMessage", 

139 { 

140 "chat_id": channel_id, 

141 "text": content, 

142 "parse_mode": self._parse_mode, 

143 "reply_to_message_id": kwargs.get("reply_token"), 

144 }, 

145 ) 

146 

147 async def reply_inline_keyboard( 

148 self, 

149 channel_id: str, 

150 text: str, 

151 buttons: list[list[dict]], 

152 **kwargs, 

153 ) -> ReplyResult: 

154 """Send message with inline keyboard buttons. 

155 

156 buttons = [[{"text": "Yes", "callback_data": "yes"}], ...] 

157 """ 

158 return await self._api_call( 

159 "sendMessage", 

160 { 

161 "chat_id": channel_id, 

162 "text": text, 

163 "parse_mode": self._parse_mode, 

164 "reply_markup": json.dumps({"inline_keyboard": buttons}), 

165 }, 

166 ) 

167 

168 async def reply_photo( 

169 self, channel_id: str, photo_url: str, caption: str = "", **kwargs 

170 ) -> ReplyResult: 

171 """Send a photo.""" 

172 return await self._api_call( 

173 "sendPhoto", 

174 { 

175 "chat_id": channel_id, 

176 "photo": photo_url, 

177 "caption": caption, 

178 }, 

179 ) 

180 

181 async def answer_callback(self, callback_query_id: str, text: str = "") -> ReplyResult: 

182 """Answer a callback query (dismiss loading spinner).""" 

183 return await self._api_call( 

184 "answerCallbackQuery", 

185 { 

186 "callback_query_id": callback_query_id, 

187 "text": text, 

188 }, 

189 ) 

190 

191 # ── Internal ── 

192 

193 async def _api_call(self, method: str, params: dict) -> ReplyResult: 

194 """Call Telegram Bot API.""" 

195 url = f"{self._api}/{method}" 

196 

197 try: 

198 import aiohttp 

199 

200 async with aiohttp.ClientSession() as session: 

201 async with session.post(url, json=params) as resp: 

202 data = await resp.json() 

203 if data.get("ok"): 

204 return ReplyResult( 

205 success=True, 

206 message_id=str(data.get("result", {}).get("message_id", "")), 

207 ) 

208 return ReplyResult( 

209 success=False, 

210 error=data.get("description", "unknown"), 

211 ) 

212 except ImportError: 

213 import urllib.request 

214 

215 req = urllib.request.Request( 

216 url, 

217 data=json.dumps(params).encode(), 

218 headers={"Content-Type": "application/json"}, 

219 ) 

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

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

222 return ReplyResult( 

223 success=data.get("ok", False), 

224 message_id=str(data.get("result", {}).get("message_id", "")), 

225 ) 

226 

227 # ── Webhook ── 

228 

229 async def set_webhook(self, url: str) -> bool: 

230 """Register webhook URL with Telegram.""" 

231 result = await self._api_call("setWebhook", {"url": url}) 

232 return result.success 

233 

234 async def delete_webhook(self) -> bool: 

235 """Remove webhook (switch to long-polling).""" 

236 result = await self._api_call("deleteWebhook", {}) 

237 return result.success