Merge pull request chigwell#281 from vsem-azamat/feat/exposed-tools-exclusions

feat(runtime): exclude tools from TELEGRAM_EXPOSED_TOOLS with a - list

Author: chigwell

.env.example

@@ -64,6 +64,11 @@ TELEGRAM_SESSION_NAME=telegram_session
 # Append '+name,name' to read-only to also expose specific write tools. Every
 # other write tool stays unregistered. Unknown names abort startup.
 # TELEGRAM_EXPOSED_TOOLS=read-only+send_message,reply_to_message
+#
+# Append '-name,name' to either base (after any '+' list) to hide specific
+# tools. Unknown names, or names the base already hides, abort startup.
+# TELEGRAM_EXPOSED_TOOLS=all-delete_chat_history,delete_messages_bulk
+# TELEGRAM_EXPOSED_TOOLS=read-only+send_message-get_participants
 
 # --- Per-tool file extension allowlists (optional) ---
 # A hardcoded allowlist already restricts send_voice/send_sticker/

README.md

@@ -267,17 +267,33 @@ Every other write tool stays unregistered:
 TELEGRAM_EXPOSED_TOOLS=read-only+send_message,reply_to_message,send_file
 ```
 
-An unknown name in the allowlist aborts startup, so a typo cannot silently
-degrade into a narrower surface that looks like it worked.
+To hide specific tools instead, append `-` and a comma-separated list to either
+base. It works after a `+` list too:
+
+```env
+TELEGRAM_EXPOSED_TOOLS=all-delete_chat_history,delete_messages_bulk
+TELEGRAM_EXPOSED_TOOLS=read-only-get_participants
+TELEGRAM_EXPOSED_TOOLS=read-only+send_message,reply_to_message-get_participants
+```
+
+An unknown name in either list aborts startup, so a typo cannot silently
+produce a different surface that looks like it worked. Startup also aborts on
+an empty list, a name in both lists, or a `-` name the base already hides (such
+as `read-only-send_message`), which would be a no-op. To turn off voice
+transcription, set `TELEGRAM_TRANSCRIBE=off` rather than `-transcribe_voice`:
+hiding the tool leaves listings' pending-transcript hints and `auto` prefetch in
+place.
 
 At startup the server prints the names of the tools the mode hides to stderr;
-those are the names the `+` list accepts.
+those are the names the `+` list accepts. Tools hidden by the `-` list are
+printed on a separate line.
 
 This is an MCP tool-surface restriction, not a Telegram session sandbox or
 reduced Telegram account permission. The Telegram session string still has its
 normal authority inside the server process; read-only mode only prevents
 non-read-only tools from being registered and exposed through MCP. Accepted
-values are `all` (the default), `read-only`, and `read-only+<tool>,<tool>`.
+values are `all` (the default), `read-only`, and `read-only+<tool>,<tool>`,
+each optionally followed by `-<tool>,<tool>`.
 
 A separate, hardcoded allowlist restricts `send_voice`, `send_sticker`,
 `set_profile_photo`, and `edit_chat_photo` to their expected file extensions;

telegram_mcp/runner.py

@@ -314,7 +314,7 @@ def main() -> None:
     # send_file is not exposed is a valid configuration, so the name check
     # has to see the full tool set.
     _runtime._apply_file_extension_overrides()
-    hidden = _runtime._apply_exposed_tools_mode()
+    hidden, excluded = _runtime._apply_exposed_tools_mode()
     if hidden:
         # These names are the menu for the "+" allowlist; without this line the
         # only way to find them is reading the tool annotations in the source.
@@ -323,6 +323,12 @@ def main() -> None:
             f"after '+' to expose it: {', '.join(sorted(hidden))}",
             file=sys.stderr,
         )
+    if excluded:
+        print(
+            f"TELEGRAM_EXPOSED_TOOLS excludes {len(excluded)} tool(s) listed "
+            f"after '-': {', '.join(sorted(excluded))}",
+            file=sys.stderr,
+        )
     _transcription.validate_transcription_config()
     _session_lock_shared()  # fail loudly at startup on a bad toggle
     asyncio.run(_main())

telegram_mcp/runtime.py

@@ -20,7 +20,7 @@
 from difflib import SequenceMatcher
 from datetime import datetime, timedelta, timezone
 from enum import Enum
-from typing import List, Dict, Optional, Union, Any, Iterable, get_args
+from typing import List, Dict, NamedTuple, Optional, Union, Any, Iterable, get_args
 from pathlib import Path
 from urllib.parse import unquote, urlparse
 
@@ -241,74 +241,194 @@ async def annotated_handler(req):
 
 _EXPOSED_TOOLS_MODES = {"all", "read-only"}
 _EXPOSED_TOOLS_ALLOW_SEPARATOR = "+"
+_EXPOSED_TOOLS_EXCLUDE_SEPARATOR = "-"
+_EXPOSED_TOOLS_HYPHEN_HINT = (
+    f" Tool names use underscores; '{_EXPOSED_TOOLS_EXCLUDE_SEPARATOR}' starts the "
+    "exclusion list."
+)
 
 
-def _split_exposed_tools_mode(mode: str) -> tuple[str, list[str]]:
-    """Split a normalised exposure mode into its base mode and write allowlist."""
-    base, separator, raw_allowlist = mode.partition(_EXPOSED_TOOLS_ALLOW_SEPARATOR)
-    if not separator:
-        return base, []
-    return base, [name.strip() for name in raw_allowlist.split(",") if name.strip()]
+class _ExposedToolsMode(NamedTuple):
+    """A validated ``TELEGRAM_EXPOSED_TOOLS`` value."""
 
+    base: str
+    allowlist: list[str]
+    exclusions: list[str]
 
-def _get_exposed_tools_mode(value: Optional[str] = None) -> str:
-    """Return the configured MCP tool exposure mode.
+    def __str__(self) -> str:
+        normalised = self.base
+        if self.allowlist:
+            normalised += _EXPOSED_TOOLS_ALLOW_SEPARATOR + ",".join(self.allowlist)
+        if self.exclusions:
+            normalised += _EXPOSED_TOOLS_EXCLUDE_SEPARATOR + ",".join(self.exclusions)
+        return normalised
+
+
+class _ExposedToolsResult(NamedTuple):
+    """Tools removed by the base mode (``hidden``) and by the ``-`` list (``excluded``)."""
+
+    hidden: list[str]
+    excluded: list[str]
+
+
+def _split_tool_list(raw: Optional[str]) -> Optional[list[str]]:
+    """Split a comma-separated tool list; ``None`` means the separator was absent."""
+    if raw is None:
+        return None
+    return [name.strip() for name in raw.split(",") if name.strip()]
 
-    ``TELEGRAM_EXPOSED_TOOLS=read-only`` keeps only tools annotated with
-    ``readOnlyHint=True``. ``read-only+send_message,reply_to_message`` keeps
-    those plus the named write tools. The default is ``all`` for backward
-    compatibility.
+
+def _split_exposed_tools_value(
+    mode: str,
+) -> tuple[Optional[str], Optional[list[str]], Optional[list[str]]]:
+    """Split a lowercased value into base mode, ``+`` allowlist and ``-`` exclusions.
+
+    The base is matched as a known-mode prefix rather than by splitting on
+    ``-``, because ``read-only`` itself contains one. Tool names use
+    underscores, so a ``-`` after the base always starts the exclusion list.
+    A ``None`` base means no known mode matched; a ``None`` list means its
+    separator was absent.
+    """
+    for base in sorted(_EXPOSED_TOOLS_MODES, key=len, reverse=True):
+        rest = mode[len(base) :]
+        if mode.startswith(base) and rest[:1] in (
+            "",
+            _EXPOSED_TOOLS_ALLOW_SEPARATOR,
+            _EXPOSED_TOOLS_EXCLUDE_SEPARATOR,
+        ):
+            break
+    else:
+        return None, None, None
+
+    raw_allowlist: Optional[str] = None
+    raw_exclusions: Optional[str] = None
+    if rest.startswith(_EXPOSED_TOOLS_ALLOW_SEPARATOR):
+        raw_allowlist, separator, tail = rest[1:].partition(_EXPOSED_TOOLS_EXCLUDE_SEPARATOR)
+        if separator:
+            raw_exclusions = tail
+    elif rest.startswith(_EXPOSED_TOOLS_EXCLUDE_SEPARATOR):
+        raw_exclusions = rest[1:]
+    return base, _split_tool_list(raw_allowlist), _split_tool_list(raw_exclusions)
+
+
+def _parse_exposed_tools_mode(value: Optional[str] = None) -> _ExposedToolsMode:
+    """Parse and validate ``TELEGRAM_EXPOSED_TOOLS`` (or ``value``).
+
+    ``read-only`` keeps only tools annotated with ``readOnlyHint=True``.
+    ``read-only+send_message,reply_to_message`` keeps those plus the named
+    write tools. A trailing ``-tool,tool`` list hides named tools from either
+    base, e.g. ``all-delete_chat_history``. The default is ``all`` for
+    backward compatibility. Tool names are checked against the registry
+    later, in ``_apply_exposed_tools_mode``.
     """
     raw_value = os.getenv("TELEGRAM_EXPOSED_TOOLS", "all") if value is None else value
-    mode = raw_value.strip().lower()
-    base_mode, allowlist = _split_exposed_tools_mode(mode)
-    if base_mode not in _EXPOSED_TOOLS_MODES:
+    base_mode, allowlist, exclusions = _split_exposed_tools_value(raw_value.strip().lower())
+    invalid = f"Invalid TELEGRAM_EXPOSED_TOOLS '{raw_value}'."
+    if base_mode is None:
         accepted = ", ".join(sorted(_EXPOSED_TOOLS_MODES))
+        raise SystemExit(f"{invalid} Expected one of: {accepted}.")
+    if allowlist is not None and base_mode != "read-only":
         raise SystemExit(
-            f"Invalid TELEGRAM_EXPOSED_TOOLS '{raw_value}'. Expected one of: {accepted}."
-        )
-    if _EXPOSED_TOOLS_ALLOW_SEPARATOR not in mode:
-        return base_mode
-    if base_mode != "read-only":
-        raise SystemExit(
-            f"Invalid TELEGRAM_EXPOSED_TOOLS '{raw_value}'. The "
+            f"{invalid} The "
             f"'{_EXPOSED_TOOLS_ALLOW_SEPARATOR}tool,tool' allowlist is only valid "
             "with read-only."
         )
-    if not allowlist:
+    if allowlist is not None and not allowlist:
         raise SystemExit(
-            f"Invalid TELEGRAM_EXPOSED_TOOLS '{raw_value}'. The "
+            f"{invalid} The "
             f"'{_EXPOSED_TOOLS_ALLOW_SEPARATOR}' allowlist must name at least one tool."
         )
-    return f"{base_mode}{_EXPOSED_TOOLS_ALLOW_SEPARATOR}{','.join(allowlist)}"
+    if exclusions is not None:
+        _validate_exposed_tools_exclusions(invalid, allowlist or [], exclusions)
+    return _ExposedToolsMode(base_mode, allowlist or [], exclusions or [])
 
 
-def _apply_exposed_tools_mode(server: FastMCP = mcp, mode: Optional[str] = None) -> list[str]:
-    """Prune registered MCP tools according to the configured exposure mode."""
-    selected_mode = _get_exposed_tools_mode() if mode is None else _get_exposed_tools_mode(mode)
-    base_mode, allowlist = _split_exposed_tools_mode(selected_mode)
-    if base_mode == "all":
-        return []
+def _get_exposed_tools_mode(value: Optional[str] = None) -> str:
+    """Return the configured MCP tool exposure mode, normalised."""
+    return str(_parse_exposed_tools_mode(value))
+
+
+def _validate_exposed_tools_exclusions(
+    invalid: str, allowlist: list[str], exclusions: list[str]
+) -> None:
+    """Reject syntax mistakes in the ``-`` exclusion list."""
+    sep = _EXPOSED_TOOLS_EXCLUDE_SEPARATOR
+    if not exclusions:
+        raise SystemExit(f"{invalid} The '{sep}' exclusion list must name at least one tool.")
+    if any(_EXPOSED_TOOLS_ALLOW_SEPARATOR in name or sep in name for name in exclusions):
+        raise SystemExit(
+            f"{invalid} Use one '{_EXPOSED_TOOLS_ALLOW_SEPARATOR}' list followed by one "
+            f"'{sep}' list, each comma-separated."
+        )
+    conflicting = sorted(set(allowlist) & set(exclusions))
+    if conflicting:
+        raise SystemExit(
+            f"{invalid} Tool(s) both exposed with '{_EXPOSED_TOOLS_ALLOW_SEPARATOR}' and "
+            f"excluded with '{sep}': {', '.join(conflicting)}."
+        )
+
 
-    registered = {tool.name for tool in server._tool_manager.list_tools()}
-    unknown = sorted(set(allowlist) - registered)
+def _reject_unknown_exposed_tools(
+    kind: str, names: list[str], registered: set[str], hint: str = ""
+) -> None:
+    unknown = sorted(set(names) - registered)
     if unknown:
-        # Fail loudly: a typo must not silently degrade into a narrower allowlist
+        # Fail loudly: a typo must not silently produce a different surface
         # that looks like it worked.
         raise SystemExit(
-            f"Invalid TELEGRAM_EXPOSED_TOOLS allowlist: unknown tool(s) {', '.join(unknown)}."
+            f"Invalid TELEGRAM_EXPOSED_TOOLS {kind}: unknown tool(s) "
+            f"{', '.join(unknown)}.{hint}"
         )
 
-    allowed = set(allowlist)
-    removed: list[str] = []
-    for tool in list(server._tool_manager.list_tools()):
-        if tool.name in allowed:
+
+def _apply_exposed_tools_mode(
+    server: FastMCP = mcp, mode: Optional[str] = None
+) -> _ExposedToolsResult:
+    """Prune registered MCP tools according to the configured exposure mode."""
+    selected = _parse_exposed_tools_mode(mode)
+    if selected.base == "all" and not selected.exclusions:
+        return _ExposedToolsResult([], [])
+
+    tools = list(server._tool_manager.list_tools())
+    registered = {tool.name for tool in tools}
+    # "read-only+send-message" splits into "+send" and "-message"; point at the
+    # hyphen when a "-" list could be the result of one.
+    hint = _EXPOSED_TOOLS_HYPHEN_HINT if selected.exclusions else ""
+    _reject_unknown_exposed_tools("allowlist", selected.allowlist, registered, hint)
+    _reject_unknown_exposed_tools("exclusion list", selected.exclusions, registered)
+
+    allowed = set(selected.allowlist)
+    excluded = set(selected.exclusions)
+
+    def base_exposes(tool) -> bool:
+        if selected.base == "all" or tool.name in allowed:
+            return True
+        return bool(getattr(getattr(tool, "annotations", None), "readOnlyHint", False))
+
+    already_hidden = sorted(
+        tool.name for tool in tools if tool.name in excluded and not base_exposes(tool)
+    )
+    if already_hidden:
+        # A no-op exclusion usually means the operator expected a different base,
+        # or typed "-" for "_" inside a "+" name.
+        verb, pronoun = ("is", "it") if len(already_hidden) == 1 else ("are", "them")
+        raise SystemExit(
+            f"Invalid TELEGRAM_EXPOSED_TOOLS exclusion list: {', '.join(already_hidden)} "
+            f"{verb} already hidden by {selected.base}; remove {pronoun} from the "
+            f"'{_EXPOSED_TOOLS_EXCLUDE_SEPARATOR}' list."
+            f"{_EXPOSED_TOOLS_HYPHEN_HINT if selected.allowlist else ''}"
+        )
+
+    result = _ExposedToolsResult([], [])
+    for tool in tools:
+        if tool.name in excluded:
+            result.excluded.append(tool.name)
+        elif not base_exposes(tool):
+            result.hidden.append(tool.name)
+        else:
             continue
-        annotations = getattr(tool, "annotations", None)
-        if not getattr(annotations, "readOnlyHint", False):
-            server._tool_manager.remove_tool(tool.name)
-            removed.append(tool.name)
-    return removed
+        server._tool_manager.remove_tool(tool.name)
+    return result
 
 
 _FILE_EXTENSION_TOKEN_PATTERN = re.compile(r"^\.[A-Za-z0-9_-]+$")

tests/test_runner.py

@@ -472,7 +472,7 @@ def test_file_extension_overrides_are_validated_before_tools_are_pruned(monkeypa
     monkeypatch.setattr(
         runner._runtime,
         "_apply_exposed_tools_mode",
-        lambda *a, **k: calls.append("exposed") or [],
+        lambda *a, **k: calls.append("exposed") or ([], []),
     )
     monkeypatch.setattr(
         runner._runtime,
@@ -505,7 +505,7 @@ def test_cli_transport_flag_sets_env_var(monkeypatch):
     monkeypatch.setattr(runner._runtime, "_CLI_HOST", None)
     monkeypatch.setattr(runner._runtime, "_CLI_PORT", None)
     monkeypatch.setattr(runner._runtime, "_apply_file_extension_overrides", lambda: None)
-    monkeypatch.setattr(runner._runtime, "_apply_exposed_tools_mode", lambda: None)
+    monkeypatch.setattr(runner._runtime, "_apply_exposed_tools_mode", lambda: ([], []))
     monkeypatch.setattr(runner._transcription, "validate_transcription_config", lambda: None)
     monkeypatch.setattr(runner, "_session_lock_shared", lambda: None)
     monkeypatch.setattr(runner.asyncio, "run", lambda coro: coro.close())
@@ -566,7 +566,7 @@ def test_main_prints_the_tools_exposure_hides(monkeypatch, capsys):
     monkeypatch.setattr(
         runner._runtime,
         "_apply_exposed_tools_mode",
-        lambda: ["send_message", "export_chat_invite"],
+        lambda: (["send_message", "export_chat_invite"], []),
     )
     monkeypatch.setattr(runner._transcription, "validate_transcription_config", lambda: None)
     monkeypatch.setattr(runner, "_session_lock_shared", lambda: None)
@@ -577,3 +577,35 @@ def test_main_prints_the_tools_exposure_hides(monkeypatch, capsys):
     err = capsys.readouterr().err
     assert "hides 2 tool(s)" in err
     assert "export_chat_invite, send_message" in err
+
+
+def _run_main_with_exposure_result(monkeypatch, result):
+    monkeypatch.setattr(runner, "_configure_allowed_roots_from_cli", lambda *a, **k: None)
+    monkeypatch.setattr(runner._runtime, "_apply_file_extension_overrides", lambda: None)
+    monkeypatch.setattr(runner._runtime, "_apply_exposed_tools_mode", lambda: result)
+    monkeypatch.setattr(runner._transcription, "validate_transcription_config", lambda: None)
+    monkeypatch.setattr(runner, "_session_lock_shared", lambda: None)
+    monkeypatch.setattr(runner.asyncio, "run", lambda coro: coro.close())
+
+    runner.main()
+
+
+def test_main_reports_excluded_tools_separately(monkeypatch, capsys):
+    """Excluded names are not '+' candidates, so they get their own line."""
+    _run_main_with_exposure_result(
+        monkeypatch, (["delete_message", "ban_user"], ["list_chats", "get_participants"])
+    )
+
+    err = capsys.readouterr().err
+    assert (
+        "hides 2 tool(s); list any of them after '+' to expose it: ban_user, delete_message" in err
+    )
+    assert "excludes 2 tool(s) listed after '-': get_participants, list_chats" in err
+
+
+def test_main_reports_only_exclusions_for_all_base(monkeypatch, capsys):
+    _run_main_with_exposure_result(monkeypatch, ([], ["delete_message", "delete_messages_bulk"]))
+
+    err = capsys.readouterr().err
+    assert "after '+'" not in err
+    assert "excludes 2 tool(s) listed after '-': delete_message, delete_messages_bulk" in err