Merge pull request #136 from iqdoctor/feat/exposed-tools-read-only

Add read-only exposed tools mode

Author: chigwell

.env.example

@@ -14,6 +14,12 @@ TELEGRAM_SESSION_NAME=telegram_session
 # TELEGRAM_SESSION_STRING_WORK=<session string for work account>
 # TELEGRAM_SESSION_STRING_PERSONAL=<session string for personal account>
 
+# --- MCP tool exposure (optional) ---
+# Default is all. Set to read-only to expose only tools annotated with
+# readOnlyHint=True through MCP. This does not reduce Telegram session authority
+# inside the server process.
+# TELEGRAM_EXPOSED_TOOLS=read-only
+
 # --- Proxy (optional) ---
 # Route Telegram traffic through a proxy. Set TELEGRAM_PROXY_TYPE to enable.
 # Supported types: socks5, socks4, http, mtproxy.

README.md

@@ -112,6 +112,21 @@ TELEGRAM_API_HASH=your_api_hash_here
 TELEGRAM_SESSION_STRING=your_session_string_here
 ```
 
+By default, all Telegram MCP tools are exposed. If you want to prevent MCP
+clients from sending messages or performing chat/account mutations, set
+`TELEGRAM_EXPOSED_TOOLS=read-only` to expose only tools annotated with
+`readOnlyHint=True`:
+
+```env
+TELEGRAM_EXPOSED_TOOLS=read-only
+```
+
+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`.
+
 Run the server locally:
 
 ```bash
@@ -144,6 +159,13 @@ this project:
 }
 ```
 
+To expose only read-only tools in Claude Desktop or Cursor, add this to the
+server `env` block:
+
+```json
+"TELEGRAM_EXPOSED_TOOLS": "read-only"
+```
+
 Alternatively, install this repository directly from GitHub into a virtual
 environment using a specific release tag or commit:
 

telegram_mcp/runner.py

@@ -7,6 +7,7 @@
 except UnsafeInstallationError as exc:
     raise SystemExit(str(exc)) from None
 
+from telegram_mcp import runtime as _runtime
 from telegram_mcp.runtime import *
 import telegram_mcp.tools  # noqa: F401 - registers MCP tools via decorators
 
@@ -60,6 +61,7 @@ async def _main() -> None:
 
 def main() -> None:
     _configure_allowed_roots_from_cli(sys.argv[1:])
+    _runtime._apply_exposed_tools_mode()
     nest_asyncio.apply()
     asyncio.run(_main())
 

telegram_mcp/runtime.py

@@ -137,6 +137,40 @@ async def annotated_handler(req):
 _install_annotation_hook()
 
 
+_EXPOSED_TOOLS_MODES = {"all", "read-only"}
+
+
+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.
+    """
+    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:
+        accepted = ", ".join(sorted(_EXPOSED_TOOLS_MODES))
+        raise SystemExit(
+            f"Invalid TELEGRAM_EXPOSED_TOOLS '{raw_value}'. Expected one of: {accepted}."
+        )
+    return mode
+
+
+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":
+        return []
+
+    removed: list[str] = []
+    for tool in list(server._tool_manager.list_tools()):
+        annotations = getattr(tool, "annotations", None)
+        if not getattr(annotations, "readOnlyHint", False):
+            server._tool_manager.remove_tool(tool.name)
+            removed.append(tool.name)
+    return removed
+
+
 # ---------------------------------------------------------------------------
 # Multi-account configuration
 # ---------------------------------------------------------------------------

tests/test_runtime.py

@@ -4,8 +4,9 @@
 from types import SimpleNamespace
 
 import pytest
+from mcp.server.fastmcp import FastMCP
 from mcp.shared.exceptions import McpError
-from mcp.types import ErrorData
+from mcp.types import ErrorData, ToolAnnotations
 from telethon.tl.types import Channel, Chat, PeerUser, User
 
 import main
@@ -24,6 +25,60 @@ def __init__(self, *args, **kwargs):
         self.kwargs = kwargs
 
 
+def _tool_names(server):
+    return {tool.name for tool in server._tool_manager.list_tools()}
+
+
+def _synthetic_mcp():
+    server = FastMCP("test")
+
+    @server.tool(annotations=ToolAnnotations(title="Read", readOnlyHint=True))
+    def read_tool():
+        return "read"
+
+    @server.tool(annotations=ToolAnnotations(title="Write", destructiveHint=True))
+    def write_tool():
+        return "write"
+
+    return server
+
+
+def test_get_exposed_tools_mode_defaults_to_all(monkeypatch):
+    monkeypatch.delenv("TELEGRAM_EXPOSED_TOOLS", raising=False)
+
+    assert runtime._get_exposed_tools_mode() == "all"
+
+
+def test_apply_exposed_tools_all_keeps_tools():
+    server = _synthetic_mcp()
+
+    removed = runtime._apply_exposed_tools_mode(server, "all")
+
+    assert removed == []
+    assert _tool_names(server) == {"read_tool", "write_tool"}
+
+
+def test_apply_exposed_tools_read_only_removes_non_read_only_tools():
+    server = _synthetic_mcp()
+
+    removed = runtime._apply_exposed_tools_mode(server, "read-only")
+
+    assert removed == ["write_tool"]
+    assert _tool_names(server) == {"read_tool"}
+
+
+def test_get_exposed_tools_mode_rejects_invalid_value(monkeypatch):
+    monkeypatch.setenv("TELEGRAM_EXPOSED_TOOLS", "send-everything")
+
+    with pytest.raises(SystemExit) as excinfo:
+        runtime._get_exposed_tools_mode()
+
+    message = str(excinfo.value)
+    assert "TELEGRAM_EXPOSED_TOOLS" in message
+    assert "all" in message
+    assert "read-only" in message
+
+
 def test_discover_accounts_supports_suffixed_and_default_sessions(monkeypatch):
     _clear_session_env(monkeypatch)
     monkeypatch.setenv("TELEGRAM_SESSION_STRING_WORK", "work-session")