Merge pull request chigwell#168 from cyb3ralbert/feat/exposed-tools-allowlist

feat(runtime): allow read-only mode to expose named write tools

Author: chigwell

.env.example

@@ -39,6 +39,10 @@ TELEGRAM_SESSION_NAME=telegram_session
 # readOnlyHint=True through MCP. This does not reduce Telegram session authority
 # inside the server process.
 # TELEGRAM_EXPOSED_TOOLS=read-only
+#
+# 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
 
 # --- File-path tools / allowed roots (optional) ---
 # When the MCP client implements Roots but advertises an empty list — or when

README.md

@@ -122,11 +122,22 @@ clients from sending messages or performing chat/account mutations, set
 TELEGRAM_EXPOSED_TOOLS=read-only
 ```
 
+If read-only is too strict but `all` is too broad, append `+` and a
+comma-separated list of tool names to also expose those specific write tools.
+Every other write tool stays unregistered:
+
+```env
+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.
+
 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) and `read-only`.
+values are `all` (the default), `read-only`, and `read-only+<tool>,<tool>`.
 
 Run the server locally:
 
@@ -167,6 +178,12 @@ server `env` block:
 "TELEGRAM_EXPOSED_TOOLS": "read-only"
 ```
 
+Or keep read-only as the baseline and allow a few write tools on top:
+
+```json
+"TELEGRAM_EXPOSED_TOOLS": "read-only+send_message,reply_to_message"
+```
+
 Alternatively, install this repository directly from GitHub into a virtual
 environment using a specific release tag or commit:
 

telegram_mcp/runtime.py

@@ -149,32 +149,70 @@ async def annotated_handler(req):
 
 
 _EXPOSED_TOOLS_MODES = {"all", "read-only"}
+_EXPOSED_TOOLS_ALLOW_SEPARATOR = "+"
+
+
+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()]
 
 
 def _get_exposed_tools_mode(value: Optional[str] = None) -> str:
     """Return the configured MCP tool exposure mode.
 
     ``TELEGRAM_EXPOSED_TOOLS=read-only`` keeps only tools annotated with
-    ``readOnlyHint=True``. The default is ``all`` for backward compatibility.
+    ``readOnlyHint=True``. ``read-only+send_message,reply_to_message`` keeps
+    those plus the named write tools. The default is ``all`` for backward
+    compatibility.
     """
     raw_value = os.getenv("TELEGRAM_EXPOSED_TOOLS", "all") if value is None else value
     mode = raw_value.strip().lower()
-    if mode not in _EXPOSED_TOOLS_MODES:
+    base_mode, allowlist = _split_exposed_tools_mode(mode)
+    if base_mode not in _EXPOSED_TOOLS_MODES:
         accepted = ", ".join(sorted(_EXPOSED_TOOLS_MODES))
         raise SystemExit(
             f"Invalid TELEGRAM_EXPOSED_TOOLS '{raw_value}'. Expected one of: {accepted}."
         )
-    return mode
+    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"'{_EXPOSED_TOOLS_ALLOW_SEPARATOR}tool,tool' allowlist is only valid "
+            "with read-only."
+        )
+    if not allowlist:
+        raise SystemExit(
+            f"Invalid TELEGRAM_EXPOSED_TOOLS '{raw_value}'. The "
+            f"'{_EXPOSED_TOOLS_ALLOW_SEPARATOR}' allowlist must name at least one tool."
+        )
+    return f"{base_mode}{_EXPOSED_TOOLS_ALLOW_SEPARATOR}{','.join(allowlist)}"
 
 
 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)
-    if selected_mode == "all":
+    base_mode, allowlist = _split_exposed_tools_mode(selected_mode)
+    if base_mode == "all":
         return []
 
+    registered = {tool.name for tool in server._tool_manager.list_tools()}
+    unknown = sorted(set(allowlist) - registered)
+    if unknown:
+        # Fail loudly: a typo must not silently degrade into a narrower allowlist
+        # that looks like it worked.
+        raise SystemExit(
+            f"Invalid TELEGRAM_EXPOSED_TOOLS allowlist: unknown tool(s) {', '.join(unknown)}."
+        )
+
+    allowed = set(allowlist)
     removed: list[str] = []
     for tool in list(server._tool_manager.list_tools()):
+        if tool.name in allowed:
+            continue
         annotations = getattr(tool, "annotations", None)
         if not getattr(annotations, "readOnlyHint", False):
             server._tool_manager.remove_tool(tool.name)

tests/test_runtime.py

@@ -84,6 +84,55 @@ def test_get_exposed_tools_mode_rejects_invalid_value(monkeypatch):
     assert "read-only" in message
 
 
+def _synthetic_mcp_with_two_writes():
+    server = _synthetic_mcp()
+
+    @server.tool(annotations=ToolAnnotations(title="Send", destructiveHint=True))
+    def send_tool():
+        return "send"
+
+    return server
+
+
+def test_get_exposed_tools_mode_normalises_allowlist(monkeypatch):
+    monkeypatch.setenv("TELEGRAM_EXPOSED_TOOLS", " Read-Only+ send_tool , write_tool ")
+
+    assert runtime._get_exposed_tools_mode() == "read-only+send_tool,write_tool"
+
+
+def test_apply_exposed_tools_allowlist_keeps_named_write_tools():
+    server = _synthetic_mcp_with_two_writes()
+
+    removed = runtime._apply_exposed_tools_mode(server, "read-only+send_tool")
+
+    assert removed == ["write_tool"]
+    assert _tool_names(server) == {"read_tool", "send_tool"}
+
+
+def test_apply_exposed_tools_allowlist_rejects_unknown_tool():
+    server = _synthetic_mcp_with_two_writes()
+
+    with pytest.raises(SystemExit) as excinfo:
+        runtime._apply_exposed_tools_mode(server, "read-only+send_mesage")
+
+    assert "send_mesage" in str(excinfo.value)
+    assert _tool_names(server) == {"read_tool", "write_tool", "send_tool"}
+
+
+def test_get_exposed_tools_mode_rejects_allowlist_with_all():
+    with pytest.raises(SystemExit) as excinfo:
+        runtime._get_exposed_tools_mode("all+send_tool")
+
+    assert "read-only" in str(excinfo.value)
+
+
+def test_get_exposed_tools_mode_rejects_empty_allowlist():
+    with pytest.raises(SystemExit) as excinfo:
+        runtime._get_exposed_tools_mode("read-only+")
+
+    assert "at least one tool" in str(excinfo.value)
+
+
 def test_discover_accounts_supports_suffixed_and_default_sessions(monkeypatch):
     _clear_session_env(monkeypatch)
     monkeypatch.setenv("TELEGRAM_SESSION_STRING_WORK", "work-session")