From 494c5b77c1019463b8e0479523499a138a365b04 Mon Sep 17 00:00:00 2001
From: speech115 <201258852+speech115@users.noreply.github.com>
Date: Fri, 26 Jun 2026 19:09:35 +0400
Subject: [PATCH] Fix Telegram MCP error boundaries
---
README.md | 101 ++++--
control-plane/.gitignore | 1 +
control-plane/tests/test_command_registry.py | 4 +-
control-plane/tests/test_surface_docs.py | 14 +-
docs/operator-workflows.md | 63 ++--
mcp/src/telegram_mcp/__init__.py | 2 +
mcp/src/telegram_mcp/approval_server.py | 79 ++++-
mcp/src/telegram_mcp/client_groups.py | 10 +-
mcp/src/telegram_mcp/errors.py | 40 +++
mcp/src/telegram_mcp/fast_read_today.py | 189 ++--------
mcp/src/telegram_mcp/mcp_http_client.py | 57 ++-
mcp/src/telegram_mcp/mcp_prewarm.py | 2 +-
mcp/src/telegram_mcp/member_export_paths.py | 4 +
mcp/src/telegram_mcp/runtime.py | 8 +-
mcp/src/telegram_mcp/send_confirmation.py | 162 +++++----
mcp/src/telegram_mcp/tg_cli.py | 335 ++++++++----------
mcp/src/telegram_mcp/tools/__init__.py | 1 -
.../telegram_mcp/tools/dialog_facade_tools.py | 16 +-
mcp/src/telegram_mcp/tools/group_tools.py | 18 +-
mcp/src/telegram_mcp/tools/media_tools.py | 2 -
mcp/tests/test_dialog_facade_tools.py | 28 ++
mcp/tests/test_fast_read_today.py | 31 ++
mcp/tests/test_mcp_http_client.py | 64 +++-
mcp/tests/test_member_export_paths.py | 14 +
mcp/tests/test_registration.py | 1 +
mcp/tests/test_runtime.py | 27 ++
mcp/tests/test_send_confirmation.py | 63 +++-
mcp/tests/test_tg_cli.py | 47 ++-
28 files changed, 836 insertions(+), 547 deletions(-)
diff --git a/README.md b/README.md
index ddaa723..78e3ef5 100644
--- a/README.md
+++ b/README.md
@@ -18,33 +18,35 @@
> Community-maintained integration. Not an official Telegram product or
> Telegram LLC publication.
-Telegram Plugin packages a working local stack for agent-safe Telegram access:
+Telegram Plugin packages a working local stack for owner-local Telegram access:
a Telethon-backed MCP server, a Codex plugin bundle, and an optional
control-plane for local audits and repair planning.
-Most Telegram automation tools expose too much too quickly. This project takes
-the opposite position: agents should start with a narrow read/search/context
-surface, clear source routing, drift checks, and explicit boundaries around
-sessions, media, archives, and write actions.
+Most Telegram automation tools hide their safety boundary. This project takes
+the opposite position: the owner-local full MCP surface is explicit, the
+restricted facade profile is still available for narrow compatibility, and
+control-plane checks make drift visible.
## Why It Exists
-Use this project when you want an AI coding agent to inspect Telegram context
-without handing it the whole account by default.
+Use this project when you want an AI coding agent to work with a local Telegram
+account under an explicit owner-controlled MCP contract.
- **Local-first:** Telegram credentials and sessions stay on your machine.
-- **Narrow by default:** Default Mode reads, searches, collects context, drafts,
- previews, downloads scoped media, and transcribes voice locally.
-- **Explicit writes:** sending, admin actions, profile changes, and broad export
- workflows require separate Power Mode or operator wiring.
+- **Owner-local full surface:** `plugin/.mcp.json` points at the owner-local
+ MCP daemons and intentionally does not use a legacy `allowedTools` allowlist.
+- **Restricted facade available:** set `TELEGRAM_MCP_TOOL_PROFILE=facade` when
+ you need the narrow read/search/context/draft surface.
+- **Explicit write discipline:** sending, admin actions, profile changes, and
+ broad export workflows are visible in the full surface and should be routed
+ through preview/confirmation or operator workflows when risk warrants it.
- **Auditable setup:** contract smoke checks, plugin drift checks, and
control-plane reports make the local state explainable.
## What Is Included
- `mcp/` - a Telethon-backed MCP server with high-level dialog facade tools.
-- `plugin/` - a Codex plugin bundle that points at a local MCP daemon and
- exposes a restricted default allowlist.
+- `plugin/` - a Codex plugin bundle that points at local owner MCP daemons.
- `control-plane/` - optional local doctor/status/audit commands for plugin
drift, LaunchAgent inventory, sessions, source routing, and repair planning.
- `docs/` - safety model and routing notes for operating the stack.
@@ -53,32 +55,54 @@ without handing it the whole account by default.
| Mode | Use it for | Can change Telegram? | Enabled by default |
| --- | --- | --- | --- |
-| Default Mode | read, search, context, drafts, previews, scoped media, voice transcription | No direct writes | Yes |
-| Power Mode | send/reply, contacts, groups/channels, stories, profile, privacy state | Yes | No |
+| Owner Local Full MCP | owner-controlled live work across the local full Telegram surface | Yes | Yes |
+| Facade Profile | read, search, context, drafts, previews, scoped media inspection/download | No direct writes | No |
+| Power Mode Example | wildcard client config for the same full local surface | Yes | No |
| Operator Workflows | mirror/archive, subscriber export, control-plane repair and audits | Can read in bulk or create sensitive artifacts | No |
-Power Mode is not a hidden feature. It is the explicit mode for users who want
-the broader Telegram MCP surface and accept that agents can perform externally
-visible actions.
+The full surface is not a hidden feature. It is the explicit owner-local mode
+for users who want agents to work with the broader Telegram MCP surface and
+accept that tools can perform externally visible actions.
For a short, private-data-free example, see
[Default Mode demo](docs/demo-default-mode.md).
+## Surface Contract
+
+Current healthy local mode is `owner_local_full_mcp`.
+
+- `plugin/.mcp.json` points at owner-local MCP daemons and intentionally exposes
+ their full local surface without a legacy `allowedTools` allowlist.
+- `TELEGRAM_MCP_TOOL_PROFILE=default` is not a restricted profile. In current
+ runtime behavior, unset/unknown/default profile names register the full
+ surface.
+- The restricted facade profile is explicit: use
+ `TELEGRAM_MCP_TOOL_PROFILE=facade` (or `safe` / `restricted`) for narrow
+ read/search/context/draft workflows.
+- `plugin/.mcp.full.example.json` is only a wildcard client example, not the
+ only way to reach full MCP.
+
+## Release Gate
+
+Before publishing or treating the local stack as healthy, run the release gates
+that compare runtime tools, plugin metadata, docs, and control-plane policy.
+
## Safety Model
-The runtime boundary is enforced in two layers:
+The runtime boundary is enforced by explicit local operator choice and audit
+checks:
-- MCP runtime profile (`TELEGRAM_MCP_TOOL_PROFILE=default`) keeps write/admin
- tools outside the default served surface.
-- Plugin allowlist (`plugin/.mcp.json`) exposes only Default Mode facade tools
- even when a broader local daemon exists.
+- Owner-local full MCP is represented by
+ `control-plane/policy/surface-contract.json`.
+- Restricted facade mode is represented by explicit MCP profile selection
+ (`TELEGRAM_MCP_TOOL_PROFILE=facade`, `safe`, or `restricted`).
- HTTP/SSE daemon transports require `TELEGRAM_MCP_AUTH_TOKEN`; stdio remains
local process-only.
```mermaid
flowchart LR
- Agent["AI agent"] --> Plugin["Codex plugin
restricted allowlist"]
- Plugin --> MCP["Local Telegram MCP
default profile"]
+ Agent["AI agent"] --> Plugin["Codex plugin
owner-local config"]
+ Plugin --> MCP["Local Telegram MCP
owner_local_full_mcp"]
MCP --> Telethon["Telethon session
on this machine"]
Control["Control-plane audits"] -.-> Plugin
Control -.-> MCP
@@ -155,18 +179,18 @@ manual `.mcp.json` wiring only as a fallback:
- For HTTP daemon mode, set `TELEGRAM_MCP_AUTH_TOKEN` in client environment;
the plugin MCP config references it via `bearer_token_env_var`.
-6. Unified workflow rule: use facade tools in Default Mode first. Switch to
-Power Mode only for explicit write/admin operations. Direct Telethon calls are
-an operator/debug path, not normal user onboarding.
+6. Unified workflow rule: use task-shaped tools first, and treat direct
+write/admin operations as explicit owner-local actions. Direct Telethon calls
+are an operator/debug path, not normal user onboarding.
-To inspect the full server surface locally, run the daemon with:
+To inspect a restricted facade server surface locally, run the daemon with:
```bash
-TELEGRAM_MCP_POWER_MODE=enabled TELEGRAM_MCP_TOOL_PROFILE=full .venv/bin/telegram-mcp
+TELEGRAM_MCP_TOOL_PROFILE=facade .venv/bin/telegram-mcp
```
-Use `plugin/.mcp.full.example.json` only when you intentionally want Power Mode
-in a local agent client.
+Use `plugin/.mcp.full.example.json` only when you intentionally want a wildcard
+client config for the full local surface.
Dependency strategy for `mcp/`: commit and review `uv.lock` changes for
reproducible installs. Do not run broad upgrades as part of routine docs or
@@ -222,13 +246,14 @@ blocks those by default.
## Capability Boundaries
-- Default Mode: live read/search/context/draft/preview, scoped local media
- download, and selected voice/video transcription.
-- Power Mode: broader Telegram API operations, including writes, contacts,
- groups/channels, stories, profile, and privacy state. This is opt-in and not
- enabled by the Default Mode allowlist.
+- Owner Local Full MCP: broader Telegram API operations, including writes,
+ contacts, groups/channels, stories, profile, and privacy state. This is the
+ current local owner default.
+- Facade Profile: live read/search/context/draft/preview, scoped local media
+ inspection/download, and selected voice/video transcription. This is opt-in
+ via `TELEGRAM_MCP_TOOL_PROFILE=facade`.
- Operator Workflows: mirror/archive/subscriber export/control-plane work. These
- are not Default Mode functions and require their own setup and safety checks.
+ require their own setup and safety checks.
## Safety Defaults
diff --git a/control-plane/.gitignore b/control-plane/.gitignore
index 544171a..b3f9f7d 100644
--- a/control-plane/.gitignore
+++ b/control-plane/.gitignore
@@ -2,6 +2,7 @@
__pycache__/
*.pyc
.DS_Store
+*.egg-info/
.cursor/
backups/
outputs/
diff --git a/control-plane/tests/test_command_registry.py b/control-plane/tests/test_command_registry.py
index 5e3945c..2fd2016 100644
--- a/control-plane/tests/test_command_registry.py
+++ b/control-plane/tests/test_command_registry.py
@@ -24,7 +24,9 @@ def bin_wrapper_names() -> set[str]:
return {
path.name
for path in BIN_DIR.iterdir()
- if path.is_file() and path.name not in NON_COMMAND_BIN
+ if path.is_file()
+ and not path.name.startswith(".")
+ and path.name not in NON_COMMAND_BIN
}
diff --git a/control-plane/tests/test_surface_docs.py b/control-plane/tests/test_surface_docs.py
index 1852413..5470535 100644
--- a/control-plane/tests/test_surface_docs.py
+++ b/control-plane/tests/test_surface_docs.py
@@ -2,12 +2,24 @@
from pathlib import Path
+REPO_ROOT = Path(__file__).resolve().parents[2]
+
def test_readme_surface_contract_uses_owner_local_full_mcp_as_healthy_default() -> None:
- readme = Path("README.md").read_text(encoding="utf-8")
+ readme = (REPO_ROOT / "README.md").read_text(encoding="utf-8")
surface_start = readme.index("## Surface Contract")
release_start = readme.index("## Release Gate")
surface_section = readme[surface_start:release_start]
assert "owner_local_full_mcp" in surface_section
+ assert "TELEGRAM_MCP_TOOL_PROFILE=default" in surface_section
+ assert "is not a restricted profile" in surface_section
assert "must not expose raw send/reply" not in surface_section
+
+
+def test_operator_workflows_do_not_claim_default_profile_is_restricted() -> None:
+ text = (REPO_ROOT / "docs/operator-workflows.md").read_text(encoding="utf-8")
+
+ assert "owner_local_full_mcp" in text
+ assert "TELEGRAM_MCP_TOOL_PROFILE=default" not in text
+ assert "TELEGRAM_MCP_TOOL_PROFILE=facade" in text
diff --git a/docs/operator-workflows.md b/docs/operator-workflows.md
index 88b4733..f379bcc 100644
--- a/docs/operator-workflows.md
+++ b/docs/operator-workflows.md
@@ -1,31 +1,47 @@
# Power Mode And Operator Workflows
-Default Mode is intentionally safe for first install. The repository also
-contains a full Telegram MCP surface and operator workflows. Those are not
-second-class features; they are explicit modes because they can change Telegram
-state, read in bulk, or create sensitive artifacts.
+The current local owner setup uses `owner_local_full_mcp`. The repository also
+keeps a restricted facade profile and broader operator workflows. These modes
+must stay explicit because they differ in how much Telegram state they can read
+or change.
Unified workflow for users:
-- Default path: Telegram facade in Default Mode.
-- Escalation path: Power Mode for explicit write/admin requests.
+- Default owner path: local full MCP on explicit owner accounts.
+- Restricted path: Telegram facade profile for read/search/context/draft work.
- Non-default path: direct Telethon is operator/debug only.
-## Default Mode
+## Owner Local Full MCP
-Default Mode is the normal plugin path. It supports live read/search/context,
-non-sending drafts and previews, scoped local media download, and selected
-voice/video transcription.
+Owner local full MCP is the normal local plugin path in this repository. It
+exposes the full Telegram MCP surface on explicit owner accounts and relies on
+local operator discipline, bearer auth, and control-plane checks.
-Default Mode should be safe to try without the agent sending messages, changing
-chats, changing profile state, or launching background mirror/archive jobs.
+This boundary is runtime-enforced when these controls are kept intact:
-This boundary is runtime-enforced when both controls are kept intact:
-
-- MCP daemon runs with `TELEGRAM_MCP_TOOL_PROFILE=default`.
+- MCP daemon runs with unset/default/full profile for the owner local full
+ surface, or with an explicit account-specific launchd config.
- HTTP/SSE daemon transport has `TELEGRAM_MCP_AUTH_TOKEN` configured in both
server and client.
-- Plugin MCP config stays on `plugin/.mcp.json` allowlist and is not replaced by
- `plugin/.mcp.full.example.json`.
+- Plugin MCP config stays on `plugin/.mcp.json` without legacy `allowedTools`
+ allowlists.
+- Control-plane policy stays on `owner_local_full_mcp`.
+
+## Facade Profile
+
+Facade profile is the restricted compatibility path. It supports live
+read/search/context, non-sending drafts and previews, scoped local media
+download, and selected voice/video transcription.
+
+Facade profile should be safe to try without the agent sending messages,
+changing chats, changing profile state, or launching background mirror/archive
+jobs.
+
+Enable it intentionally:
+
+```bash
+cd mcp
+TELEGRAM_MCP_TOOL_PROFILE=facade .venv/bin/telegram-mcp
+```
## Power Mode
@@ -41,14 +57,13 @@ TELEGRAM_MCP_POWER_MODE=enabled TELEGRAM_MCP_TOOL_PROFILE=full .venv/bin/telegra
```
Then point a local client at the same MCP endpoint using
-`plugin/.mcp.full.example.json` as a starting point.
+`plugin/.mcp.full.example.json` as a wildcard client-config starting point.
-Power Mode is enforced by explicit operator choice (runtime profile + allowlist
-switch). It is not reachable through the default plugin files alone.
+Power Mode is enforced by explicit operator choice and local client config.
Before using Power Mode:
-- verify Default Mode first;
+- verify owner-local full MCP status first;
- use a test chat or explicit stable target;
- keep exact message text and target identity stable;
- do not copy Power Mode allowlists into `plugin/.mcp.json`;
@@ -65,7 +80,7 @@ Operator Workflows are broader local operations around the Telegram stack:
These workflows often involve allowlists, local databases, background jobs,
freshness checks, or PII-heavy outputs. They are intentionally separate from
-Default Mode and Power Mode.
+owner-local full MCP and facade profile.
## Mirror
@@ -81,7 +96,7 @@ Mirror use should be:
- kept cold unless there is a separate runtime plan.
The control-plane can audit mirror state, but it does not turn mirror runtime
-jobs into Default Mode tools.
+jobs into facade tools.
## Telecrawl Archive
@@ -95,7 +110,7 @@ freshness, and known import gaps.
## Subscriber Export
Subscriber/member export is sensitive and can produce PII-heavy local artifacts.
-It is not exposed in the Default Mode allowlist.
+It is not part of the facade profile.
Use subscriber export only with explicit user intent, private local output
paths, and clear reporting of `visible_count`, `exported_count`, `missing`, and
diff --git a/mcp/src/telegram_mcp/__init__.py b/mcp/src/telegram_mcp/__init__.py
index 4cbd20a..b4ac1c3 100644
--- a/mcp/src/telegram_mcp/__init__.py
+++ b/mcp/src/telegram_mcp/__init__.py
@@ -3,3 +3,5 @@
from .telethon_compat import apply_telethon_compat
apply_telethon_compat()
+
+from . import mcp_prewarm as mcp_prewarm # noqa: E402,F401
diff --git a/mcp/src/telegram_mcp/approval_server.py b/mcp/src/telegram_mcp/approval_server.py
index 36b8a5d..5b0d425 100644
--- a/mcp/src/telegram_mcp/approval_server.py
+++ b/mcp/src/telegram_mcp/approval_server.py
@@ -50,6 +50,48 @@ def _send(self, status: int, content_type: str, body: bytes) -> None:
self.end_headers()
self.wfile.write(body)
+ def _form_fields(self) -> dict[str, str]:
+ length = int(self.headers.get("Content-Length") or 0)
+ raw = self.rfile.read(length).decode("utf-8") if length else ""
+ return {
+ key: values[0]
+ for key, values in parse_qs(raw).items()
+ if values
+ }
+
+ def _mutate(self, *, token: str, nonce: str, action: str) -> str:
+ store = get_confirmation_store()
+ record = store.get(token)
+ if record is None:
+ raise ToolContractError("invalid_confirmation_token", "confirmation token is unknown")
+ if nonce != record.one_time_nonce:
+ raise ToolContractError("invalid_confirmation_token", "approval nonce is invalid")
+ if action == "approve":
+ store.approve(token)
+ return "Одобрено
Можно отправлять через telegram_confirmed_send.
"
+ if action == "reject":
+ store.reject(token)
+ return "Отклонено
Отправка заблокирована.
"
+ raise ToolContractError("invalid_input", "unknown approval action")
+
+ def do_POST(self) -> None: # noqa: N802
+ parsed = urlparse(self.path)
+ if parsed.path not in {"/telegram/approve", "/telegram/approve/"}:
+ self._send(404, "text/plain", b"not found\n")
+ return
+ fields = self._form_fields()
+ token = fields.get("token") or ""
+ nonce = fields.get("nonce") or ""
+ action = fields.get("action") or ""
+ try:
+ inner = self._mutate(token=token, nonce=nonce, action=action)
+ status = 200
+ except ToolContractError as exc:
+ inner = f"Ошибка
{html.escape(exc.message)}
"
+ status = 400
+ page_status, ctype, body = _page("Telegram approve", inner, status=status)
+ self._send(page_status, ctype, body)
+
def do_GET(self) -> None: # noqa: N802
parsed = urlparse(self.path)
if parsed.path not in {"/telegram/approve", "/telegram/approve/"}:
@@ -101,23 +143,22 @@ def do_GET(self) -> None: # noqa: N802
Кому: {chat} · инструмент: {tool} · статус: {state}
{reply_line}
{text}
-
-После одобрения агент может вызвать telegram_confirmed_send с тем же текстом.
-"""
-
- action = (parse_qs(parsed.query).get("action") or [None])[0]
- if action == "approve" and record.approval_state == "pending":
- try:
- store.approve(token)
- inner = "Одобрено
Можно отправлять через telegram_confirmed_send.
" + inner
- except ToolContractError as exc:
- inner = f"Ошибка
{html.escape(exc.message)}
"
- elif action == "reject" and record.approval_state == "pending":
- store.reject(token)
- inner = "Отклонено
Отправка заблокирована.
"
+
+
+
+
+ После одобрения агент может вызвать telegram_confirmed_send с тем же текстом.
+ """
status, ctype, body = _page("Telegram approve", inner)
self._send(status, ctype, body)
@@ -125,6 +166,8 @@ def do_GET(self) -> None: # noqa: N802
def start_approval_server(*, host: str, port: int) -> None:
global _server
+ if host not in {"127.0.0.1", "localhost", "::1"}:
+ raise ValueError("approval server host must be loopback")
with _server_lock:
if _server is not None:
return
@@ -145,4 +188,4 @@ def stop_approval_server() -> None:
return
_server.shutdown()
_server.server_close()
- _server = None
\ No newline at end of file
+ _server = None
diff --git a/mcp/src/telegram_mcp/client_groups.py b/mcp/src/telegram_mcp/client_groups.py
index c0ea02f..d0702ac 100644
--- a/mcp/src/telegram_mcp/client_groups.py
+++ b/mcp/src/telegram_mcp/client_groups.py
@@ -181,7 +181,7 @@ async def get_participants(
return participants, total
async def get_admins(
- self, chat: str | int
+ self, chat: str | int, limit: int = 200
) -> list[Participant]:
entity = await self._resolve_entity(chat)
admins = []
@@ -193,7 +193,7 @@ async def get_admins(
channel=entity,
filter=ChannelParticipantsAdmins(),
offset=0,
- limit=200,
+ limit=limit,
hash=0,
)
),
@@ -205,7 +205,7 @@ async def get_admins(
role = "creator" if "Creator" in type(p).__name__ else "admin"
admins.append(self._user_to_participant(user, role))
else:
- async for user in self.client.iter_participants(entity, aggressive=False):
+ async for user in self.client.iter_participants(entity, aggressive=False, limit=limit):
ptype = type(getattr(user, "participant", None)).__name__
if "Creator" in ptype or "Admin" in ptype:
role = "creator" if "Creator" in ptype else "admin"
@@ -213,7 +213,7 @@ async def get_admins(
return admins
async def get_banned_users(
- self, chat: str | int
+ self, chat: str | int, limit: int = 200
) -> list[Participant]:
entity = await self._resolve_entity(chat)
banned = []
@@ -225,7 +225,7 @@ async def get_banned_users(
channel=entity,
filter=ChannelParticipantsBanned(""),
offset=0,
- limit=200,
+ limit=limit,
hash=0,
)
),
diff --git a/mcp/src/telegram_mcp/errors.py b/mcp/src/telegram_mcp/errors.py
index 0a882c4..7ad0ac2 100644
--- a/mcp/src/telegram_mcp/errors.py
+++ b/mcp/src/telegram_mcp/errors.py
@@ -32,6 +32,46 @@ def __init__(self, code: str, message: str):
super().__init__(f"{code}: {message}")
+CONTRACT_ERROR_CODES = frozenset(
+ {
+ "archive_fallback_blocked",
+ "archive_route_blocked",
+ "confirmation_payload_mismatch",
+ "confirmation_rejected",
+ "expired_confirmation_token",
+ "human_approval_required",
+ "invalid_confirmation_token",
+ "invalid_context_mode",
+ "invalid_date_range",
+ "invalid_input",
+ "invalid_intent",
+ "live_intent_conflict",
+ "missing_confirmation_token",
+ "missing_send_target",
+ "permission_denied",
+ "rate_limited",
+ "telegram_tool_error",
+ "transport_unavailable",
+ }
+)
+
+
+def normalize_contract_error_code(value: object) -> str:
+ return str(value).strip().lower().split(":", 1)[0].strip()
+
+
+def payload_is_contract_error(payload: object | None) -> bool:
+ if isinstance(payload, str):
+ return normalize_contract_error_code(payload) in CONTRACT_ERROR_CODES
+ if isinstance(payload, dict):
+ code = payload.get("code") or payload.get("error_code")
+ if code is not None and normalize_contract_error_code(code) in CONTRACT_ERROR_CODES:
+ return True
+ message = payload.get("message") or payload.get("error")
+ return message is not None and payload_is_contract_error(str(message))
+ return False
+
+
_FRIENDLY: dict[type, tuple[str, str]] = {
AuthKeyUnregisteredError: (
"transport_unavailable",
diff --git a/mcp/src/telegram_mcp/fast_read_today.py b/mcp/src/telegram_mcp/fast_read_today.py
index 297eec2..7dc8348 100644
--- a/mcp/src/telegram_mcp/fast_read_today.py
+++ b/mcp/src/telegram_mcp/fast_read_today.py
@@ -5,138 +5,27 @@
import argparse
import asyncio
import json
-import os
-import time
-from dataclasses import asdict, dataclass
-from datetime import date, timedelta
+from datetime import date
from pathlib import Path
import httpx
-from mcp.client.session import ClientSession
-from mcp.client.streamable_http import streamable_http_client
-
-@dataclass(frozen=True)
-class EndpointAttempt:
- endpoint: str
- env_file: str
- port: int
+from .mcp_http_client import (
+ ACCOUNT_ENDPOINTS,
+ EndpointAttempt,
+ McpToolError,
+ call_tool_once,
+ endpoint_attempts,
+ payload_is_tool_error,
+)
class FastReadError(RuntimeError):
pass
-ACCOUNT_ENDPOINTS = {
- "main": (8799, "~/.telegram-mcp/launchd.env"),
- "crwddy": (8799, "~/.telegram-mcp/launchd.env"),
- "pl": (8800, "~/.telegram-mcp-pl/launchd.env"),
- "recklessou": (8801, "~/.telegram-mcp-recklessou/launchd.env"),
- "teamsyncsage": (8802, "~/.telegram-mcp-teamsyncsage/launchd.env"),
- "vermassov": (8803, "~/.telegram-mcp-vermassov/launchd.env"),
-}
-
-
-def load_env_file(path: Path) -> None:
- if not path.exists():
- return
- for raw_line in path.read_text().splitlines():
- line = raw_line.strip()
- if not line or line.startswith("#"):
- continue
- if line.startswith("export "):
- line = line[len("export ") :]
- if "=" not in line:
- continue
- key, value = line.split("=", 1)
- key = key.strip()
- value = value.strip().strip("'\"")
- if key.startswith("TELEGRAM_") and key not in os.environ:
- os.environ[key] = value
-
-
-def build_endpoint(host: str, port: int, path: str) -> str:
- if not path.startswith("/"):
- path = "/" + path
- return f"http://{host}:{port}{path}"
-
-
-def endpoint_attempts(
- *,
- explicit_endpoint: str | None = None,
- host: str | None = None,
- account: str | None = None,
- primary_port: int | None = None,
- failover_ports: list[int] | None = None,
- primary_env_file: str | Path | None = None,
- pl_env_file: str | Path | None = None,
-) -> list[EndpointAttempt]:
- if explicit_endpoint:
- return [EndpointAttempt(endpoint=explicit_endpoint, env_file=str(primary_env_file or ""), port=0)]
-
- host_name = host or os.environ.get("TELEGRAM_MCP_HOST", "127.0.0.1")
- path = os.environ.get("TELEGRAM_MCP_HTTP_PATH", "/mcp")
- selected_account = (account or os.environ.get("TELEGRAM_MCP_ACCOUNT", "main")).strip().lower()
- account_config = ACCOUNT_ENDPOINTS.get(selected_account)
- if account_config is None:
- known = ", ".join(sorted(ACCOUNT_ENDPOINTS))
- raise FastReadError(f"TELEGRAM_MCP_ACCOUNT must be one of: {known}")
-
- account_port, account_env_file = account_config
- primary = primary_port or account_port
- extra = failover_ports
- if extra is None:
- raw = os.environ.get("TELEGRAM_MCP_FAILOVER_PORTS", "")
- extra = [int(item.strip()) for item in raw.split(",") if item.strip()]
-
- ports: list[int] = []
- for candidate in [primary, *extra]:
- if candidate not in ports:
- ports.append(candidate)
-
- selected_env = Path(primary_env_file or account_env_file).expanduser()
-
- attempts: list[EndpointAttempt] = []
- for port in ports:
- attempts.append(
- EndpointAttempt(
- endpoint=build_endpoint(host_name, port, path),
- env_file=str(selected_env),
- port=port,
- )
- )
- return attempts
-
-
-def content_payload(result) -> object | None:
- if not result.content:
- return None
- first = result.content[0]
- text = getattr(first, "text", None)
- if text is None:
- return str(first)
- try:
- return json.loads(text)
- except json.JSONDecodeError:
- return text
-
-
-def payload_is_tool_error(payload: object | None) -> bool:
- if isinstance(payload, str):
- lower = payload.lower()
- return (
- "unknown tool" in lower
- or lower.startswith("error executing tool ")
- or "error executing tool " in lower
- )
- if isinstance(payload, dict):
- message = str(payload.get("message") or payload.get("error") or "")
- return payload_is_tool_error(message)
- return False
-
-
def exception_is_tool_error(exc: Exception) -> bool:
- return payload_is_tool_error(str(exc))
+ return isinstance(exc, McpToolError) or payload_is_tool_error(str(exc))
async def read_once(
@@ -149,48 +38,17 @@ async def read_once(
sender_names: bool,
timeout: float,
) -> dict[str, object]:
- if attempt.env_file:
- load_env_file(Path(attempt.env_file).expanduser())
-
- token = os.environ.get("TELEGRAM_MCP_AUTH_TOKEN", "").strip()
- headers = {"Authorization": f"Bearer {token}"} if token else {}
- http_timeout = httpx.Timeout(
- timeout,
- connect=min(timeout, 3.0),
- read=timeout,
- write=timeout,
- pool=min(timeout, 3.0),
- )
-
- started = time.perf_counter()
- async with httpx.AsyncClient(headers=headers, timeout=http_timeout) as http_client:
- async with streamable_http_client(attempt.endpoint, http_client=http_client) as (
- read_stream,
- write_stream,
- _,
- ):
- async with ClientSession(
- read_stream,
- write_stream,
- read_timeout_seconds=timedelta(seconds=timeout),
- ) as session:
- await session.initialize()
- mode = "full" if (voice or sender_names) else "fast"
- result = await session.call_tool(
- "telegram_read",
- {
- "chat": chat,
- "day": day,
- "limit": limit,
- "mode": mode,
- },
- )
-
- payload = content_payload(result)
- if payload_is_tool_error(payload):
- raise FastReadError(f"MCP tool error at {attempt.endpoint}: {payload!r}")
+ mode = "full" if (voice or sender_names) else "fast"
+ try:
+ payload, elapsed_seconds, completed_attempt = await call_tool_once(
+ attempt=attempt,
+ tool_name="telegram_read",
+ arguments={"chat": chat, "day": day, "limit": limit, "mode": mode},
+ timeout=timeout,
+ )
+ except McpToolError as exc:
+ raise FastReadError(str(exc)) from None
- elapsed_seconds = round(time.perf_counter() - started, 3)
from .agent_preflight import observe_fast_read
from .telemetry import record_telemetry, telemetry_fields_from_result
@@ -206,7 +64,7 @@ async def read_once(
status="ok",
duration_ms=duration_ms,
source="fast_read_cli",
- endpoint_port=attempt.port or None,
+ endpoint_port=completed_attempt.port or None,
arg_chat=chat,
arg_day=day,
arg_limit=limit,
@@ -215,8 +73,8 @@ async def read_once(
return {
"ok": True,
"mode": "telegram_fast_read_today",
- "endpoint": attempt.endpoint,
- "endpoint_port": attempt.port or None,
+ "endpoint": completed_attempt.endpoint,
+ "endpoint_port": completed_attempt.port or None,
"elapsed_seconds": elapsed_seconds,
"payload": payload,
}
@@ -258,7 +116,6 @@ async def read_with_failover(
httpx.NetworkError,
ConnectionError,
OSError,
- FastReadError,
) as exc:
errors.append(f"{attempt.endpoint}: {type(exc).__name__}: {exc}")
diff --git a/mcp/src/telegram_mcp/mcp_http_client.py b/mcp/src/telegram_mcp/mcp_http_client.py
index 93e4aab..b54c68f 100644
--- a/mcp/src/telegram_mcp/mcp_http_client.py
+++ b/mcp/src/telegram_mcp/mcp_http_client.py
@@ -13,6 +13,8 @@
from mcp.client.session import ClientSession
from mcp.client.streamable_http import streamable_http_client
+from .errors import payload_is_contract_error
+
@dataclass(frozen=True)
class EndpointAttempt:
@@ -25,6 +27,22 @@ class McpCliError(RuntimeError):
pass
+class McpToolError(McpCliError):
+ def __init__(
+ self,
+ *,
+ endpoint: str,
+ port: int | None,
+ payload: object | None,
+ elapsed_seconds: float,
+ ) -> None:
+ self.endpoint = endpoint
+ self.port = port
+ self.payload = payload
+ self.elapsed_seconds = elapsed_seconds
+ super().__init__(f"MCP tool error at {endpoint}: {payload!r}")
+
+
ACCOUNT_ENDPOINTS = {
"main": (8799, "~/.telegram-mcp/launchd.env"),
"crwddy": (8799, "~/.telegram-mcp/launchd.env"),
@@ -121,13 +139,32 @@ def content_payload(result) -> object | None:
def payload_is_tool_error(payload: object | None) -> bool:
if isinstance(payload, str):
- return "Unknown tool" in payload or "unknown tool" in payload.lower()
+ lower = payload.strip().lower()
+ return (
+ "unknown tool" in lower
+ or lower.startswith("error executing tool ")
+ or "error executing tool " in lower
+ or payload_is_contract_error(lower)
+ )
if isinstance(payload, dict):
- message = str(payload.get("message") or payload.get("error") or "")
- return "Unknown tool" in message
+ message = payload.get("message") or payload.get("error")
+ if message is not None and payload_is_tool_error(str(message)):
+ return True
+ return payload_is_contract_error(payload)
return False
+def result_is_tool_error(result, payload: object | None) -> bool:
+ return bool(getattr(result, "isError", False)) or payload_is_tool_error(payload)
+
+
+def tool_error_payload(result, payload: object | None) -> object | None:
+ structured = getattr(result, "structuredContent", None)
+ if structured is not None:
+ return structured
+ return payload
+
+
async def call_tool_once(
*,
attempt: EndpointAttempt,
@@ -164,10 +201,16 @@ async def call_tool_once(
result = await session.call_tool(tool_name, arguments)
payload = content_payload(result)
- if payload_is_tool_error(payload):
- raise McpCliError(f"MCP tool error at {attempt.endpoint}: {payload!r}")
-
elapsed_seconds = round(time.perf_counter() - started, 3)
+ if result_is_tool_error(result, payload):
+ error_payload = tool_error_payload(result, payload)
+ raise McpToolError(
+ endpoint=attempt.endpoint,
+ port=attempt.port or None,
+ payload=error_payload,
+ elapsed_seconds=elapsed_seconds,
+ )
+
return payload, elapsed_seconds, attempt
@@ -239,7 +282,6 @@ async def call_tool_with_failover(
httpx.NetworkError,
ConnectionError,
OSError,
- McpCliError,
) as exc:
errors.append(f"{attempt.endpoint}: {type(exc).__name__}: {exc}")
@@ -269,7 +311,6 @@ async def list_tools_with_failover(
httpx.NetworkError,
ConnectionError,
OSError,
- McpCliError,
) as exc:
errors.append(f"{attempt.endpoint}: {type(exc).__name__}: {exc}")
diff --git a/mcp/src/telegram_mcp/mcp_prewarm.py b/mcp/src/telegram_mcp/mcp_prewarm.py
index 3c15027..805f163 100644
--- a/mcp/src/telegram_mcp/mcp_prewarm.py
+++ b/mcp/src/telegram_mcp/mcp_prewarm.py
@@ -13,7 +13,7 @@
from mcp.client.session import ClientSession
from mcp.client.streamable_http import streamable_http_client
-from .fast_read_today import build_endpoint, endpoint_attempts, load_env_file
+from .mcp_http_client import build_endpoint, endpoint_attempts, load_env_file
@dataclass(frozen=True)
diff --git a/mcp/src/telegram_mcp/member_export_paths.py b/mcp/src/telegram_mcp/member_export_paths.py
index 533d421..7cb2559 100644
--- a/mcp/src/telegram_mcp/member_export_paths.py
+++ b/mcp/src/telegram_mcp/member_export_paths.py
@@ -6,6 +6,8 @@
def default_member_export_dir() -> Path:
path = Path.home() / ".cache" / "telegram-mcp" / "member-exports"
+ if path.is_symlink():
+ raise ValueError("member export root must not be a symlink")
path.mkdir(parents=True, exist_ok=True)
path.chmod(0o700)
return path
@@ -14,6 +16,8 @@ def default_member_export_dir() -> Path:
def resolve_member_export_dir(output_dir: str | None) -> Path:
root = default_member_export_dir().resolve(strict=False)
candidate = root if output_dir is None else Path(output_dir).expanduser()
+ if candidate.exists() and candidate.is_symlink():
+ raise ValueError("member export output_dir must not be a symlink")
resolved = candidate.resolve(strict=False)
try:
resolved.relative_to(root)
diff --git a/mcp/src/telegram_mcp/runtime.py b/mcp/src/telegram_mcp/runtime.py
index c581825..3453998 100644
--- a/mcp/src/telegram_mcp/runtime.py
+++ b/mcp/src/telegram_mcp/runtime.py
@@ -31,6 +31,7 @@
log = structlog.get_logger()
_shared_wrapper: TelegramWrapper | None = None
_shared_wrapper_lock = asyncio.Lock()
+OAUTH_AUTHORIZATION_SERVER_METADATA_PATH = "/.well-known/oauth-authorization-server"
class StaticBearerTokenVerifier:
@@ -248,8 +249,13 @@ async def _oauth_as_metadata(request: _Request) -> _JSONResponse:
}
)
+ mcp._custom_starlette_routes = [
+ route
+ for route in mcp._custom_starlette_routes
+ if getattr(route, "path", None) != OAUTH_AUTHORIZATION_SERVER_METADATA_PATH
+ ]
mcp._custom_starlette_routes.append(
- _Route("/.well-known/oauth-authorization-server", _oauth_as_metadata, methods=["GET"])
+ _Route(OAUTH_AUTHORIZATION_SERVER_METADATA_PATH, _oauth_as_metadata, methods=["GET"])
)
diff --git a/mcp/src/telegram_mcp/send_confirmation.py b/mcp/src/telegram_mcp/send_confirmation.py
index c6313da..f6827da 100644
--- a/mcp/src/telegram_mcp/send_confirmation.py
+++ b/mcp/src/telegram_mcp/send_confirmation.py
@@ -3,6 +3,7 @@
from __future__ import annotations
import secrets
+import threading
import time
from dataclasses import dataclass
from datetime import datetime, timezone
@@ -44,6 +45,7 @@ def __init__(self, *, ttl_seconds: int = 600) -> None:
self._ttl_seconds = ttl_seconds
self._records: dict[str, SendConfirmationRecord] = {}
self._token_index: dict[str, str] = {}
+ self._lock = threading.RLock()
def _now(self) -> float:
return time.time()
@@ -68,53 +70,65 @@ def mint(
preview_text: str,
risk_class: str = "standard",
) -> tuple[str, str, datetime]:
- preview_id = secrets.token_urlsafe(16)
- token = secrets.token_urlsafe(24)
- nonce = secrets.token_urlsafe(12)
- expires_at = self._now() + self._ttl_seconds
- self._records[preview_id] = SendConfirmationRecord(
- preview_id=preview_id,
- expires_at=expires_at,
- payload=dict(payload),
- preview_text=preview_text,
- approval_state="pending",
- risk_class=risk_class,
- confirmation_token=token,
- one_time_nonce=nonce,
- )
- self._token_index[token] = preview_id
- return preview_id, token, datetime.fromtimestamp(expires_at, tz=timezone.utc)
+ with self._lock:
+ preview_id = secrets.token_urlsafe(16)
+ token = secrets.token_urlsafe(24)
+ nonce = secrets.token_urlsafe(12)
+ expires_at = self._now() + self._ttl_seconds
+ self._records[preview_id] = SendConfirmationRecord(
+ preview_id=preview_id,
+ expires_at=expires_at,
+ payload=dict(payload),
+ preview_text=preview_text,
+ approval_state="pending",
+ risk_class=risk_class,
+ confirmation_token=token,
+ one_time_nonce=nonce,
+ )
+ self._token_index[token] = preview_id
+ return preview_id, token, datetime.fromtimestamp(expires_at, tz=timezone.utc)
def get(self, key: str) -> SendConfirmationRecord | None:
- preview_id = self._resolve_key(key)
- if preview_id is None:
- return None
- record = self._records.get(preview_id)
- if record is None:
- return None
- return self._expire_if_needed(preview_id, record)
+ with self._lock:
+ preview_id = self._resolve_key(key)
+ if preview_id is None:
+ return None
+ record = self._records.get(preview_id)
+ if record is None:
+ return None
+ return self._expire_if_needed(preview_id, record)
def approve(self, token: str) -> SendConfirmationRecord:
- record = self.get(token)
- if record is None:
- raise ToolContractError("invalid_confirmation_token", "confirmation token is unknown")
- if record.approval_state == "expired":
- raise ToolContractError("expired_confirmation_token", "confirmation token has expired")
- if record.approval_state == "used":
- raise ToolContractError("invalid_confirmation_token", "confirmation token was already used")
- if record.approval_state == "rejected":
- raise ToolContractError("confirmation_rejected", "confirmation was rejected by the operator")
- record.approval_state = "approved"
- self._records[token] = record
- return record
+ with self._lock:
+ preview_id = self._resolve_key(token)
+ if preview_id is None:
+ raise ToolContractError("invalid_confirmation_token", "confirmation token is unknown")
+ record = self._records.get(preview_id)
+ if record is None:
+ raise ToolContractError("invalid_confirmation_token", "confirmation token is unknown")
+ record = self._expire_if_needed(preview_id, record)
+ if record.approval_state == "expired":
+ raise ToolContractError("expired_confirmation_token", "confirmation token has expired")
+ if record.approval_state == "used":
+ raise ToolContractError("invalid_confirmation_token", "confirmation token was already used")
+ if record.approval_state == "rejected":
+ raise ToolContractError("confirmation_rejected", "confirmation was rejected by the operator")
+ record.approval_state = "approved"
+ return record
def reject(self, token: str) -> SendConfirmationRecord:
- record = self.get(token)
- if record is None:
- raise ToolContractError("invalid_confirmation_token", "confirmation token is unknown")
- record.approval_state = "rejected"
- self._records[token] = record
- return record
+ with self._lock:
+ preview_id = self._resolve_key(token)
+ if preview_id is None:
+ raise ToolContractError("invalid_confirmation_token", "confirmation token is unknown")
+ record = self._records.get(preview_id)
+ if record is None:
+ raise ToolContractError("invalid_confirmation_token", "confirmation token is unknown")
+ record = self._expire_if_needed(preview_id, record)
+ if record.approval_state == "expired":
+ raise ToolContractError("expired_confirmation_token", "confirmation token has expired")
+ record.approval_state = "rejected"
+ return record
def consume(
self,
@@ -124,35 +138,37 @@ def consume(
approval_required: bool,
preview_id_only: bool = False,
) -> SendConfirmationRecord:
- if not key:
- raise ToolContractError(
- "missing_confirmation_token",
- "send/reply requires preview_id or confirmation_token from prepare_*",
- )
- resolved = self._resolve_key(key)
- if resolved is None:
- raise ToolContractError("invalid_confirmation_token", "confirmation token is unknown or already used")
- record = self._records.get(resolved)
- if record is None:
- raise ToolContractError("invalid_confirmation_token", "confirmation token is unknown or already used")
- record = self._expire_if_needed(resolved, record)
- if record.approval_state == "expired":
- raise ToolContractError("expired_confirmation_token", "confirmation token has expired")
- if record.approval_state == "used":
- raise ToolContractError("invalid_confirmation_token", "confirmation token was already used")
- if record.approval_state == "rejected":
- raise ToolContractError("confirmation_rejected", "confirmation was rejected by the operator")
- if approval_required and record.approval_state != "approved":
- raise ToolContractError(
- "human_approval_required",
- "open the approval URL and click Approve before sending",
- )
- if not preview_id_only and expected is not None and record.payload != expected:
- raise ToolContractError(
- "confirmation_payload_mismatch",
- "send/reply arguments do not match the preview confirmation",
- )
- record.approval_state = "used"
- self._records.pop(resolved, None)
- self._token_index.pop(record.confirmation_token, None)
- return record
\ No newline at end of file
+ with self._lock:
+ if not key:
+ raise ToolContractError(
+ "missing_confirmation_token",
+ "send/reply requires preview_id or confirmation_token from prepare_*",
+ )
+ resolved = self._resolve_key(key)
+ if resolved is None:
+ raise ToolContractError("invalid_confirmation_token", "confirmation token is unknown or already used")
+ record = self._records.get(resolved)
+ if record is None:
+ raise ToolContractError("invalid_confirmation_token", "confirmation token is unknown or already used")
+ record = self._expire_if_needed(resolved, record)
+ if record.approval_state == "expired":
+ raise ToolContractError("expired_confirmation_token", "confirmation token has expired")
+ if record.approval_state == "used":
+ raise ToolContractError("invalid_confirmation_token", "confirmation token was already used")
+ if record.approval_state == "rejected":
+ raise ToolContractError("confirmation_rejected", "confirmation was rejected by the operator")
+ if approval_required and record.approval_state != "approved":
+ raise ToolContractError(
+ "human_approval_required",
+ "open the approval URL and click Approve before sending",
+ )
+ if not preview_id_only and expected is not None and record.payload != expected:
+ raise ToolContractError(
+ "confirmation_payload_mismatch",
+ "send/reply arguments do not match the preview confirmation",
+ )
+ record.approval_state = "used"
+ self._records.pop(resolved, None)
+ self._records.pop(record.confirmation_token, None)
+ self._token_index.pop(record.confirmation_token, None)
+ return record
diff --git a/mcp/src/telegram_mcp/tg_cli.py b/mcp/src/telegram_mcp/tg_cli.py
index f8701da..db54675 100644
--- a/mcp/src/telegram_mcp/tg_cli.py
+++ b/mcp/src/telegram_mcp/tg_cli.py
@@ -16,17 +16,17 @@
METADATA_LIST_SPECS,
MetadataCountSpec,
)
-from .mcp_http_client import ACCOUNT_ENDPOINTS, McpCliError, call_tool_with_failover
+from .mcp_http_client import ACCOUNT_ENDPOINTS, McpCliError, McpToolError, call_tool_with_failover
from .telemetry import record_telemetry, telemetry_fields_from_result
-def _payload_is_tool_error(payload: object | None) -> bool:
- if isinstance(payload, str):
- lower = payload.lower()
- return "unknown tool" in lower or "error executing tool " in lower
- if isinstance(payload, dict):
- return _payload_is_tool_error(payload.get("error") or payload.get("message"))
- return False
+def _record_tool_telemetry(event_name: str, *, status: str, payload: object | None, **fields: object) -> None:
+ record_telemetry(
+ event_name,
+ status=status,
+ **fields,
+ **telemetry_fields_from_result(payload if isinstance(payload, dict) else None),
+ )
def _wrap_ok(
@@ -41,23 +41,6 @@ def _wrap_ok(
data_source = None
if isinstance(payload, dict):
data_source = payload.get("data_source")
- if _payload_is_tool_error(payload):
- wrapped = {
- "ok": False,
- "command": command,
- "intent": intent,
- "data_source": data_source or "live_telegram",
- "endpoint": endpoint,
- "endpoint_port": endpoint_port,
- "elapsed_seconds": elapsed_seconds,
- "error": "telegram_tool_error",
- "message": "Live Telegram tool returned an error payload.",
- }
- if isinstance(payload, dict):
- wrapped["tool_error_payload"] = payload
- elif isinstance(payload, str):
- wrapped["tool_error_payload"] = payload[:2000]
- return wrapped
return {
"ok": True,
"command": command,
@@ -70,6 +53,36 @@ def _wrap_ok(
}
+def _wrap_tool_error(
+ *,
+ command: str,
+ endpoint: str,
+ endpoint_port: int | None,
+ elapsed_seconds: float,
+ payload: object | None,
+ intent: str,
+) -> dict[str, object]:
+ data_source = payload.get("data_source") if isinstance(payload, dict) else None
+ wrapped = {
+ "ok": False,
+ "command": command,
+ "intent": intent,
+ "data_source": data_source or "live_telegram",
+ "endpoint": endpoint,
+ "endpoint_port": endpoint_port,
+ "elapsed_seconds": elapsed_seconds,
+ "error": "telegram_tool_error",
+ "message": "Live Telegram tool returned an error.",
+ }
+ if isinstance(payload, dict):
+ wrapped["tool_error_payload"] = payload
+ elif isinstance(payload, str):
+ wrapped["tool_error_payload"] = payload[:2000]
+ elif payload is not None:
+ wrapped["tool_error_payload"] = repr(payload)[:2000]
+ return wrapped
+
+
def _wrap_err(*, command: str, exc: Exception) -> dict[str, object]:
return {
"ok": False,
@@ -79,6 +92,66 @@ def _wrap_err(*, command: str, exc: Exception) -> dict[str, object]:
}
+async def _run_tool_command(
+ *,
+ command: str,
+ intent: str,
+ event_name: str,
+ tool_name: str,
+ arguments: dict[str, object],
+ timeout: float,
+ endpoint: str | None,
+ env_file: str | None,
+ account: str,
+ telemetry_fields: dict[str, object],
+) -> dict[str, object]:
+ try:
+ payload, elapsed, attempt = await call_tool_with_failover(
+ tool_name=tool_name,
+ arguments=arguments,
+ timeout=timeout,
+ explicit_endpoint=endpoint,
+ env_file=env_file,
+ account=account,
+ )
+ except McpToolError as exc:
+ _record_tool_telemetry(
+ event_name,
+ status="error",
+ payload=exc.payload,
+ duration_ms=round(exc.elapsed_seconds * 1000, 3),
+ source="tg_cli",
+ endpoint_port=exc.port,
+ **telemetry_fields,
+ )
+ return _wrap_tool_error(
+ command=command,
+ endpoint=exc.endpoint,
+ endpoint_port=exc.port,
+ elapsed_seconds=exc.elapsed_seconds,
+ payload=exc.payload,
+ intent=intent,
+ )
+
+ _record_tool_telemetry(
+ event_name,
+ status="ok",
+ payload=payload,
+ duration_ms=round(elapsed * 1000, 3),
+ source="tg_cli",
+ endpoint_port=attempt.port or None,
+ **telemetry_fields,
+ )
+ return _wrap_ok(
+ command=command,
+ endpoint=attempt.endpoint,
+ endpoint_port=attempt.port or None,
+ elapsed_seconds=elapsed,
+ payload=payload,
+ intent=intent,
+ )
+
+
CHAT_RE = re.compile(r"(?P@[A-Za-z0-9_]{5,}|tg://dialog/[^\s]+|-100\d{6,}|\bme\b)", re.IGNORECASE)
COUNT_POSTS_RE = re.compile(
r"(сколько|count|how many|total).{0,80}(пост|posts?|messages?|сообщен)",
@@ -272,37 +345,24 @@ async def cmd_read_today(
env_file: str | None,
account: str,
) -> dict[str, object]:
- payload, elapsed, attempt = await call_tool_with_failover(
+ output = await _run_tool_command(
+ command="read today",
+ intent="live_today",
+ event_name="tg_read_today",
tool_name="telegram_read",
arguments={"chat": chat, "day": day, "limit": limit, "mode": "fast"},
timeout=timeout,
- explicit_endpoint=endpoint,
+ endpoint=endpoint,
env_file=env_file,
account=account,
+ telemetry_fields={"arg_chat": chat, "arg_day": day, "arg_limit": limit},
)
- duration_ms = round(elapsed * 1000, 3)
from .agent_preflight import observe_fast_read
- observe_fast_read(tool="tg_read_today", status="ok", source="tg_cli", duration_ms=duration_ms)
- record_telemetry(
- "tg_read_today",
- status="ok",
- duration_ms=duration_ms,
- source="tg_cli",
- endpoint_port=attempt.port or None,
- arg_chat=chat,
- arg_day=day,
- arg_limit=limit,
- **telemetry_fields_from_result(payload if isinstance(payload, dict) else None),
- )
- return _wrap_ok(
- command="read today",
- endpoint=attempt.endpoint,
- endpoint_port=attempt.port or None,
- elapsed_seconds=elapsed,
- payload=payload,
- intent="live_today",
- )
+ duration_ms = round(float(output.get("elapsed_seconds") or 0.0) * 1000, 3)
+ status = "ok" if output.get("ok") else "error"
+ observe_fast_read(tool="tg_read_today", status=status, source="tg_cli", duration_ms=duration_ms)
+ return output
async def cmd_read_recent(
@@ -314,31 +374,17 @@ async def cmd_read_recent(
env_file: str | None,
account: str,
) -> dict[str, object]:
- payload, elapsed, attempt = await call_tool_with_failover(
+ return await _run_tool_command(
+ command="read recent",
+ intent="live_recent",
+ event_name="tg_read_recent",
tool_name="telegram_read",
arguments={"chat": chat, "limit": limit, "mode": "fast"},
timeout=timeout,
- explicit_endpoint=endpoint,
+ endpoint=endpoint,
env_file=env_file,
account=account,
- )
- record_telemetry(
- "tg_read_recent",
- status="ok",
- duration_ms=round(elapsed * 1000, 3),
- source="tg_cli",
- endpoint_port=attempt.port or None,
- arg_chat=chat,
- arg_limit=limit,
- **telemetry_fields_from_result(payload if isinstance(payload, dict) else None),
- )
- return _wrap_ok(
- command="read recent",
- endpoint=attempt.endpoint,
- endpoint_port=attempt.port or None,
- elapsed_seconds=elapsed,
- payload=payload,
- intent="live_recent",
+ telemetry_fields={"arg_chat": chat, "arg_limit": limit},
)
@@ -352,30 +398,17 @@ async def cmd_search(
env_file: str | None,
account: str,
) -> dict[str, object]:
- payload, elapsed, attempt = await call_tool_with_failover(
+ return await _run_tool_command(
+ command="search",
+ intent="live_search",
+ event_name="tg_search",
tool_name="telegram_search",
arguments={"chat": chat, "query": query, "limit": limit},
timeout=timeout,
- explicit_endpoint=endpoint,
+ endpoint=endpoint,
env_file=env_file,
account=account,
- )
- record_telemetry(
- "tg_search",
- status="ok",
- duration_ms=round(elapsed * 1000, 3),
- source="tg_cli",
- endpoint_port=attempt.port or None,
- arg_chat=chat,
- **telemetry_fields_from_result(payload if isinstance(payload, dict) else None),
- )
- return _wrap_ok(
- command="search",
- endpoint=attempt.endpoint,
- endpoint_port=attempt.port or None,
- elapsed_seconds=elapsed,
- payload=payload,
- intent="live_search",
+ telemetry_fields={"arg_chat": chat},
)
@@ -388,30 +421,17 @@ async def cmd_count_metadata(
env_file: str | None,
account: str,
) -> dict[str, object]:
- payload, elapsed, attempt = await call_tool_with_failover(
+ return await _run_tool_command(
+ command=f"count {spec.cli_name}",
+ intent=f"count_channel_{spec.key}",
+ event_name=f"tg_count_{spec.key}",
tool_name=spec.tool_name,
arguments={"chat": chat},
timeout=timeout,
- explicit_endpoint=endpoint,
+ endpoint=endpoint,
env_file=env_file,
account=account,
- )
- record_telemetry(
- f"tg_count_{spec.key}",
- status="ok",
- duration_ms=round(elapsed * 1000, 3),
- source="tg_cli",
- endpoint_port=attempt.port or None,
- arg_chat=chat,
- **telemetry_fields_from_result(payload if isinstance(payload, dict) else None),
- )
- return _wrap_ok(
- command=f"count {spec.cli_name}",
- endpoint=attempt.endpoint,
- endpoint_port=attempt.port or None,
- elapsed_seconds=elapsed,
- payload=payload,
- intent=f"count_channel_{spec.key}",
+ telemetry_fields={"arg_chat": chat},
)
@@ -446,32 +466,17 @@ async def cmd_list_metadata(
) -> dict[str, object]:
if not spec.list_tool_name:
raise McpCliError(f"{spec.cli_name} does not support bounded list")
- payload, elapsed, attempt = await call_tool_with_failover(
+ return await _run_tool_command(
+ command=f"list {spec.list_cli_name}",
+ intent=f"list_channel_{spec.key}",
+ event_name=f"tg_list_{spec.key}",
tool_name=spec.list_tool_name,
arguments={"chat": chat, "limit": limit, "offset_id": offset_id},
timeout=timeout,
- explicit_endpoint=endpoint,
+ endpoint=endpoint,
env_file=env_file,
account=account,
- )
- record_telemetry(
- f"tg_list_{spec.key}",
- status="ok",
- duration_ms=round(elapsed * 1000, 3),
- source="tg_cli",
- endpoint_port=attempt.port or None,
- arg_chat=chat,
- arg_limit=limit,
- arg_offset_id=offset_id,
- **telemetry_fields_from_result(payload if isinstance(payload, dict) else None),
- )
- return _wrap_ok(
- command=f"list {spec.list_cli_name}",
- endpoint=attempt.endpoint,
- endpoint_port=attempt.port or None,
- elapsed_seconds=elapsed,
- payload=payload,
- intent=f"list_channel_{spec.key}",
+ telemetry_fields={"arg_chat": chat, "arg_limit": limit, "arg_offset_id": offset_id},
)
@@ -483,30 +488,17 @@ async def cmd_latest(
env_file: str | None,
account: str,
) -> dict[str, object]:
- payload, elapsed, attempt = await call_tool_with_failover(
+ return await _run_tool_command(
+ command="latest",
+ intent="latest_message_metadata",
+ event_name="tg_latest_message",
tool_name="telegram_latest_message",
arguments={"chat": chat},
timeout=timeout,
- explicit_endpoint=endpoint,
+ endpoint=endpoint,
env_file=env_file,
account=account,
- )
- record_telemetry(
- "tg_latest_message",
- status="ok",
- duration_ms=round(elapsed * 1000, 3),
- source="tg_cli",
- endpoint_port=attempt.port or None,
- arg_chat=chat,
- **telemetry_fields_from_result(payload if isinstance(payload, dict) else None),
- )
- return _wrap_ok(
- command="latest",
- endpoint=attempt.endpoint,
- endpoint_port=attempt.port or None,
- elapsed_seconds=elapsed,
- payload=payload,
- intent="latest_message_metadata",
+ telemetry_fields={"arg_chat": chat},
)
@@ -518,30 +510,17 @@ async def cmd_info(
env_file: str | None,
account: str,
) -> dict[str, object]:
- payload, elapsed, attempt = await call_tool_with_failover(
+ return await _run_tool_command(
+ command="info",
+ intent="dialog_metadata",
+ event_name="tg_dialog_metadata",
tool_name="telegram_dialog_metadata",
arguments={"chat": chat},
timeout=timeout,
- explicit_endpoint=endpoint,
+ endpoint=endpoint,
env_file=env_file,
account=account,
- )
- record_telemetry(
- "tg_dialog_metadata",
- status="ok",
- duration_ms=round(elapsed * 1000, 3),
- source="tg_cli",
- endpoint_port=attempt.port or None,
- arg_chat=chat,
- **telemetry_fields_from_result(payload if isinstance(payload, dict) else None),
- )
- return _wrap_ok(
- command="info",
- endpoint=attempt.endpoint,
- endpoint_port=attempt.port or None,
- elapsed_seconds=elapsed,
- payload=payload,
- intent="dialog_metadata",
+ telemetry_fields={"arg_chat": chat},
)
@@ -554,31 +533,17 @@ async def cmd_message(
env_file: str | None,
account: str,
) -> dict[str, object]:
- payload, elapsed, attempt = await call_tool_with_failover(
+ return await _run_tool_command(
+ command="message",
+ intent="get_message_by_id",
+ event_name="tg_get_message",
tool_name="telegram_get_message",
arguments={"chat": chat, "message_id": message_id},
timeout=timeout,
- explicit_endpoint=endpoint,
+ endpoint=endpoint,
env_file=env_file,
account=account,
- )
- record_telemetry(
- "tg_get_message",
- status="ok",
- duration_ms=round(elapsed * 1000, 3),
- source="tg_cli",
- endpoint_port=attempt.port or None,
- arg_chat=chat,
- arg_message_id=message_id,
- **telemetry_fields_from_result(payload if isinstance(payload, dict) else None),
- )
- return _wrap_ok(
- command="message",
- endpoint=attempt.endpoint,
- endpoint_port=attempt.port or None,
- elapsed_seconds=elapsed,
- payload=payload,
- intent="get_message_by_id",
+ telemetry_fields={"arg_chat": chat, "arg_message_id": message_id},
)
diff --git a/mcp/src/telegram_mcp/tools/__init__.py b/mcp/src/telegram_mcp/tools/__init__.py
index adf68ea..a913ce4 100644
--- a/mcp/src/telegram_mcp/tools/__init__.py
+++ b/mcp/src/telegram_mcp/tools/__init__.py
@@ -192,7 +192,6 @@
"telegram_prepare_reply",
"telegram_read",
"telegram_search",
- "send_file",
}
FACADE_TOOL_PROFILES = {"facade", "safe", "restricted"}
diff --git a/mcp/src/telegram_mcp/tools/dialog_facade_tools.py b/mcp/src/telegram_mcp/tools/dialog_facade_tools.py
index 672a7e8..be9a10d 100644
--- a/mcp/src/telegram_mcp/tools/dialog_facade_tools.py
+++ b/mcp/src/telegram_mcp/tools/dialog_facade_tools.py
@@ -34,6 +34,17 @@
READONLY = ToolAnnotations(readOnlyHint=True, destructiveHint=False, idempotentHint=True, openWorldHint=True)
ADDITIVE = ToolAnnotations(readOnlyHint=False, destructiveHint=False, openWorldHint=True)
CONFIRMED_WRITE = ToolAnnotations(readOnlyHint=False, destructiveHint=False, openWorldHint=True)
+MEMBER_EXPORT_FILTERS = {"all", "admins", "banned"}
+
+
+def _normalize_member_export_filter(filter_value: str) -> str:
+ normalized = filter_value.strip().lower()
+ if normalized not in MEMBER_EXPORT_FILTERS:
+ raise ToolContractError(
+ "invalid_input",
+ "member export filter must be one of: all, admins, banned",
+ )
+ return normalized
async def resolve_dialog(query: str | int):
@@ -584,13 +595,14 @@ async def telegram_export_members(
):
"""Export channel/group members to a private local JSON artifact."""
limit = clamp_member_export_limit(limit)
+ filter = _normalize_member_export_filter(filter)
tg = await runtime.get_tg()
handle = await tg.resolve_dialog(chat)
if filter == "admins":
- participants = await tg.get_admins(chat=handle.dialog_ref)
+ participants = await tg.get_admins(chat=handle.dialog_ref, limit=limit)
total = len(participants)
elif filter == "banned":
- participants = await tg.get_banned_users(chat=handle.dialog_ref)
+ participants = await tg.get_banned_users(chat=handle.dialog_ref, limit=limit)
total = len(participants)
else:
participants, total = await tg.get_participants(chat=handle.dialog_ref, limit=limit)
diff --git a/mcp/src/telegram_mcp/tools/group_tools.py b/mcp/src/telegram_mcp/tools/group_tools.py
index c908139..4f71734 100644
--- a/mcp/src/telegram_mcp/tools/group_tools.py
+++ b/mcp/src/telegram_mcp/tools/group_tools.py
@@ -5,12 +5,23 @@
from mcp.types import ToolAnnotations
from .. import runtime
-from ..errors import tool_error_handler
+from ..errors import ToolContractError, tool_error_handler
from ..types import ChatInfo, InviteLinkInfo, OperationResult, ParticipantsResult
READONLY = ToolAnnotations(readOnlyHint=True, destructiveHint=False, idempotentHint=True, openWorldHint=True)
ADDITIVE = ToolAnnotations(readOnlyHint=False, destructiveHint=False, openWorldHint=True)
DESTRUCTIVE = ToolAnnotations(readOnlyHint=False, destructiveHint=True, openWorldHint=True)
+PARTICIPANT_FILTERS = {"all", "admins", "banned"}
+
+
+def _normalize_participant_filter(filter_value: str) -> str:
+ normalized = filter_value.strip().lower()
+ if normalized not in PARTICIPANT_FILTERS:
+ raise ToolContractError(
+ "invalid_input",
+ "participant filter must be one of: all, admins, banned",
+ )
+ return normalized
async def create_group(title: str, user_ids: list[int]) -> ChatInfo:
@@ -53,11 +64,12 @@ async def get_participants(
) -> ParticipantsResult:
"""Get members of a group or channel (filter: 'all', 'admins', 'banned')."""
tg = await runtime.get_tg()
+ filter = _normalize_participant_filter(filter)
if filter == "admins":
- participants = await tg.get_admins(chat=chat)
+ participants = await tg.get_admins(chat=chat, limit=limit)
return ParticipantsResult(participants=participants, total=len(participants))
elif filter == "banned":
- participants = await tg.get_banned_users(chat=chat)
+ participants = await tg.get_banned_users(chat=chat, limit=limit)
return ParticipantsResult(participants=participants, total=len(participants))
else:
participants, total = await tg.get_participants(chat=chat, limit=limit)
diff --git a/mcp/src/telegram_mcp/tools/media_tools.py b/mcp/src/telegram_mcp/tools/media_tools.py
index 7b0d0c4..3aceea5 100644
--- a/mcp/src/telegram_mcp/tools/media_tools.py
+++ b/mcp/src/telegram_mcp/tools/media_tools.py
@@ -116,5 +116,3 @@ def register_facade(mcp) -> None:
mcp.tool(annotations=READONLY)(tool_error_handler(download_dialog_media))
mcp.tool(annotations=READONLY)(tool_error_handler(prepare_media_inspection_manifest))
mcp.tool(annotations=READONLY)(tool_error_handler(telegram_inspect_media))
- # notes-runner assemble delivery uses digest-runner send-file -> MCP send_file.
- mcp.tool(annotations=ADDITIVE)(tool_error_handler(send_file))
diff --git a/mcp/tests/test_dialog_facade_tools.py b/mcp/tests/test_dialog_facade_tools.py
index c9fdb2f..0e9646a 100644
--- a/mcp/tests/test_dialog_facade_tools.py
+++ b/mcp/tests/test_dialog_facade_tools.py
@@ -5,6 +5,7 @@
from unittest.mock import AsyncMock, patch
from telegram_mcp import server
+from telegram_mcp.errors import ToolContractError
from telegram_mcp.tools.dialog_facade_tools import telegram_export_members
from telegram_mcp.types import (
DialogContextResult,
@@ -520,6 +521,33 @@ def test_telegram_export_members_runs_without_extra_pii_acknowledgement(self) ->
self.assertEqual(result.participants[0].username, "ada")
wrapper.get_participants.assert_awaited_once_with(chat="tg://dialog/channel/10", limit=200)
+ def test_telegram_export_members_applies_limit_to_admin_filter(self) -> None:
+ wrapper = AsyncMock()
+ wrapper.resolve_dialog.return_value = DialogHandle(
+ dialog_ref="tg://dialog/channel/10",
+ id=10,
+ name="Target",
+ type="channel",
+ username="targetdaddy",
+ resolved_from="@targetdaddy",
+ )
+ wrapper.get_admins.return_value = [Participant(id=1, first_name="Ada", username="ada")]
+
+ with TemporaryDirectory() as tmp:
+ export_dir = Path(tmp)
+ with patch("telegram_mcp.runtime.get_tg", AsyncMock(return_value=wrapper)), patch(
+ "telegram_mcp.tools.dialog_facade_tools.resolve_member_export_dir",
+ return_value=export_dir,
+ ):
+ result = _run(telegram_export_members(chat="@targetdaddy", filter="admins", limit=1))
+
+ self.assertEqual(result.total, 1)
+ wrapper.get_admins.assert_awaited_once_with(chat="tg://dialog/channel/10", limit=1)
+
+ def test_telegram_export_members_rejects_unknown_filter(self) -> None:
+ with self.assertRaisesRegex(ToolContractError, "member export filter"):
+ _run(telegram_export_members(chat="@targetdaddy", filter="owners"))
+
def test_search_dialog_messages_returns_dialog_read_result(self):
wrapper = AsyncMock()
wrapper.search_dialog_messages.return_value = DialogReadResult(
diff --git a/mcp/tests/test_fast_read_today.py b/mcp/tests/test_fast_read_today.py
index c81f803..f22c405 100644
--- a/mcp/tests/test_fast_read_today.py
+++ b/mcp/tests/test_fast_read_today.py
@@ -58,6 +58,8 @@ def test_payload_is_tool_error_detects_unknown_tool(self):
self.assertTrue(payload_is_tool_error("Unknown tool: telegram_read"))
self.assertTrue(payload_is_tool_error("Error executing tool telegram_read: raw failure"))
self.assertTrue(payload_is_tool_error({"error": "Error executing tool telegram_read: raw failure"}))
+ self.assertTrue(payload_is_tool_error("permission_denied: private channel | next: ask user"))
+ self.assertTrue(payload_is_tool_error({"error": "rate_limited: retry later"}))
self.assertFalse(payload_is_tool_error({"data_source": "live_telegram"}))
def test_exception_is_tool_error_detects_nested_tool_failure(self):
@@ -98,3 +100,32 @@ async def fake_read_once(**kwargs):
)
self.assertIn("8799", str(ctx.exception))
+
+ def test_read_with_failover_does_not_retry_tool_error(self):
+ attempts = [
+ EndpointAttempt("http://127.0.0.1:8799/mcp", "/tmp/a.env", 8799),
+ EndpointAttempt("http://127.0.0.1:8798/mcp", "/tmp/a.env", 8798),
+ ]
+
+ async def fake_read_once(**kwargs):
+ raise FastReadError("MCP tool error at http://127.0.0.1:8799/mcp: permission_denied")
+
+ with patch("telegram_mcp.fast_read_today.endpoint_attempts", return_value=attempts), patch(
+ "telegram_mcp.fast_read_today.read_once",
+ side_effect=fake_read_once,
+ ) as read_once:
+ import asyncio
+
+ with self.assertRaises(FastReadError):
+ asyncio.run(
+ read_with_failover(
+ chat="me",
+ day="2026-06-02",
+ limit=1,
+ voice=False,
+ sender_names=False,
+ timeout=5.0,
+ )
+ )
+
+ self.assertEqual(read_once.await_count, 1)
diff --git a/mcp/tests/test_mcp_http_client.py b/mcp/tests/test_mcp_http_client.py
index aefe0c8..5f872d4 100644
--- a/mcp/tests/test_mcp_http_client.py
+++ b/mcp/tests/test_mcp_http_client.py
@@ -1,6 +1,17 @@
import unittest
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, patch
-from telegram_mcp.mcp_http_client import McpCliError, endpoint_attempts
+from telegram_mcp.mcp_http_client import (
+ McpCliError,
+ McpToolError,
+ call_tool_with_failover,
+ content_payload,
+ endpoint_attempts,
+ payload_is_tool_error,
+ result_is_tool_error,
+ tool_error_payload,
+)
class McpHttpClientTests(unittest.TestCase):
@@ -50,6 +61,57 @@ def test_endpoint_attempts_allow_explicit_same_account_failover_ports(self):
self.assertEqual([attempt.port for attempt in attempts], [8799, 8798])
+ def test_payload_is_tool_error_detects_contract_error_strings(self):
+ for payload in (
+ "permission_denied: no access | next: ask user",
+ "rate_limited: retry later",
+ "archive_route_blocked: use live Telegram",
+ {"error": "invalid_date_range: date_from must not exceed date_to"},
+ {"code": "human_approval_required"},
+ ):
+ with self.subTest(payload=payload):
+ self.assertTrue(payload_is_tool_error(payload))
+
+ def test_result_is_tool_error_honors_mcp_is_error(self):
+ result = SimpleNamespace(
+ isError=True,
+ structuredContent={"error": "invalid_input: bad chat"},
+ content=[SimpleNamespace(text='{"error": "invalid_input: bad chat"}')],
+ )
+ payload = content_payload(result)
+
+ self.assertTrue(result_is_tool_error(result, payload))
+ self.assertEqual(tool_error_payload(result, payload), {"error": "invalid_input: bad chat"})
+
+ def test_tool_errors_do_not_fail_over_to_next_endpoint(self):
+ attempts = [
+ SimpleNamespace(endpoint="http://127.0.0.1:8799/mcp", env_file="", port=8799),
+ SimpleNamespace(endpoint="http://127.0.0.1:8798/mcp", env_file="", port=8798),
+ ]
+ tool_error = McpToolError(
+ endpoint=attempts[0].endpoint,
+ port=attempts[0].port,
+ payload={"error": "permission_denied: private channel"},
+ elapsed_seconds=0.1,
+ )
+
+ with patch("telegram_mcp.mcp_http_client.endpoint_attempts", return_value=attempts), patch(
+ "telegram_mcp.mcp_http_client.call_tool_once",
+ AsyncMock(side_effect=tool_error),
+ ) as call_once:
+ import asyncio
+
+ with self.assertRaises(McpToolError):
+ asyncio.run(
+ call_tool_with_failover(
+ tool_name="telegram_count_posts",
+ arguments={"chat": "@private"},
+ timeout=5,
+ )
+ )
+
+ self.assertEqual(call_once.await_count, 1)
+
if __name__ == "__main__":
unittest.main()
diff --git a/mcp/tests/test_member_export_paths.py b/mcp/tests/test_member_export_paths.py
index ee72d57..a419985 100644
--- a/mcp/tests/test_member_export_paths.py
+++ b/mcp/tests/test_member_export_paths.py
@@ -27,6 +27,20 @@ def test_explicit_output_dir_outside_private_export_root_is_rejected(self) -> No
with self.assertRaisesRegex(ValueError, "private export root"):
resolve_member_export_dir(str(outside))
+ @unittest.skipIf(not hasattr(Path, "symlink_to"), "symlinks are not supported")
+ def test_symlinked_private_export_root_is_rejected(self) -> None:
+ with TemporaryDirectory() as tmp:
+ home = Path(tmp) / "home"
+ target = Path(tmp) / "outside"
+ target.mkdir(parents=True)
+ root = home / ".cache" / "telegram-mcp" / "member-exports"
+ root.parent.mkdir(parents=True)
+ root.symlink_to(target, target_is_directory=True)
+
+ with patch.dict("os.environ", {"HOME": str(home)}):
+ with self.assertRaisesRegex(ValueError, "must not be a symlink"):
+ resolve_member_export_dir(None)
+
if __name__ == "__main__":
unittest.main()
diff --git a/mcp/tests/test_registration.py b/mcp/tests/test_registration.py
index 59b53c6..88067bb 100644
--- a/mcp/tests/test_registration.py
+++ b/mcp/tests/test_registration.py
@@ -33,6 +33,7 @@ def test_facade_profile_remains_available_when_explicit(self):
self.assertNotIn("create_channel", names)
self.assertNotIn("delete_messages", names)
self.assertNotIn("telegram_export_members", names)
+ self.assertNotIn("send_file", names)
def test_full_tool_registration_surface_is_stable(self):
mcp = FastMCP("test")
diff --git a/mcp/tests/test_runtime.py b/mcp/tests/test_runtime.py
index dfd95e5..9b62cc4 100644
--- a/mcp/tests/test_runtime.py
+++ b/mcp/tests/test_runtime.py
@@ -69,6 +69,33 @@ def test_run_server_allows_disabling_json_response(self):
self.assertFalse(runtime.mcp.settings.json_response)
mcp_run.assert_called_once_with(transport="streamable-http")
+ def test_configure_transport_auth_replaces_oauth_metadata_route(self):
+ original_routes = list(runtime.mcp._custom_starlette_routes)
+ try:
+ with patch.dict(
+ "os.environ",
+ {
+ "TELEGRAM_MCP_TRANSPORT": "streamable-http",
+ "TELEGRAM_MCP_HOST": "127.0.0.1",
+ "TELEGRAM_MCP_PORT": "8799",
+ "TELEGRAM_MCP_AUTH_TOKEN": "test-token",
+ },
+ clear=False,
+ ):
+ get_settings.cache_clear()
+ runtime.configure_transport_auth("streamable-http")
+ runtime.configure_transport_auth("streamable-http")
+
+ routes = [
+ route
+ for route in runtime.mcp._custom_starlette_routes
+ if getattr(route, "path", None)
+ == runtime.OAUTH_AUTHORIZATION_SERVER_METADATA_PATH
+ ]
+ self.assertEqual(len(routes), 1)
+ finally:
+ runtime.mcp._custom_starlette_routes = original_routes
+
def test_run_server_disconnects_shared_wrapper_on_shutdown(self):
with patch("telegram_mcp.runtime.shared_mode_enabled", return_value=True):
with patch.object(runtime.mcp, "run") as mcp_run:
diff --git a/mcp/tests/test_send_confirmation.py b/mcp/tests/test_send_confirmation.py
index 2e0cb38..42273ff 100644
--- a/mcp/tests/test_send_confirmation.py
+++ b/mcp/tests/test_send_confirmation.py
@@ -3,12 +3,15 @@
from __future__ import annotations
import unittest
+from http.client import HTTPConnection
+from urllib.parse import urlencode
from unittest.mock import patch
+import telegram_mcp.approval_server as approval_server
from telegram_mcp.client import TelegramWrapper
from telegram_mcp.config import Settings
from telegram_mcp.errors import ToolContractError
-from telegram_mcp.send_confirmation import SendConfirmationStore
+from telegram_mcp.send_confirmation import SendConfirmationStore, bind_confirmation_store
from tests.test_client import DummyTelegramClient, _run
@@ -104,6 +107,64 @@ def test_reject_blocks_consume(self):
store.consume(token, payload, approval_required=True)
self.assertEqual(ctx.exception.code, "confirmation_rejected")
+ def test_approve_by_token_then_consume_removes_token_lookup(self):
+ store = SendConfirmationStore(ttl_seconds=60)
+ payload = {"chat": "@x", "text_hash": "abc"}
+ preview_id, token, _ = store.mint(payload, preview_text="hi")
+
+ store.approve(token)
+ store.consume(preview_id, None, approval_required=True, preview_id_only=True)
+
+ self.assertIsNone(store.get(token))
+ self.assertIsNone(store.get(preview_id))
+
+
+class ApprovalServerTests(unittest.TestCase):
+ def tearDown(self):
+ approval_server.stop_approval_server()
+
+ def test_rejects_non_loopback_host(self):
+ with self.assertRaisesRegex(ValueError, "loopback"):
+ approval_server.start_approval_server(host="0.0.0.0", port=0)
+
+ def test_get_does_not_approve_and_post_requires_nonce(self):
+ store = SendConfirmationStore(ttl_seconds=60)
+ payload = {"chat": "@x", "send_tool": "send_dialog_message", "text_hash": "abc"}
+ _preview_id, token, _ = store.mint(payload, preview_text="hi")
+ record = store.get(token)
+ assert record is not None
+ bind_confirmation_store(store)
+ approval_server.start_approval_server(host="127.0.0.1", port=0)
+ server = approval_server._server
+ assert server is not None
+ port = server.server_address[1]
+
+ conn = HTTPConnection("127.0.0.1", port)
+ conn.request("GET", f"/telegram/approve?token={token}&action=approve")
+ response = conn.getresponse()
+ response.read()
+ conn.close()
+ self.assertEqual(response.status, 200)
+ self.assertEqual(store.get(token).approval_state, "pending") # type: ignore[union-attr]
+
+ body = urlencode({"token": token, "nonce": "wrong", "action": "approve"})
+ conn = HTTPConnection("127.0.0.1", port)
+ conn.request("POST", "/telegram/approve", body=body, headers={"Content-Type": "application/x-www-form-urlencoded"})
+ response = conn.getresponse()
+ response.read()
+ conn.close()
+ self.assertEqual(response.status, 400)
+ self.assertEqual(store.get(token).approval_state, "pending") # type: ignore[union-attr]
+
+ body = urlencode({"token": token, "nonce": record.one_time_nonce, "action": "approve"})
+ conn = HTTPConnection("127.0.0.1", port)
+ conn.request("POST", "/telegram/approve", body=body, headers={"Content-Type": "application/x-www-form-urlencoded"})
+ response = conn.getresponse()
+ response.read()
+ conn.close()
+ self.assertEqual(response.status, 200)
+ self.assertEqual(store.get(token).approval_state, "approved") # type: ignore[union-attr]
+
if __name__ == "__main__":
unittest.main()
diff --git a/mcp/tests/test_tg_cli.py b/mcp/tests/test_tg_cli.py
index e26b09e..e10686b 100644
--- a/mcp/tests/test_tg_cli.py
+++ b/mcp/tests/test_tg_cli.py
@@ -2,25 +2,11 @@
from unittest.mock import AsyncMock, patch
from telegram_mcp.metadata_tools_spec import COUNT_SPECS_BY_CLI, LIST_SPECS_BY_CLI
-from telegram_mcp.tg_cli import _wrap_ok, cmd_count_metadata, cmd_count_posts, cmd_list_metadata, route_task
+from telegram_mcp.mcp_http_client import McpToolError
+from telegram_mcp.tg_cli import cmd_count_metadata, cmd_count_posts, cmd_list_metadata, route_task
class TgCliTests(unittest.TestCase):
- def test_wrap_ok_marks_tool_error_payload_as_failure(self):
- payload = _wrap_ok(
- command="read today",
- endpoint="http://127.0.0.1:8799/mcp",
- endpoint_port=8799,
- elapsed_seconds=0.1,
- payload="Error executing tool telegram_read: raw failure",
- intent="live_today",
- )
-
- self.assertIs(payload["ok"], False)
- self.assertEqual(payload["error"], "telegram_tool_error")
- self.assertEqual(payload["data_source"], "live_telegram")
- self.assertEqual(payload["tool_error_payload"], "Error executing tool telegram_read: raw failure")
-
def test_route_task_plans_channel_post_count_without_execution(self):
payload = route_task("@sral_v_nastav сколько постов в этом канале всего?")
@@ -116,6 +102,35 @@ async def fake_call_tool_with_failover(**kwargs):
self.assertTrue(payload["ok"])
self.assertEqual(payload["intent"], "count_channel_videos")
+ def test_count_metadata_records_error_telemetry_for_tool_error(self):
+ async def fake_call_tool_with_failover(**kwargs):
+ raise McpToolError(
+ endpoint="http://127.0.0.1:8799/mcp",
+ port=8799,
+ payload={"error": "invalid_input: bad chat"},
+ elapsed_seconds=0.1,
+ )
+
+ with patch("telegram_mcp.tg_cli.call_tool_with_failover", AsyncMock(side_effect=fake_call_tool_with_failover)):
+ with patch("telegram_mcp.tg_cli.record_telemetry") as telemetry:
+ import asyncio
+
+ payload = asyncio.run(
+ cmd_count_metadata(
+ chat="@bad",
+ spec=COUNT_SPECS_BY_CLI["videos"],
+ timeout=5,
+ endpoint=None,
+ env_file=None,
+ account="main",
+ )
+ )
+
+ self.assertFalse(payload["ok"])
+ self.assertEqual(payload["error"], "telegram_tool_error")
+ self.assertEqual(payload["tool_error_payload"], {"error": "invalid_input: bad chat"})
+ self.assertEqual(telemetry.call_args.kwargs["status"], "error")
+
def test_list_metadata_uses_spec_tool(self):
async def fake_call_tool_with_failover(**kwargs):
self.assertEqual(kwargs["tool_name"], "telegram_list_links")