Scope Telegram integration commands
Anmol Malik committed
Jun 7, 2026 at 21:20 UTC
abf0cef8e8d7f783ce404ec777ceb0f4dde128e1
5 files changed
+46
-20
helpers/integration_commands.py
+39
-13
@@ -25,6 +25,7 @@ class IntegrationCommandDef:
25
aliases: tuple[str, ...] = ()
26
args_hint: str = ""
27
menu: bool = True
28
+ integrations: tuple[str, ...] = ()
29
30
31
COMMAND_REGISTRY: tuple[IntegrationCommandDef, ...] = (
@@ -40,6 +41,7 @@ COMMAND_REGISTRY: tuple[IntegrationCommandDef, ...] = (
41
"Show or switch recent chat sessions.",
42
"Session",
43
aliases=("session",),
44
+ integrations=("telegram",),
45
),
46
IntegrationCommandDef("clear", "Reset the current chat context.", "Session", aliases=("reset",)),
47
IntegrationCommandDef(
@@ -63,12 +65,14 @@ COMMAND_REGISTRY: tuple[IntegrationCommandDef, ...] = (
65
"Enable or disable Telegram response streaming.",
66
"Configuration",
67
args_hint="[on|off]",
68
+ integrations=("telegram",),
69
),
70
IntegrationCommandDef(
71
"tools",
72
"Show or hide Telegram tool progress.",
73
"Configuration",
74
args_hint="[on|off]",
75
+ integrations=("telegram",),
76
),
77
IntegrationCommandDef(
78
"project",
@@ -109,46 +113,55 @@ def extract_command_line(text: str) -> str:
113
return ""
114
115
112
-def parse_command(text: str) -> tuple[str, str] | None:
116
+def parse_command(text: str, *, integration: str | None = None) -> tuple[str, str] | None:
117
line = extract_command_line(text)
118
if not line.startswith("/"):
119
return None
120
121
command, _, args = line.partition(" ")
122
command = _normalize_command_token(command)
119
- resolved = resolve_command(command)
123
+ resolved = resolve_command(command, integration=integration)
124
if not resolved:
125
return None
126
127
return f"/{resolved.name}", args.strip()
128
129
126
-def resolve_command(command: str) -> IntegrationCommandDef | None:
130
+def resolve_command(command: str, *, integration: str | None = None) -> IntegrationCommandDef | None:
131
normalized = _normalize_command_token(command)
132
if not normalized.startswith("/"):
133
normalized = f"/{normalized}"
130
- return _COMMAND_LOOKUP.get(normalized)
134
+ command_def = _COMMAND_LOOKUP.get(normalized)
135
+ if not command_def or not _is_command_available(command_def, integration):
136
+ return None
137
+ return command_def
138
139
140
def telegram_menu_commands() -> list[tuple[str, str]]:
141
return [
142
(command.name, _telegram_description(command))
143
for command in COMMAND_REGISTRY
137
- if command.menu
144
+ if command.menu and _is_command_available(command, "telegram")
145
]
146
147
141
-def command_names(include_aliases: bool = True) -> list[str]:
148
+def command_names(include_aliases: bool = True, *, integration: str | None = None) -> list[str]:
149
names: list[str] = []
150
for command in COMMAND_REGISTRY:
151
+ if not _is_command_available(command, integration):
152
+ continue
153
names.append(command.name)
154
if include_aliases:
155
names.extend(command.aliases)
156
return names
157
158
150
-def help_text(*, full: bool = False) -> str:
151
- commands = COMMAND_REGISTRY if full else tuple(c for c in COMMAND_REGISTRY if c.menu)
159
+def help_text(*, full: bool = False, integration: str | None = None) -> str:
160
+ commands = tuple(
161
+ command
162
+ for command in COMMAND_REGISTRY
163
+ if _is_command_available(command, integration) and (full or command.menu)
164
+ )
165
lines = ["Available commands:"]
166
for command in commands:
167
args = f" {command.args_hint}" if command.args_hint else ""
@@ -159,19 +172,24 @@ def help_text(*, full: bool = False) -> str:
172
return "\n".join(lines)
173
174
162
-def unknown_command_text(command: str) -> str:
175
+def unknown_command_text(command: str, *, integration: str | None = None) -> str:
176
token = _normalize_command_token(command).split(" ", 1)[0]
164
- return f"Unknown command: {token}\n\n{help_text(full=True)}"
177
+ return f"Unknown command: {token}\n\n{help_text(full=True, integration=integration)}"
178
179
167
-def try_handle_command(context: "AgentContext", text: str) -> str | None:
168
- parsed = parse_command(text)
180
+def try_handle_command(
181
+ context: "AgentContext",
182
+ text: str,
183
+ *,
184
+ integration: str | None = None,
185
+) -> str | None:
186
+ parsed = parse_command(text, integration=integration)
187
if not parsed:
188
return None
189
190
command, args = parsed
191
if command == "/commands":
174
- return help_text(full=True)
192
+ return help_text(full=True, integration=integration)
193
if command == "/status":
194
return _handle_status(context)
195
if command == "/sessions":
@@ -213,6 +231,14 @@ def _normalize_command_token(command: str) -> str:
231
return f"{token} {rest[0]}".strip() if rest else token
232
233
234
+def _is_command_available(command: IntegrationCommandDef, integration: str | None) -> bool:
235
+ if not command.integrations:
236
+ return True
237
+ if not integration:
238
+ return False
239
+ return integration.lower() in command.integrations
240
+
241
+
242
def _handle_queue(context: "AgentContext", args: str) -> str:
243
queue = mq.get_queue(context)
244
count = len(queue)
plugins/_telegram_integration/extensions/python/system_prompt/_20_telegram_context.py
+2
-2
@@ -40,8 +40,8 @@ def _telegram_commands_prompt() -> str:
40
"If the user asks what commands exist, refer them to /commands.",
41
"Current integration commands:",
42
]
43
- for name in integration_commands.command_names(include_aliases=False):
44
- definition = integration_commands.resolve_command(name)
43
+ for name in integration_commands.command_names(include_aliases=False, integration="telegram"):
44
+ definition = integration_commands.resolve_command(name, integration="telegram")
45
if definition:
46
args = f" {definition.args_hint}" if definition.args_hint else ""
47
lines.append(f"- /{definition.name}{args}: {definition.description}")
plugins/_telegram_integration/helpers/bot_manager.py
+1
-1
@@ -59,7 +59,7 @@ def create_bot(
59
if on_command_control:
60
router.message.register(
61
on_command_control,
62
- Command(commands=integration_commands.command_names()),
62
+ Command(commands=integration_commands.command_names(integration="telegram")),
63
)
64
65
if on_callback_query:
plugins/_telegram_integration/helpers/command_ui.py
+1
-1
@@ -47,7 +47,7 @@ async def handle_command(
47
reply_to_message_id: int | None,
48
text: str,
49
) -> bool:
50
- parsed = integration_commands.parse_command(text or "")
50
+ parsed = integration_commands.parse_command(text or "", integration="telegram")
51
if not parsed:
52
return False
53
command, args = parsed
plugins/_telegram_integration/helpers/handler.py
+3
-3
@@ -186,7 +186,7 @@ async def handle_message(message: TgMessage, bot_name: str, bot_cfg: dict):
186
):
187
return
188
189
- command_reply = integration_commands.try_handle_command(context, text)
189
+ command_reply = integration_commands.try_handle_command(context, text, integration="telegram")
190
if command_reply is not None:
191
await _send_with_temp_bot(
192
instance.bot.token,
@@ -201,7 +201,7 @@ async def handle_message(message: TgMessage, bot_name: str, bot_cfg: dict):
201
await _send_with_temp_bot(
202
instance.bot.token,
203
message.chat.id,
204
- integration_commands.unknown_command_text(command),
204
+ integration_commands.unknown_command_text(command, integration="telegram"),
205
parse_mode=None,
206
reply_to_message_id=message.message_id,
207
)
@@ -303,7 +303,7 @@ async def handle_callback_query(query: CallbackQuery, bot_name: str, bot_cfg: di
303
if text.startswith("tg:"):
304
return
305
306
- command_reply = integration_commands.try_handle_command(context, text)
306
+ command_reply = integration_commands.try_handle_command(context, text, integration="telegram")
307
if command_reply is not None:
308
if instance:
309
await _send_with_temp_bot(