feat(telegram): add config option to disable link previews (#8728)

cli-config.yaml.example

@@ -564,6 +564,18 @@ platform_toolsets:
   homeassistant: [hermes-homeassistant]
   qqbot: [hermes-qqbot]
 
+# =============================================================================
+# Gateway Platform Settings
+# =============================================================================
+# Optional per-platform messaging settings.
+# Platform-specific knobs live under `extra`.
+#
+# platforms:
+#   telegram:
+#     reply_to_mode: "first"  # off | first | all
+#     extra:
+#       disable_link_previews: false  # Set true to suppress Telegram URL previews in bot messages
+
 # ─────────────────────────────────────────────────────────────────────────────
 # Available toolsets (use these names in platform_toolsets or the toolsets list)
 #

gateway/config.py

@@ -638,6 +638,16 @@ def load_gateway_config() -> GatewayConfig:
                     os.environ["TELEGRAM_IGNORED_THREADS"] = str(ignored_threads)
                 if "reactions" in telegram_cfg and not os.getenv("TELEGRAM_REACTIONS"):
                     os.environ["TELEGRAM_REACTIONS"] = str(telegram_cfg["reactions"]).lower()
+                if "disable_link_previews" in telegram_cfg:
+                    plat_data = platforms_data.setdefault(Platform.TELEGRAM.value, {})
+                    if not isinstance(plat_data, dict):
+                        plat_data = {}
+                        platforms_data[Platform.TELEGRAM.value] = plat_data
+                    extra = plat_data.setdefault("extra", {})
+                    if not isinstance(extra, dict):
+                        extra = {}
+                        plat_data["extra"] = extra
+                    extra["disable_link_previews"] = telegram_cfg["disable_link_previews"]
 
             whatsapp_cfg = yaml_cfg.get("whatsapp", {})
             if isinstance(whatsapp_cfg, dict):

gateway/platforms/telegram.py

@@ -18,6 +18,10 @@
 
 try:
     from telegram import Update, Bot, Message, InlineKeyboardButton, InlineKeyboardMarkup
+    try:
+        from telegram import LinkPreviewOptions
+    except ImportError:
+        LinkPreviewOptions = None
     from telegram.ext import (
         Application,
         CommandHandler,
@@ -36,6 +40,7 @@
     Message = Any
     InlineKeyboardButton = Any
     InlineKeyboardMarkup = Any
+    LinkPreviewOptions = None
     Application = Any
     CommandHandler = Any
     CallbackQueryHandler = Any
@@ -137,6 +142,7 @@ def __init__(self, config: PlatformConfig):
         self._webhook_mode: bool = False
         self._mention_patterns = self._compile_mention_patterns()
         self._reply_to_mode: str = getattr(config, 'reply_to_mode', 'first') or 'first'
+        self._disable_link_previews: bool = self._coerce_bool_extra("disable_link_previews", False)
         # Buffer rapid/album photo updates so Telegram image bursts are handled
         # as a single MessageEvent instead of self-interrupting multiple turns.
         self._media_batch_delay_seconds = float(os.getenv("HERMES_TELEGRAM_MEDIA_BATCH_DELAY_SECONDS", "0.8"))
@@ -202,6 +208,26 @@ def _looks_like_network_error(error: Exception) -> bool:
             pass
         return isinstance(error, OSError)
 
+    def _coerce_bool_extra(self, key: str, default: bool = False) -> bool:
+        value = self.config.extra.get(key) if getattr(self.config, "extra", None) else None
+        if value is None:
+            return default
+        if isinstance(value, str):
+            lowered = value.strip().lower()
+            if lowered in ("true", "1", "yes", "on"):
+                return True
+            if lowered in ("false", "0", "no", "off"):
+                return False
+            return default
+        return bool(value)
+
+    def _link_preview_kwargs(self) -> Dict[str, Any]:
+        if not getattr(self, "_disable_link_previews", False):
+            return {}
+        if LinkPreviewOptions is not None:
+            return {"link_preview_options": LinkPreviewOptions(is_disabled=True)}
+        return {"disable_web_page_preview": True}
+
     async def _handle_polling_network_error(self, error: Exception) -> None:
         """Reconnect polling after a transient network interruption.
 
@@ -856,6 +882,7 @@ async def send(
                                 parse_mode=ParseMode.MARKDOWN_V2,
                                 reply_to_message_id=reply_to_id,
                                 message_thread_id=effective_thread_id,
+                                **self._link_preview_kwargs(),
                             )
                         except Exception as md_error:
                             # Markdown parsing failed, try plain text
@@ -868,6 +895,7 @@ async def send(
                                     parse_mode=None,
                                     reply_to_message_id=reply_to_id,
                                     message_thread_id=effective_thread_id,
+                                    **self._link_preview_kwargs(),
                                 )
                             else:
                                 raise
@@ -1055,6 +1083,7 @@ async def send_update_prompt(
                 text=text,
                 parse_mode=ParseMode.MARKDOWN,
                 reply_markup=keyboard,
+                **self._link_preview_kwargs(),
             )
             return SendResult(success=True, message_id=str(msg.message_id))
         except Exception as e:
@@ -1111,6 +1140,7 @@ async def send_exec_approval(
                 "text": text,
                 "parse_mode": ParseMode.MARKDOWN,
                 "reply_markup": keyboard,
+                **self._link_preview_kwargs(),
             }
             if thread_id:
                 kwargs["message_thread_id"] = int(thread_id)
@@ -1181,6 +1211,7 @@ def get_label(slug):
                 parse_mode=ParseMode.MARKDOWN,
                 reply_markup=keyboard,
                 message_thread_id=int(thread_id) if thread_id else None,
+                **self._link_preview_kwargs(),
             )
 
             # Store picker state keyed by chat_id

tests/gateway/test_config.py

@@ -284,6 +284,22 @@ def test_bridges_unauthorized_dm_behavior_from_config_yaml(self, tmp_path, monke
         assert config.unauthorized_dm_behavior == "ignore"
         assert config.platforms[Platform.WHATSAPP].extra["unauthorized_dm_behavior"] == "pair"
 
+    def test_bridges_telegram_disable_link_previews_from_config_yaml(self, tmp_path, monkeypatch):
+        hermes_home = tmp_path / ".hermes"
+        hermes_home.mkdir()
+        config_path = hermes_home / "config.yaml"
+        config_path.write_text(
+            "telegram:\n"
+            "  disable_link_previews: true\n",
+            encoding="utf-8",
+        )
+
+        monkeypatch.setenv("HERMES_HOME", str(hermes_home))
+
+        config = load_gateway_config()
+
+        assert config.platforms[Platform.TELEGRAM].extra["disable_link_previews"] is True
+
 
 class TestHomeChannelEnvOverrides:
     """Home channel env vars should apply even when the platform was already

tests/gateway/test_telegram_approval_buttons.py

@@ -50,9 +50,9 @@ def _ensure_telegram_mock():
 from gateway.config import Platform, PlatformConfig
 
 
-def _make_adapter():
+def _make_adapter(extra=None):
     """Create a TelegramAdapter with mocked internals."""
-    config = PlatformConfig(enabled=True, token="test-token")
+    config = PlatformConfig(enabled=True, token="test-token", extra=extra or {})
     adapter = TelegramAdapter(config)
     adapter._bot = AsyncMock()
     adapter._app = MagicMock()
@@ -134,6 +134,23 @@ async def test_not_connected(self):
         )
         assert result.success is False
 
+    @pytest.mark.asyncio
+    async def test_disable_link_previews_sets_preview_kwargs(self):
+        adapter = _make_adapter(extra={"disable_link_previews": True})
+        mock_msg = MagicMock()
+        mock_msg.message_id = 42
+        adapter._bot.send_message = AsyncMock(return_value=mock_msg)
+
+        await adapter.send_exec_approval(
+            chat_id="12345", command="ls", session_key="s"
+        )
+
+        kwargs = adapter._bot.send_message.call_args[1]
+        assert (
+            kwargs.get("disable_web_page_preview") is True
+            or kwargs.get("link_preview_options") is not None
+        )
+
     @pytest.mark.asyncio
     async def test_truncates_long_command(self):
         adapter = _make_adapter()

tests/tools/test_send_message_tool.py

@@ -576,7 +576,7 @@ def test_telegram_media_attaches_to_last_chunk(self):
 
         sent_calls = []
 
-        async def fake_send(token, chat_id, message, media_files=None, thread_id=None):
+        async def fake_send(token, chat_id, message, media_files=None, thread_id=None, disable_link_previews=False):
             sent_calls.append(media_files or [])
             return {"success": True, "platform": "telegram", "chat_id": chat_id, "message_id": str(len(sent_calls))}
 
@@ -658,6 +658,17 @@ def test_plain_text_uses_markdown_v2(self, monkeypatch):
         kwargs = bot.send_message.await_args.kwargs
         assert kwargs["parse_mode"] == "MarkdownV2"
 
+    def test_disable_link_previews_sets_disable_web_page_preview(self, monkeypatch):
+        bot = self._make_bot()
+        _install_telegram_mock(monkeypatch, bot)
+
+        asyncio.run(
+            _send_telegram("tok", "123", "https://example.com", disable_link_previews=True)
+        )
+
+        kwargs = bot.send_message.await_args.kwargs
+        assert kwargs["disable_web_page_preview"] is True
+
     def test_html_with_code_and_pre_tags(self, monkeypatch):
         bot = self._make_bot()
         _install_telegram_mock(monkeypatch, bot)

tools/send_message_tool.py

@@ -369,6 +369,7 @@ async def _send_to_platform(platform, pconfig, chat_id, message, thread_id=None,
     # --- Telegram: special handling for media attachments ---
     if platform == Platform.TELEGRAM:
         last_result = None
+        disable_link_previews = bool(getattr(pconfig, "extra", {}) and pconfig.extra.get("disable_link_previews"))
         for i, chunk in enumerate(chunks):
             is_last = (i == len(chunks) - 1)
             result = await _send_telegram(
@@ -377,6 +378,7 @@ async def _send_to_platform(platform, pconfig, chat_id, message, thread_id=None,
                 chunk,
                 media_files=media_files if is_last else [],
                 thread_id=thread_id,
+                disable_link_previews=disable_link_previews,
             )
             if isinstance(result, dict) and result.get("error"):
                 return result
@@ -461,7 +463,7 @@ async def _send_to_platform(platform, pconfig, chat_id, message, thread_id=None,
     return last_result
 
 
-async def _send_telegram(token, chat_id, message, media_files=None, thread_id=None):
+async def _send_telegram(token, chat_id, message, media_files=None, thread_id=None, disable_link_previews=False):
     """Send via Telegram Bot API (one-shot, no polling needed).
 
     Applies markdown→MarkdownV2 formatting (same as the gateway adapter)
@@ -497,6 +499,8 @@ async def _send_telegram(token, chat_id, message, media_files=None, thread_id=No
         thread_kwargs = {}
         if thread_id is not None:
             thread_kwargs["message_thread_id"] = int(thread_id)
+        if disable_link_previews:
+            thread_kwargs["disable_web_page_preview"] = True
 
         last_msg = None
         warnings = []