feat: Telegram plugin extensions
keyboardstaff committed
Mar 20, 2026 at 10:21 UTC
88b5d97335ca8aaa6381ae4ed6255d99f09580d4
4 files changed
+286
plugins/_telegram_integration/extensions/python/job_loop/_10_telegram_bot.py
new
+117
@@ -0,0 +1,117 @@
1
+import asyncio
2
+from functools import partial
3
+from typing import Any
4
+
5
+from helpers.extension import Extension
6
+from helpers.errors import format_error
7
+from helpers.print_style import PrintStyle
8
+from helpers import plugins
9
+
10
+
11
+PLUGIN_NAME: str = "_telegram_integration"
12
+
13
+
14
+class TelegramBotManager(Extension):
15
+
16
+ async def execute(self, **kwargs: Any) -> None:
17
+ from plugins._telegram_integration.helpers.bot_manager import (
18
+ get_all_bots,
19
+ create_bot,
20
+ cache_bot_info,
21
+ start_polling,
22
+ setup_webhook,
23
+ stop_bot,
24
+ )
25
+ from plugins._telegram_integration.helpers.handler import (
26
+ handle_start,
27
+ handle_clear,
28
+ handle_message,
29
+ handle_callback_query,
30
+ cleanup_old_attachments,
31
+ )
32
+
33
+ cleanup_old_attachments()
34
+
35
+ config = plugins.get_plugin_config(PLUGIN_NAME) or {}
36
+ bots_cfg = config.get("bots", [])
37
+ enabled_names = {
38
+ b["name"] for b in bots_cfg if b.get("enabled") and b.get("name") and b.get("token")
39
+ }
40
+
41
+ running = get_all_bots()
42
+
43
+ # Stop bots that are no longer enabled
44
+ for name in list(running.keys()):
45
+ if name not in enabled_names:
46
+ await stop_bot(name)
47
+
48
+ # Start new bots
49
+ for bot_cfg in bots_cfg:
50
+ name = bot_cfg.get("name", "")
51
+ if not name or not bot_cfg.get("enabled") or not bot_cfg.get("token"):
52
+ continue
53
+ if name in running:
54
+ inst = running[name]
55
+ if inst.task and not inst.task.done():
56
+ continue # already running
57
+
58
+ try:
59
+ # Create handler closures that capture bot_name and config
60
+ _on_start = partial(_wrap_start, bot_name=name, bot_cfg=bot_cfg)
61
+ _on_clear = partial(_wrap_clear, bot_name=name, bot_cfg=bot_cfg)
62
+ _on_message = partial(_wrap_message, bot_name=name, bot_cfg=bot_cfg)
63
+ _on_callback = partial(_wrap_callback, bot_name=name, bot_cfg=bot_cfg)
64
+
65
+ instance = create_bot(
66
+ name=name,
67
+ token=bot_cfg["token"],
68
+ on_message=_on_message,
69
+ on_command_start=_on_start,
70
+ on_command_clear=_on_clear,
71
+ on_callback_query=_on_callback,
72
+ group_mode=bot_cfg.get("group_mode", "mention"),
73
+ )
74
+
75
+ await cache_bot_info(instance)
76
+
77
+ mode = bot_cfg.get("mode", "polling")
78
+ if mode == "webhook":
79
+ webhook_url = bot_cfg.get("webhook_url", "")
80
+ webhook_secret = bot_cfg.get("webhook_secret", "")
81
+ if webhook_url:
82
+ await setup_webhook(instance, webhook_url, webhook_secret)
83
+ else:
84
+ PrintStyle.error(
85
+ f"Telegram ({name}): webhook mode requires webhook_url"
86
+ )
87
+ continue
88
+ else:
89
+ await start_polling(instance)
90
+
91
+ PrintStyle.success(f"Telegram ({name}): bot started in {mode} mode")
92
+
93
+ except Exception as e:
94
+ PrintStyle.error(
95
+ f"Telegram ({name}): failed to start: {format_error(e)}"
96
+ )
97
+
98
+# Wrapper functions for aiogram handlers
99
+
100
+async def _wrap_start(message, bot_name: str, bot_cfg: dict):
101
+ from plugins._telegram_integration.helpers.handler import handle_start
102
+ await handle_start(message, bot_name, bot_cfg)
103
+
104
+
105
+async def _wrap_clear(message, bot_name: str, bot_cfg: dict):
106
+ from plugins._telegram_integration.helpers.handler import handle_clear
107
+ await handle_clear(message, bot_name, bot_cfg)
108
+
109
+
110
+async def _wrap_message(message, bot_name: str, bot_cfg: dict):
111
+ from plugins._telegram_integration.helpers.handler import handle_message
112
+ await handle_message(message, bot_name, bot_cfg)
113
+
114
+
115
+async def _wrap_callback(query, bot_name: str, bot_cfg: dict):
116
+ from plugins._telegram_integration.helpers.handler import handle_callback_query
117
+ await handle_callback_query(query, bot_name, bot_cfg)
plugins/_telegram_integration/extensions/python/process_chain_end/_55_telegram_reply.py
new
+86
@@ -0,0 +1,86 @@
1
+from helpers.extension import Extension
2
+from helpers.print_style import PrintStyle
3
+from helpers.errors import format_error
4
+from agent import AgentContext, LoopData, UserMessage
5
+from plugins._telegram_integration.helpers.handler import CTX_TG_BOT, CTX_TG_ATTACHMENTS, CTX_TG_KEYBOARD
6
+
7
+MAX_SEND_RETRIES: int = 2
8
+CTX_SEND_FAILURES: str = "_telegram_send_failures"
9
+
10
+
11
+class TelegramAutoReply(Extension):
12
+
13
+ async def execute(self, loop_data: LoopData = LoopData(), **kwargs):
14
+ if not self.agent or self.agent.number != 0:
15
+ return
16
+
17
+ context = self.agent.context
18
+ if not context.data.get(CTX_TG_BOT):
19
+ return
20
+
21
+ response_text = _extract_last_response(context)
22
+ if not response_text:
23
+ return
24
+
25
+ attachments = context.data.pop(CTX_TG_ATTACHMENTS, [])
26
+ keyboard = context.data.pop(CTX_TG_KEYBOARD, None)
27
+
28
+ try:
29
+ await self._send_reply(context, response_text, attachments, keyboard)
30
+ except Exception as e:
31
+ PrintStyle.error(f"Telegram auto-reply error: {format_error(e)}")
32
+
33
+ async def _send_reply(
34
+ self,
35
+ context: AgentContext,
36
+ response_text: str,
37
+ attachments: list[str],
38
+ keyboard: list[list[dict]] | None,
39
+ ):
40
+ from plugins._telegram_integration.helpers.handler import send_telegram_reply
41
+
42
+ error = await send_telegram_reply(
43
+ context, response_text, attachments or None, keyboard,
44
+ )
45
+ if not error:
46
+ context.data[CTX_SEND_FAILURES] = 0
47
+ return
48
+
49
+ failures = context.data.get(CTX_SEND_FAILURES, 0) + 1
50
+ context.data[CTX_SEND_FAILURES] = failures
51
+ if failures <= MAX_SEND_RETRIES:
52
+ _notify_agent_of_failure(context, error, failures)
53
+ else:
54
+ PrintStyle.error(
55
+ f"Telegram send failed {failures} times, giving up: {error}"
56
+ )
57
+ context.log.log(
58
+ type="error",
59
+ heading="Telegram send failed (max retries reached)",
60
+ content=error,
61
+ )
62
+
63
+# Helpers
64
+
65
+def _extract_last_response(context: AgentContext) -> str:
66
+ with context.log._lock:
67
+ logs = list(context.log.logs)
68
+ if not logs:
69
+ return ""
70
+ for item in reversed(logs):
71
+ if item.type == "response":
72
+ return item.content or ""
73
+ return ""
74
+
75
+
76
+def _notify_agent_of_failure(
77
+ context: AgentContext, error: str, attempt: int,
78
+):
79
+ msg = context.agent0.read_prompt(
80
+ "fw.telegram.send_failed.md",
81
+ error=error,
82
+ attempt=str(attempt),
83
+ max_retries=str(MAX_SEND_RETRIES),
84
+ )
85
+ context.log.log(type="error", heading="Telegram send failed", content=error)
86
+ context.communicate(UserMessage(message="", system_message=[msg]))
plugins/_telegram_integration/extensions/python/system_prompt/_20_telegram_context.py
new
+20
@@ -0,0 +1,20 @@
1
+from helpers.extension import Extension
2
+from agent import LoopData
3
+from plugins._telegram_integration.helpers.handler import CTX_TG_BOT
4
+
5
+
6
+class TelegramContextPrompt(Extension):
7
+
8
+ async def execute(
9
+ self,
10
+ system_prompt: list[str] = [],
11
+ loop_data: LoopData = LoopData(),
12
+ **kwargs,
13
+ ):
14
+ if not self.agent:
15
+ return
16
+
17
+ if self.agent.context.data.get(CTX_TG_BOT):
18
+ system_prompt.append(
19
+ self.agent.read_prompt("fw.telegram.system_context_reply.md")
20
+ )
plugins/_telegram_integration/extensions/python/tool_execute_after/_50_telegram_response.py
new
+63
@@ -0,0 +1,63 @@
1
+from helpers.extension import Extension
2
+from helpers.print_style import PrintStyle
3
+from helpers.tool import Response
4
+from plugins._telegram_integration.helpers.handler import (
5
+ CTX_TG_BOT,
6
+ CTX_TG_ATTACHMENTS,
7
+ CTX_TG_KEYBOARD,
8
+)
9
+
10
+
11
+class TelegramResponseIntercept(Extension):
12
+
13
+ async def execute(
14
+ self, tool_name: str = "", response: Response | None = None, **kwargs,
15
+ ):
16
+ if tool_name != "response":
17
+ return
18
+ if not self.agent:
19
+ return
20
+ context = self.agent.context
21
+ if not context.data.get(CTX_TG_BOT):
22
+ return
23
+
24
+ tool = self.agent.loop_data.current_tool
25
+ if not tool:
26
+ return
27
+
28
+ # Capture attachments for later (process_chain_end) or inline send
29
+ attachments = tool.args.get("attachments", [])
30
+ if attachments:
31
+ context.data[CTX_TG_ATTACHMENTS] = attachments
32
+
33
+ # Capture inline keyboard if provided
34
+ keyboard = tool.args.get("keyboard", None)
35
+ if keyboard:
36
+ context.data[CTX_TG_KEYBOARD] = keyboard
37
+
38
+ # Check break_loop arg from agent
39
+ agent_break = tool.args.get("break_loop", True)
40
+ if agent_break is False and response:
41
+ await self._send_inline(context, tool, response)
42
+
43
+ async def _send_inline(self, context, tool, response: Response):
44
+ from plugins._telegram_integration.helpers.handler import send_telegram_reply
45
+
46
+ agent = self.agent
47
+ assert agent is not None
48
+
49
+ text = tool.args.get("text", tool.args.get("message", ""))
50
+ attachments = context.data.pop(CTX_TG_ATTACHMENTS, [])
51
+ keyboard = context.data.pop(CTX_TG_KEYBOARD, None)
52
+
53
+ error = await send_telegram_reply(context, text, attachments or None, keyboard)
54
+
55
+ if error:
56
+ result = agent.read_prompt("fw.telegram.update_error.md", error=error)
57
+ else:
58
+ result = agent.read_prompt("fw.telegram.update_ok.md")
59
+
60
+ # Override response: don't break loop, add result to history
61
+ response.break_loop = False
62
+ response.message = result
63
+ agent.hist_add_tool_result("response", result)