@samitouri / QOSami_Agents_Zero / commits / 83ffa27d

feat: add Telegram integration plugin - core backend

keyboardstaff committed Mar 20, 2026 at 10:12 UTC 83ffa27d139c005cf82b9c8c5cbdd60380b8b898
16 files changed +1087
plugins/_telegram_integration/api/test_connection.py new
+35
@@ -0,0 +1,35 @@
1 +from helpers.api import ApiHandler, Request
2 +from helpers.errors import format_error
3 +
4 +
5 +class TestConnection(ApiHandler):
6 +
7 + async def process(self, input: dict, request: Request) -> dict:
8 + bot_cfg = input.get("bot", {})
9 + token = bot_cfg.get("token", "")
10 + results: list[dict] = []
11 +
12 + if not token:
13 + results.append({
14 + "test": "Token",
15 + "ok": False,
16 + "message": "No bot token provided",
17 + })
18 + return {"success": False, "results": results}
19 +
20 + try:
21 + from plugins._telegram_integration.helpers.bot_manager import test_token
22 + ok, message = await test_token(token)
23 + results.append({
24 + "test": "Bot Token",
25 + "ok": ok,
26 + "message": message,
27 + })
28 + except Exception as e:
29 + results.append({
30 + "test": "Bot Token",
31 + "ok": False,
32 + "message": format_error(e),
33 + })
34 +
35 + return {"success": all(r["ok"] for r in results), "results": results}
plugins/_telegram_integration/default_config.yaml new
+15
@@ -0,0 +1,15 @@
1 +bots: []
2 +# Example bot:
3 +# - name: my_bot
4 +# enabled: true
5 +# token: ""
6 +# mode: polling # polling or webhook
7 +# webhook_url: "" # required if mode=webhook, e.g. https://yourdomain.com/telegram/my_bot
8 +# webhook_secret: "" # optional shared secret for webhook verification
9 +# allowed_users: [] # Telegram user IDs or @usernames. Empty = allow all.
10 +# group_mode: mention # mention (respond when @mentioned or replied to) | all (respond to every message) | off (ignore groups)
11 +# user_projects: {} # Map Telegram user_id to project name, e.g. { "123456": "my_project" }
12 +# default_project: "" # Fallback project if user not in user_projects
13 +# attachment_max_age_hours: 0 # Hours to keep downloaded attachments. 0 = keep forever.
14 +# agent_instructions: "" # Extra instructions for the agent in Telegram chats
15 +
plugins/_telegram_integration/helpers/__init__.py
plugins/_telegram_integration/helpers/bot_manager.py new
+221
@@ -0,0 +1,221 @@
1 +import asyncio
2 +from dataclasses import dataclass, field
3 +from typing import Callable, Awaitable
4 +
5 +from aiogram import Bot, Dispatcher, Router, F
6 +from aiogram.client.default import DefaultBotProperties
7 +from aiogram.enums import ParseMode, ChatType
8 +from aiogram.exceptions import TelegramBadRequest
9 +from aiogram.filters import Command, CommandStart
10 +from aiogram.types import Message, CallbackQuery, Update
11 +from aiogram.webhook.aiohttp_server import SimpleRequestHandler, setup_application
12 +
13 +from helpers.errors import format_error
14 +from helpers.print_style import PrintStyle
15 +
16 +# Data models
17 +
18 +@dataclass
19 +class BotInstance:
20 + name: str
21 + bot: Bot
22 + dispatcher: Dispatcher
23 + router: Router
24 + task: asyncio.Task | None = None # polling task
25 + webhook_app: object | None = None # aiohttp app for webhook mode
26 + bot_info: object | None = None # cached result of bot.get_me()
27 +
28 +# Bot registry (singleton, persists across module reloads)
29 +
30 +_bots: dict[str, BotInstance] = {}
31 +
32 +
33 +def get_bot(name: str) -> BotInstance | None:
34 + return _bots.get(name)
35 +
36 +
37 +def get_all_bots() -> dict[str, BotInstance]:
38 + return _bots
39 +
40 +# Bot creation
41 +
42 +def create_bot(
43 + name: str,
44 + token: str,
45 + on_message: Callable[..., Awaitable],
46 + on_command_start: Callable[..., Awaitable],
47 + on_command_clear: Callable[..., Awaitable],
48 + on_callback_query: Callable[..., Awaitable] | None = None,
49 + group_mode: str = "mention",
50 +) -> BotInstance:
51 + bot = Bot(token=token, default=DefaultBotProperties(parse_mode=ParseMode.MARKDOWN))
52 + dp = Dispatcher()
53 + router = Router()
54 +
55 + # Register command handlers
56 + router.message.register(on_command_start, CommandStart())
57 + router.message.register(on_command_clear, Command("clear"))
58 +
59 + if on_callback_query:
60 + router.callback_query.register(on_callback_query)
61 +
62 + # Register message handler with group filtering
63 + if group_mode == "off":
64 + # Private chats only
65 + router.message.register(
66 + on_message, F.chat.type == ChatType.PRIVATE,
67 + )
68 + elif group_mode == "mention":
69 + # Private chats: all messages; Groups: only when mentioned/replied
70 + router.message.register(
71 + on_message, F.chat.type == ChatType.PRIVATE,
72 + )
73 + router.message.register(
74 + _make_group_mention_filter(on_message, bot),
75 + )
76 + else:
77 + # All messages in all chats
78 + router.message.register(on_message)
79 +
80 + dp.include_router(router)
81 + instance = BotInstance(name=name, bot=bot, dispatcher=dp, router=router)
82 + _bots[name] = instance
83 + return instance
84 +
85 +
86 +async def cache_bot_info(instance: BotInstance):
87 + """Fetch and cache bot info. Call after create_bot."""
88 + if not instance.bot_info:
89 + instance.bot_info = await instance.bot.get_me()
90 + return instance.bot_info
91 +
92 +
93 +def _make_group_mention_filter(handler: Callable, bot: Bot):
94 + """Create a group message handler that only responds to mentions and replies."""
95 + async def _group_handler(message: Message):
96 + if message.chat.type == ChatType.PRIVATE:
97 + return
98 + # Use cached bot_info from the instance
99 + inst = get_bot(handler.__name__.split('_')[-1]) if hasattr(handler, '__name__') else None
100 + bot_info = None
101 + for b in _bots.values():
102 + if b.bot is bot:
103 + bot_info = b.bot_info
104 + break
105 + if not bot_info:
106 + bot_info = await bot.get_me()
107 + bot_username = bot_info.username or ""
108 +
109 + # Check for reply to bot
110 + if message.reply_to_message and message.reply_to_message.from_user:
111 + if message.reply_to_message.from_user.id == bot_info.id:
112 + await handler(message)
113 + return
114 +
115 + # Check for @mention in text
116 + if message.text and f"@{bot_username}" in message.text:
117 + await handler(message)
118 + return
119 +
120 + # Check entities for mention
121 + if message.entities:
122 + for entity in message.entities:
123 + if entity.type == "mention":
124 + mention_text = message.text[entity.offset:entity.offset + entity.length]
125 + if mention_text.lower() == f"@{bot_username.lower()}":
126 + await handler(message)
127 + return
128 +
129 + _group_handler.__name__ = f"_group_handler_{id(handler)}"
130 + return _group_handler
131 +
132 +# Polling
133 +
134 +async def start_polling(instance: BotInstance) -> asyncio.Task:
135 + async def _poll():
136 + try:
137 + PrintStyle.info(f"Telegram ({instance.name}): starting polling")
138 + await instance.dispatcher.start_polling(
139 + instance.bot,
140 + handle_signals=False,
141 + )
142 + except asyncio.CancelledError:
143 + PrintStyle.info(f"Telegram ({instance.name}): polling cancelled")
144 + except Exception as e:
145 + PrintStyle.error(f"Telegram ({instance.name}): polling error: {format_error(e)}")
146 +
147 + task = asyncio.create_task(_poll())
148 + instance.task = task
149 + return task
150 +
151 +
152 +async def stop_polling(instance: BotInstance):
153 + if instance.task and not instance.task.done():
154 + await instance.dispatcher.stop_polling()
155 + instance.task.cancel()
156 + try:
157 + await instance.task
158 + except asyncio.CancelledError:
159 + pass
160 + instance.task = None
161 +
162 +# Webhook
163 +
164 +async def setup_webhook(instance: BotInstance, url: str, secret: str = ""):
165 + """Set Telegram webhook and return aiohttp app for local serving."""
166 + from aiohttp import web
167 +
168 + await instance.bot.set_webhook(
169 + url=url,
170 + secret_token=secret or None,
171 + )
172 +
173 + app = web.Application()
174 + handler = SimpleRequestHandler(
175 + dispatcher=instance.dispatcher,
176 + bot=instance.bot,
177 + secret_token=secret or None,
178 + )
179 + handler.register(app, path=f"/telegram/{instance.name}")
180 + instance.webhook_app = app
181 + PrintStyle.info(f"Telegram ({instance.name}): webhook set to {url}")
182 + return app
183 +
184 +
185 +async def remove_webhook(instance: BotInstance):
186 + try:
187 + await instance.bot.delete_webhook()
188 + except Exception as e:
189 + PrintStyle.error(f"Telegram ({instance.name}): remove webhook error: {format_error(e)}")
190 +
191 +# Cleanup
192 +
193 +async def stop_bot(name: str):
194 + instance = _bots.pop(name, None)
195 + if not instance:
196 + return
197 + if instance.task and not instance.task.done():
198 + await stop_polling(instance)
199 + else:
200 + await remove_webhook(instance)
201 + try:
202 + await instance.bot.session.close()
203 + except Exception:
204 + pass
205 + PrintStyle.info(f"Telegram ({name}): stopped")
206 +
207 +
208 +async def stop_all_bots():
209 + for name in list(_bots.keys()):
210 + await stop_bot(name)
211 +
212 +# Test connection
213 +
214 +async def test_token(token: str) -> tuple[bool, str]:
215 + try:
216 + bot = Bot(token=token)
217 + info = await bot.get_me()
218 + await bot.session.close()
219 + return True, f"Connected as @{info.username} ({info.first_name})"
220 + except Exception as e:
221 + return False, format_error(e)
plugins/_telegram_integration/helpers/handler.py new
+517
@@ -0,0 +1,517 @@
1 +import json
2 +import os
3 +import threading
4 +import time
5 +import uuid
6 +
7 +from aiogram.types import Message as TgMessage, CallbackQuery
8 +
9 +from agent import Agent, AgentContext, AgentContextType, UserMessage
10 +from helpers import guids, plugins, files
11 +from helpers import message_queue as mq
12 +from helpers.notification import NotificationManager, NotificationType, NotificationPriority
13 +from helpers.persist_chat import save_tmp_chat
14 +from helpers.print_style import PrintStyle
15 +from helpers.errors import format_error
16 +from initialize import initialize_agent
17 +
18 +from plugins._telegram_integration.helpers import telegram_client as tc
19 +from plugins._telegram_integration.helpers.bot_manager import get_bot
20 +
21 +
22 +PLUGIN_NAME = "_telegram_integration"
23 +DOWNLOAD_FOLDER = "usr/telegram/attachments"
24 +STATE_FILE = "usr/telegram/state.json"
25 +
26 +# Context data keys
27 +CTX_TG_BOT = "telegram_bot"
28 +CTX_TG_CHAT_ID = "telegram_chat_id"
29 +CTX_TG_USER_ID = "telegram_user_id"
30 +CTX_TG_USERNAME = "telegram_username"
31 +CTX_TG_LAST_MSG_ID = "telegram_last_msg_id"
32 +
33 +# Transient
34 +CTX_TG_ATTACHMENTS = "_telegram_response_attachments"
35 +CTX_TG_KEYBOARD = "_telegram_response_keyboard"
36 +
37 +# Chat mapping: (bot_name, tg_user_id) → AgentContext ID
38 +
39 +_chat_map_lock = threading.Lock()
40 +
41 +
42 +def _load_state() -> dict:
43 + path = files.get_abs_path(STATE_FILE)
44 + if os.path.isfile(path):
45 + try:
46 + return json.loads(files.read_file(path))
47 + except Exception:
48 + return {}
49 + return {}
50 +
51 +
52 +def _save_state(state: dict):
53 + path = files.get_abs_path(STATE_FILE)
54 + files.make_dirs(path)
55 + files.write_file(path, json.dumps(state))
56 +
57 +
58 +def _map_key(bot_name: str, user_id: int) -> str:
59 + return f"{bot_name}:{user_id}"
60 +
61 +
62 +def cleanup_old_attachments():
63 + """Remove downloaded attachment files older than per-bot max age. 0 = keep forever."""
64 + config = plugins.get_plugin_config(PLUGIN_NAME) or {}
65 + bots_cfg = config.get("bots") or []
66 + total_removed = 0
67 + for bot_cfg in bots_cfg:
68 + bot_name = bot_cfg.get("name", "")
69 + if not bot_name:
70 + continue
71 + max_age_hours = bot_cfg.get("attachment_max_age_hours", 0)
72 + if not max_age_hours or max_age_hours <= 0:
73 + continue
74 + folder = os.path.join(files.get_abs_path(DOWNLOAD_FOLDER), bot_name)
75 + if not os.path.isdir(folder):
76 + continue
77 + cutoff = time.time() - max_age_hours * 3600
78 + for name in os.listdir(folder):
79 + path = os.path.join(folder, name)
80 + try:
81 + if os.path.isfile(path) and os.path.getmtime(path) < cutoff:
82 + os.remove(path)
83 + total_removed += 1
84 + except OSError:
85 + pass
86 + if total_removed:
87 + PrintStyle.info(f"Telegram: cleaned up {total_removed} old attachment(s)")
88 +
89 +# Access control
90 +
91 +def _is_allowed(bot_cfg: dict, user_id: int, username: str | None) -> bool:
92 + allowed = bot_cfg.get("allowed_users") or []
93 + if not allowed:
94 + return True # empty = allow all
95 + for entry in allowed:
96 + entry_str = str(entry).strip()
97 + if entry_str.startswith("@"):
98 + if username and f"@{username}" == entry_str:
99 + return True
100 + else:
101 + try:
102 + if int(entry_str) == user_id:
103 + return True
104 + except ValueError:
105 + if username and entry_str.lower() == username.lower():
106 + return True
107 + return False
108 +
109 +
110 +def _get_project(bot_cfg: dict, user_id: int) -> str:
111 + user_projects = bot_cfg.get("user_projects") or {}
112 + project = user_projects.get(str(user_id), "")
113 + if not project:
114 + project = bot_cfg.get("default_project", "")
115 + return project
116 +
117 +# Message handlers (registered with aiogram by bot_manager)
118 +
119 +async def handle_start(message: TgMessage, bot_name: str, bot_cfg: dict):
120 + """Handle /start command."""
121 + user = message.from_user
122 + if not user:
123 + return
124 +
125 + if not _is_allowed(bot_cfg, user.id, user.username):
126 + await message.reply("⛔ You are not authorized to use this bot.")
127 + return
128 +
129 + instance = get_bot(bot_name)
130 + if not instance:
131 + return
132 +
133 + await message.reply(
134 + f"👋 Hello {user.first_name}! I'm connected to Agent Zero.\n\n"
135 + "Send me a message and I'll process it.\n"
136 + "Use /clear to reset the conversation."
137 + )
138 +
139 + # Ensure a chat context exists
140 + await _get_or_create_context(bot_name, bot_cfg, message)
141 +
142 +
143 +async def handle_clear(message: TgMessage, bot_name: str, bot_cfg: dict):
144 + """Handle /clear command — reset user's chat context."""
145 + user = message.from_user
146 + if not user:
147 + return
148 +
149 + if not _is_allowed(bot_cfg, user.id, user.username):
150 + return
151 +
152 + key = _map_key(bot_name, user.id)
153 +
154 + with _chat_map_lock:
155 + state = _load_state()
156 + ctx_id = state.get("chats", {}).get(key)
157 + if ctx_id:
158 + ctx = AgentContext.get(ctx_id)
159 + if ctx:
160 + ctx.reset()
161 + PrintStyle.info(f"Telegram ({bot_name}): cleared chat for user {user.id}")
162 +
163 + instance = get_bot(bot_name)
164 + if instance:
165 + await tc.send_text(
166 + instance.bot, message.chat.id,
167 + "🗑 Chat cleared. Send a new message to start fresh.",
168 + parse_mode=None,
169 + )
170 +
171 + # Send notification
172 + username_str = f"@{user.username}" if user.username else str(user.id)
173 + NotificationManager.send_notification(
174 + type=NotificationType.INFO,
175 + priority=NotificationPriority.NORMAL,
176 + title="Telegram: chat cleared",
177 + message=f"{username_str} cleared their chat via /clear",
178 + display_time=5,
179 + group="telegram",
180 + )
181 +
182 +
183 +async def handle_message(message: TgMessage, bot_name: str, bot_cfg: dict):
184 + """Handle incoming user message."""
185 + user = message.from_user
186 + if not user:
187 + return
188 +
189 + if not _is_allowed(bot_cfg, user.id, user.username):
190 + return
191 +
192 + instance = get_bot(bot_name)
193 + if not instance:
194 + return
195 +
196 + # Send typing indicator
197 + await tc.send_typing(instance.bot, message.chat.id)
198 +
199 + # Get or create agent context
200 + context = await _get_or_create_context(bot_name, bot_cfg, message)
201 + if not context:
202 + await tc.send_text(
203 + instance.bot, message.chat.id,
204 + "❌ Failed to create chat session.",
205 + parse_mode=None,
206 + )
207 + return
208 +
209 + # Build user message text
210 + text = _extract_message_content(message)
211 + attachments = await _download_attachments(instance.bot, message, bot_name=bot_name)
212 +
213 + # Store last incoming message ID for reply threading
214 + context.data[CTX_TG_LAST_MSG_ID] = message.message_id
215 +
216 + # Build user message with prompt
217 + agent = context.agent0
218 + user_msg = agent.read_prompt(
219 + "fw.telegram.user_message.md",
220 + sender=_format_user(user),
221 + body=text,
222 + )
223 +
224 + instructions = bot_cfg.get("agent_instructions", "")
225 + if instructions:
226 + user_msg += agent.read_prompt(
227 + "fw.telegram.user_message_instructions.md",
228 + instructions=instructions,
229 + )
230 +
231 + system_ctx = agent.read_prompt("fw.telegram.system_context.md")
232 +
233 + mq.log_user_message(context, user_msg, attachments, source=" (telegram)")
234 + context.communicate(UserMessage(
235 + message=user_msg,
236 + system_message=[system_ctx],
237 + attachments=attachments,
238 + ))
239 +
240 + save_tmp_chat(context)
241 +
242 + # Send notification
243 + username_str = f"@{user.username}" if user.username else str(user.id)
244 + preview = (text[:80] + "...") if len(text) > 80 else text
245 + NotificationManager.send_notification(
246 + type=NotificationType.INFO,
247 + priority=NotificationPriority.HIGH,
248 + title="Telegram: new message",
249 + message=f"From {username_str}: {preview}",
250 + display_time=10,
251 + group="telegram",
252 + )
253 +
254 +
255 +async def handle_callback_query(query: CallbackQuery, bot_name: str, bot_cfg: dict):
256 + """Handle inline keyboard button press."""
257 + user = query.from_user
258 + if not user or not query.message:
259 + return
260 +
261 + if not _is_allowed(bot_cfg, user.id, user.username):
262 + await query.answer("Not authorized.")
263 + return
264 +
265 + await query.answer()
266 +
267 + # Treat callback data as a user message
268 + text = query.data or ""
269 + if not text:
270 + return
271 +
272 + context = await _get_or_create_context_from_user(
273 + bot_name, bot_cfg, user.id, user.username, query.message.chat.id,
274 + )
275 + if not context:
276 + return
277 +
278 + agent = context.agent0
279 + user_msg = agent.read_prompt(
280 + "fw.telegram.user_message.md",
281 + sender=_format_user(user),
282 + body=f"[Button pressed: {text}]",
283 + )
284 +
285 + mq.log_user_message(context, user_msg, [], source=" (telegram)")
286 + context.communicate(UserMessage(message=user_msg))
287 + save_tmp_chat(context)
288 +
289 +# Context management
290 +
291 +async def _get_or_create_context(
292 + bot_name: str,
293 + bot_cfg: dict,
294 + message: TgMessage,
295 +) -> AgentContext | None:
296 + user = message.from_user
297 + if not user:
298 + return None
299 + return await _get_or_create_context_from_user(
300 + bot_name, bot_cfg, user.id, user.username, message.chat.id,
301 + )
302 +
303 +
304 +async def _get_or_create_context_from_user(
305 + bot_name: str,
306 + bot_cfg: dict,
307 + user_id: int,
308 + username: str | None,
309 + chat_id: int,
310 +) -> AgentContext | None:
311 + key = _map_key(bot_name, user_id)
312 +
313 + with _chat_map_lock:
314 + state = _load_state()
315 + chats = state.setdefault("chats", {})
316 + ctx_id = chats.get(key)
317 +
318 + # Check if existing context is still alive
319 + if ctx_id:
320 + ctx = AgentContext.get(ctx_id)
321 + if ctx:
322 + return ctx
323 + # Context was garbage collected, remove stale mapping
324 + chats.pop(key, None)
325 +
326 + # Create new context
327 + try:
328 + config = initialize_agent()
329 + display_name = f"@{username}" if username else str(user_id)
330 + ctx = AgentContext(config, name=f"Telegram: {display_name}")
331 +
332 + ctx.data[CTX_TG_BOT] = bot_name
333 + ctx.data[CTX_TG_CHAT_ID] = chat_id
334 + ctx.data[CTX_TG_USER_ID] = user_id
335 + ctx.data[CTX_TG_USERNAME] = username or ""
336 +
337 + project = _get_project(bot_cfg, user_id)
338 + if project:
339 + from helpers import projects
340 + projects.activate_project(ctx.id, project)
341 +
342 + chats[key] = ctx.id
343 + _save_state(state)
344 +
345 + PrintStyle.success(
346 + f"Telegram ({bot_name}): new chat {ctx.id} for user {display_name}"
347 + )
348 + return ctx
349 +
350 + except Exception as e:
351 + PrintStyle.error(f"Telegram: failed to create context: {format_error(e)}")
352 + return None
353 +
354 +# Message content extraction
355 +
356 +def _extract_message_content(message: TgMessage) -> str:
357 + parts = []
358 +
359 + if message.text:
360 + parts.append(message.text)
361 + elif message.caption:
362 + parts.append(message.caption)
363 +
364 + if message.location:
365 + parts.append(f"[Location: {message.location.latitude}, {message.location.longitude}]")
366 +
367 + if message.contact:
368 + parts.append(
369 + f"[Contact: {message.contact.first_name} "
370 + f"{message.contact.last_name or ''} "
371 + f"phone={message.contact.phone_number}]"
372 + )
373 +
374 + if message.sticker:
375 + emoji = message.sticker.emoji or ""
376 + parts.append(f"[Sticker: {emoji}]")
377 +
378 + if message.voice:
379 + parts.append("[Voice message — see attachment]")
380 +
381 + if message.video_note:
382 + parts.append("[Video note — see attachment]")
383 +
384 + return "\n".join(parts) if parts else "[No text content]"
385 +
386 +
387 +async def _download_attachments(bot, message: TgMessage, bot_name: str = "") -> list[str]:
388 + """Download photos, documents, audio, voice, video from message."""
389 + paths: list[str] = []
390 + sub = bot_name if bot_name else "_default"
391 + download_base = os.path.join(files.get_abs_path(DOWNLOAD_FOLDER), sub)
392 + os.makedirs(download_base, exist_ok=True)
393 +
394 + async def _dl(file_id: str, filename: str) -> str | None:
395 + dest = os.path.join(download_base, f"{uuid.uuid4().hex[:8]}_{filename}")
396 + return await tc.download_file(bot, file_id, dest)
397 +
398 + # Photo: get largest resolution
399 + if message.photo:
400 + photo = message.photo[-1]
401 + path = await _dl(photo.file_id, f"photo_{photo.file_unique_id}.jpg")
402 + if path:
403 + paths.append(path)
404 +
405 + # Document
406 + if message.document:
407 + fname = message.document.file_name or f"file_{message.document.file_unique_id}"
408 + path = await _dl(message.document.file_id, fname)
409 + if path:
410 + paths.append(path)
411 +
412 + # Audio
413 + if message.audio:
414 + fname = message.audio.file_name or f"audio_{message.audio.file_unique_id}.mp3"
415 + path = await _dl(message.audio.file_id, fname)
416 + if path:
417 + paths.append(path)
418 +
419 + # Voice
420 + if message.voice:
421 + path = await _dl(message.voice.file_id, f"voice_{message.voice.file_unique_id}.ogg")
422 + if path:
423 + paths.append(path)
424 +
425 + # Video
426 + if message.video:
427 + fname = message.video.file_name or f"video_{message.video.file_unique_id}.mp4"
428 + path = await _dl(message.video.file_id, fname)
429 + if path:
430 + paths.append(path)
431 +
432 + # Video note
433 + if message.video_note:
434 + path = await _dl(
435 + message.video_note.file_id,
436 + f"videonote_{message.video_note.file_unique_id}.mp4",
437 + )
438 + if path:
439 + paths.append(path)
440 +
441 + return paths
442 +
443 +# Reply sending (called from process_chain_end extension)
444 +
445 +async def send_telegram_reply(
446 + context: AgentContext,
447 + response_text: str,
448 + attachments: list[str] | None = None,
449 + keyboard: list[list[dict]] | None = None,
450 +) -> str | None:
451 + """Send reply to Telegram user. Returns error string or None on success."""
452 + from aiogram import Bot
453 + from aiogram.client.default import DefaultBotProperties
454 + from aiogram.enums import ParseMode
455 +
456 + bot_name = context.data.get(CTX_TG_BOT)
457 + if not bot_name:
458 + return "No Telegram bot configured on context"
459 +
460 + instance = get_bot(bot_name)
461 + if not instance:
462 + return f"Bot '{bot_name}' not running"
463 +
464 + chat_id = context.data.get(CTX_TG_CHAT_ID)
465 + if not chat_id:
466 + return "No chat_id on context"
467 +
468 + # Create a temporary Bot bound to the current event loop to avoid
469 + # cross-event-loop issues with the shared instance's aiohttp session.
470 + reply_bot = Bot(
471 + token=instance.bot.token,
472 + default=DefaultBotProperties(parse_mode=ParseMode.MARKDOWN),
473 + )
474 + try:
475 + # Send attachments first
476 + if attachments:
477 + for path in attachments:
478 + if tc.is_image_file(path):
479 + await tc.send_photo(reply_bot, chat_id, path)
480 + else:
481 + await tc.send_file(reply_bot, chat_id, path)
482 +
483 + # Send text (with or without keyboard)
484 + if response_text:
485 + if keyboard:
486 + await tc.send_text_with_keyboard(
487 + reply_bot, chat_id, response_text, keyboard,
488 + )
489 + else:
490 + await tc.send_text(reply_bot, chat_id, response_text)
491 +
492 + return None
493 +
494 + except Exception as e:
495 + error = format_error(e)
496 + PrintStyle.error(f"Telegram reply failed: {error}")
497 + return error
498 + finally:
499 + await reply_bot.session.close()
500 +
501 +# Helpers
502 +
503 +def _format_user(user) -> str:
504 + name = user.first_name or ""
505 + if user.last_name:
506 + name += f" {user.last_name}"
507 + if user.username:
508 + name += f" (@{user.username})"
509 + return name.strip() or str(user.id)
510 +
511 +
512 +def find_context_for_bot_chat(bot_name: str, chat_id: int) -> AgentContext | None:
513 + """Find active context matching a Telegram bot + chat_id."""
514 + for ctx in AgentContext.all():
515 + if ctx.data.get(CTX_TG_BOT) == bot_name and ctx.data.get(CTX_TG_CHAT_ID) == chat_id:
516 + return ctx
517 + return None
plugins/_telegram_integration/helpers/telegram_client.py new
+226
@@ -0,0 +1,226 @@
1 +import asyncio
2 +import os
3 +from typing import BinaryIO
4 +
5 +from aiogram import Bot
6 +from aiogram.exceptions import TelegramBadRequest
7 +from aiogram.types import (
8 + BufferedInputFile,
9 + FSInputFile,
10 + InlineKeyboardButton,
11 + InlineKeyboardMarkup,
12 +)
13 +
14 +from helpers.errors import format_error
15 +from helpers.print_style import PrintStyle
16 +
17 +# Text messages
18 +
19 +MAX_MESSAGE_LENGTH: int = 4096 # Telegram message length limit
20 +
21 +
22 +async def send_text(
23 + bot: Bot,
24 + chat_id: int,
25 + text: str,
26 + reply_to_message_id: int | None = None,
27 + parse_mode: str | None = None,
28 +) -> int | None:
29 + """Send text message, splitting if too long. Returns last message_id or None on error."""
30 + try:
31 + chunks = _split_text(text, MAX_MESSAGE_LENGTH)
32 + last_msg_id = None
33 + for chunk in chunks:
34 + try:
35 + msg = await bot.send_message(
36 + chat_id=chat_id,
37 + text=chunk,
38 + reply_to_message_id=reply_to_message_id,
39 + parse_mode=parse_mode,
40 + )
41 + last_msg_id = msg.message_id
42 + except TelegramBadRequest:
43 + # Retry without markdown if parse fails
44 + msg = await bot.send_message(
45 + chat_id=chat_id,
46 + text=chunk,
47 + reply_to_message_id=reply_to_message_id,
48 + parse_mode=None,
49 + )
50 + last_msg_id = msg.message_id
51 + return last_msg_id
52 + except Exception as e:
53 + PrintStyle.error(f"Telegram send_text failed: {format_error(e)}")
54 + return None
55 +
56 +# Files and images
57 +
58 +async def send_file(
59 + bot: Bot,
60 + chat_id: int,
61 + file_path: str,
62 + caption: str = "",
63 + reply_to_message_id: int | None = None,
64 +) -> int | None:
65 + """Send a file from local path. Returns message_id or None on error."""
66 + try:
67 + if not os.path.isfile(file_path):
68 + PrintStyle.error(f"Telegram: file not found: {file_path}")
69 + return None
70 + input_file = FSInputFile(file_path)
71 + msg = await bot.send_document(
72 + chat_id=chat_id,
73 + document=input_file,
74 + caption=caption[:1024] if caption else None,
75 + reply_to_message_id=reply_to_message_id,
76 + )
77 + return msg.message_id
78 + except Exception as e:
79 + PrintStyle.error(f"Telegram send_file failed: {format_error(e)}")
80 + return None
81 +
82 +
83 +async def send_photo(
84 + bot: Bot,
85 + chat_id: int,
86 + photo_path: str,
87 + caption: str = "",
88 + reply_to_message_id: int | None = None,
89 +) -> int | None:
90 + """Send a photo from local path. Returns message_id or None on error."""
91 + try:
92 + if not os.path.isfile(photo_path):
93 + PrintStyle.error(f"Telegram: photo not found: {photo_path}")
94 + return None
95 + input_file = FSInputFile(photo_path)
96 + msg = await bot.send_photo(
97 + chat_id=chat_id,
98 + photo=input_file,
99 + caption=caption[:1024] if caption else None,
100 + reply_to_message_id=reply_to_message_id,
101 + )
102 + return msg.message_id
103 + except Exception as e:
104 + PrintStyle.error(f"Telegram send_photo failed: {format_error(e)}")
105 + return None
106 +
107 +
108 +async def send_file_bytes(
109 + bot: Bot,
110 + chat_id: int,
111 + content: bytes,
112 + filename: str,
113 + caption: str = "",
114 +) -> int | None:
115 + """Send file from bytes content."""
116 + try:
117 + input_file = BufferedInputFile(content, filename=filename)
118 + msg = await bot.send_document(
119 + chat_id=chat_id,
120 + document=input_file,
121 + caption=caption[:1024] if caption else None,
122 + )
123 + return msg.message_id
124 + except Exception as e:
125 + PrintStyle.error(f"Telegram send_file_bytes failed: {format_error(e)}")
126 + return None
127 +
128 +# Inline keyboards
129 +
130 +def build_inline_keyboard(
131 + buttons: list[list[dict]],
132 +) -> InlineKeyboardMarkup:
133 + """Build inline keyboard from a list of rows.
134 + Each row is a list of dicts with keys: text, callback_data or url.
135 + """
136 + rows = []
137 + for row in buttons:
138 + row_buttons = []
139 + for btn in row:
140 + if "url" in btn:
141 + row_buttons.append(InlineKeyboardButton(
142 + text=btn["text"], url=btn["url"],
143 + ))
144 + else:
145 + row_buttons.append(InlineKeyboardButton(
146 + text=btn["text"],
147 + callback_data=btn.get("callback_data", btn["text"]),
148 + ))
149 + rows.append(row_buttons)
150 + return InlineKeyboardMarkup(inline_keyboard=rows)
151 +
152 +
153 +async def send_text_with_keyboard(
154 + bot: Bot,
155 + chat_id: int,
156 + text: str,
157 + buttons: list[list[dict]],
158 + reply_to_message_id: int | None = None,
159 +) -> int | None:
160 + """Send text with inline keyboard buttons."""
161 + try:
162 + keyboard = build_inline_keyboard(buttons)
163 + msg = await bot.send_message(
164 + chat_id=chat_id,
165 + text=text,
166 + reply_markup=keyboard,
167 + reply_to_message_id=reply_to_message_id,
168 + )
169 + return msg.message_id
170 + except Exception as e:
171 + PrintStyle.error(f"Telegram send_text_with_keyboard failed: {format_error(e)}")
172 + return None
173 +
174 +# Typing indicator
175 +
176 +async def send_typing(bot: Bot, chat_id: int):
177 + """Send 'typing...' action to chat."""
178 + try:
179 + await bot.send_chat_action(chat_id=chat_id, action="typing")
180 + except Exception:
181 + pass
182 +
183 +# File download
184 +
185 +async def download_file(
186 + bot: Bot,
187 + file_id: str,
188 + destination: str,
189 +) -> str | None:
190 + """Download a file by file_id to destination path. Returns path or None on error."""
191 + try:
192 + file = await bot.get_file(file_id)
193 + if not file.file_path:
194 + return None
195 + os.makedirs(os.path.dirname(destination), exist_ok=True)
196 + await bot.download_file(file.file_path, destination)
197 + return destination
198 + except Exception as e:
199 + PrintStyle.error(f"Telegram download failed: {format_error(e)}")
200 + return None
201 +
202 +# Helpers
203 +
204 +def _split_text(text: str, max_len: int) -> list[str]:
205 + if len(text) <= max_len:
206 + return [text]
207 + chunks = []
208 + while text:
209 + if len(text) <= max_len:
210 + chunks.append(text)
211 + break
212 + # Try to split at newline
213 + split_at = text.rfind("\n", 0, max_len)
214 + if split_at < max_len // 2:
215 + split_at = max_len
216 + chunks.append(text[:split_at])
217 + text = text[split_at:].lstrip("\n")
218 + return chunks
219 +
220 +
221 +_IMAGE_EXTENSIONS = {".jpg", ".jpeg", ".png", ".gif", ".webp", ".bmp"}
222 +
223 +
224 +def is_image_file(path: str) -> bool:
225 + _, ext = os.path.splitext(path.lower())
226 + return ext in _IMAGE_EXTENSIONS
plugins/_telegram_integration/plugin.yaml new
+8
@@ -0,0 +1,8 @@
1 +name: _telegram_integration
2 +title: Telegram Integration
3 +description: Communicate with Agent Zero via Telegram. Supports polling and webhook modes with per-user chat sessions.
4 +version: 1.0.0
5 +settings_sections:
6 + - external
7 +per_project_config: false
8 +per_agent_config: false
plugins/_telegram_integration/prompts/fw.telegram.cleared.md new
+3
@@ -0,0 +1,3 @@
1 +chat cleared by user via /clear command
2 +previous context has been reset
3 +greet user and wait for new input
\ No newline at end of file
plugins/_telegram_integration/prompts/fw.telegram.send_failed.md new
+4
@@ -0,0 +1,4 @@
1 +telegram send failed (attempt {{attempt}}/{{max_retries}}): {{error}}
2 +previous response was NOT delivered to user
3 +retry without problematic content or inform user via alternative means
4 +if attachment caused the issue, retry without it
\ No newline at end of file
plugins/_telegram_integration/prompts/fw.telegram.system_context.md new
+4
@@ -0,0 +1,4 @@
1 +telegram session user communicates via Telegram
2 +response tool sends the message to the user dont use python code
3 +break_loop true > stop working and wait for user reply
4 +break_loop false > only for mid-task progress updates then keep working
\ No newline at end of file
plugins/_telegram_integration/prompts/fw.telegram.system_context_reply.md new
+46
@@ -0,0 +1,46 @@
1 +# Telegram session behavior
2 +user communicates via Telegram messenger
3 +response tool = send message to user on Telegram
4 +dont use code to send messages
5 +break_loop true > stop working and wait for user reply
6 +break_loop false > only for mid-task progress updates then keep working
7 +include file paths in attachments array to send files/images
8 +for multiple files zip first then attach single archive
9 +optionally set keyboard array for inline buttons
10 +
11 +usage:
12 +
13 +~~~json
14 +{
15 + ...
16 + "tool_name": "response",
17 + "tool_args": {
18 + "text": "working on it...",
19 + "break_loop": false
20 + }
21 +}
22 +~~~
23 +
24 +~~~json
25 +{
26 + ...
27 + "tool_name": "response",
28 + "tool_args": {
29 + "text": "Here is the result",
30 + "attachments": ["/path/to/file.zip"],
31 + "break_loop": true
32 + }
33 +}
34 +~~~
35 +
36 +~~~json
37 +{
38 + ...
39 + "tool_name": "response",
40 + "tool_args": {
41 + "text": "Choose an option:",
42 + "keyboard": [[{"text": "Option A", "callback_data": "a"}, {"text": "Option B", "callback_data": "b"}]],
43 + "break_loop": true
44 + }
45 +}
46 +~~~
\ No newline at end of file
plugins/_telegram_integration/prompts/fw.telegram.update_error.md new
+1
@@ -0,0 +1 @@
1 +telegram update failed: {{error}}
\ No newline at end of file
plugins/_telegram_integration/prompts/fw.telegram.update_ok.md new
+1
@@ -0,0 +1 @@
1 +telegram update sent, continue working
\ No newline at end of file
plugins/_telegram_integration/prompts/fw.telegram.user_message.md new
+3
@@ -0,0 +1,3 @@
1 +[Telegram message from {{sender}}]
2 +
3 +{{body}}
\ No newline at end of file
plugins/_telegram_integration/prompts/fw.telegram.user_message_instructions.md new
+2
@@ -0,0 +1,2 @@
1 +
2 +[Handler instructions: {{instructions}}]
\ No newline at end of file
requirements.txt
+1
@@ -48,6 +48,7 @@ html2text>=2024.2.26
48 beautifulsoup4>=4.12.3
49 boto3>=1.35.0
50 exchangelib>=5.4.3
51 +aiogram>=3.15.0
52 pywinpty==3.0.2; sys_platform == "win32"
53 python-socketio>=5.14.2
54 uvicorn>=0.38.0