diff --git a/backend/.env.example b/backend/.env.example index ded015af..7f4865ac 100644 --- a/backend/.env.example +++ b/backend/.env.example @@ -17,3 +17,13 @@ APPLE_TEAM_ID=ABCDE12345 # GitHub Releases (required for --publish) # ============================================================================= GH_TOKEN=ghp_your-github-personal-access-token + +# ============================================================================= +# Telegram MCP (app-level credentials, NOT per-user) +# ============================================================================= +# Register an app once at https://my.telegram.org/apps (Platform: Desktop). +# These identify OpenSwarm itself to Telegram; end users sign in with their +# own phone + OTP, never these values. Per Telegram TOS keep the hash secret — +# never commit it. Rotate via the same page if it ever leaks. +OPENSWARM_TELEGRAM_API_ID= +OPENSWARM_TELEGRAM_API_HASH= diff --git a/backend/apps/agents/agent_manager.py b/backend/apps/agents/agent_manager.py index 598233cd..c1553483 100644 --- a/backend/apps/agents/agent_manager.py +++ b/backend/apps/agents/agent_manager.py @@ -492,7 +492,7 @@ class AgentManager: # the agent MUST stop the task, tell the user the retry-after, and # NOT retry. Without this guidance, agents tend to loop trying # alternative tools or even shell out to filesystem search. - if tool.name.lower() in ("instagram", "linkedin"): + if tool.name.lower() in ("instagram", "linkedin", "telegram"): lines.append( f" RATE LIMIT BEHAVIOR (HARD RULE): If a {tool.name} tool returns " "rate_limited: true, or any tool call here returns a 'deny' with " diff --git a/backend/apps/telegram_mcp/__init__.py b/backend/apps/telegram_mcp/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/backend/apps/telegram_mcp/__main__.py b/backend/apps/telegram_mcp/__main__.py new file mode 100644 index 00000000..4b3433d7 --- /dev/null +++ b/backend/apps/telegram_mcp/__main__.py @@ -0,0 +1,12 @@ +"""Module entrypoint: `python -m backend.apps.telegram_mcp`. + +Spawned by OpenSwarm when an agent invokes a Telegram tool. Auth happens +before this server starts — the backend's /credentials/telegram/* endpoints +drive the phone -> code -> (optional 2FA) flow via Telethon and persist a +session file at ~/.telegram_mcp/sessions/.session. This server +loads that session, never the password. +""" +from backend.apps.telegram_mcp.server import main + +if __name__ == "__main__": + main() diff --git a/backend/apps/telegram_mcp/rate_limiter.py b/backend/apps/telegram_mcp/rate_limiter.py new file mode 100644 index 00000000..39657b71 --- /dev/null +++ b/backend/apps/telegram_mcp/rate_limiter.py @@ -0,0 +1,174 @@ +""" +Rate limiter for Telegram MCP tools. + +Why this exists: Telegram bans on spam patterns (rapid identical sends, +broad-pattern forwards, mass DMs to non-contacts). Proactive caps + jitter +keep an unsupervised agent well below the threshold that triggers a ban. + +Caps are more generous than Instagram because Telegram is itself more +permissive, but still well under the "this is automation" line. + +State persists across server restarts to ~/.telegram-mcp-rate-limits.json +so a relaunch does not reset the daily budget. + +Per-category defaults: + + category per_minute per_hour per_day jitter + send 30 200 1000 0.5-2.0s + forward 20 150 600 0.5-2.0s + search 60 500 3000 0.0-0.5s + lookup 60 500 3000 0.0-0.5s + +Override any cap via env var, e.g.: + TG_RATE_LIMIT_SEND_PER_DAY=500 +""" +from __future__ import annotations + +import functools +import json +import logging +import os +import random +import time +from pathlib import Path +from typing import Any, Callable, Dict, List, Tuple + +logger = logging.getLogger(__name__) + +_STATE_PATH = Path.home() / ".telegram-mcp-rate-limits.json" + +DEFAULTS: Dict[str, Dict[str, Any]] = { + "send": {"per_minute": 30, "per_hour": 200, "per_day": 1000, "jitter": (0.5, 2.0)}, + "forward": {"per_minute": 20, "per_hour": 150, "per_day": 600, "jitter": (0.5, 2.0)}, + "search": {"per_minute": 60, "per_hour": 500, "per_day": 3000, "jitter": (0.0, 0.5)}, + "lookup": {"per_minute": 60, "per_hour": 500, "per_day": 3000, "jitter": (0.0, 0.5)}, +} + + +def _env_override(category: str, key: str, default: int) -> int: + var = f"TG_RATE_LIMIT_{category.upper()}_{key.upper()}" + raw = os.environ.get(var) + if raw is None: + return default + try: + value = int(raw) + if value <= 0: + return default + return value + except ValueError: + return default + + +def _get_limits(category: str) -> Dict[str, Any]: + d = DEFAULTS[category] + return { + "per_minute": _env_override(category, "per_minute", d["per_minute"]), + "per_hour": _env_override(category, "per_hour", d["per_hour"]), + "per_day": _env_override(category, "per_day", d["per_day"]), + "jitter": d["jitter"], + } + + +def _load_state() -> Dict[str, List[float]]: + if not _STATE_PATH.exists(): + return {} + try: + data = json.loads(_STATE_PATH.read_text()) + return {k: [float(t) for t in v] for k, v in data.items() if isinstance(v, list)} + except Exception as exc: + logger.warning("Could not load rate-limit state, starting fresh: %s", exc) + return {} + + +def _save_state(state: Dict[str, List[float]]) -> None: + try: + _STATE_PATH.write_text(json.dumps(state)) + except Exception as exc: + logger.warning("Failed to persist rate-limit state: %s", exc) + + +def _prune(timestamps: List[float], now: float, window_s: int) -> List[float]: + cutoff = now - window_s + return [t for t in timestamps if t >= cutoff] + + +def _fmt_duration(seconds: int) -> str: + if seconds < 60: + return f"{seconds}s" + if seconds < 3600: + return f"{seconds // 60}m {seconds % 60}s" + return f"{seconds // 3600}h {(seconds % 3600) // 60}m" + + +def _check_budget( + category: str, + limits: Dict[str, Any], + state: Dict[str, List[float]], +) -> Tuple[bool, str, int, Dict[str, int]]: + """Returns (ok, reason_if_blocked, retry_after_seconds, current_counts).""" + now = time.time() + pruned_day = _prune(state.get(category, []), now, 24 * 3600) + state[category] = pruned_day + + counts: Dict[str, int] = {} + for window_name, window_s in (("per_minute", 60), ("per_hour", 3600), ("per_day", 86400)): + in_window = _prune(pruned_day, now, window_s) + counts[window_name] = len(in_window) + + for window_name, window_s in (("per_minute", 60), ("per_hour", 3600), ("per_day", 86400)): + in_window = _prune(pruned_day, now, window_s) + limit = limits[window_name] + if len(in_window) >= limit: + oldest = min(in_window) + retry_after = int((oldest + window_s) - now) + 1 + label = window_name.replace("per_", "") + return ( + False, + f"{category} hit {limit}/{label} cap (currently {len(in_window)}). Retry in {_fmt_duration(retry_after)}.", + retry_after, + counts, + ) + return (True, "", 0, counts) + + +def rate_limited(category: str) -> Callable[[Callable[..., Dict[str, Any]]], Callable[..., Dict[str, Any]]]: + """Decorator: enforce per-category limits and apply jitter before the call. + + Returns a structured error dict to the MCP client when blocked instead of + raising, so the agent can surface "try again in 4h 12m" to the user + instead of failing opaquely. + """ + if category not in DEFAULTS: + raise ValueError(f"Unknown rate-limit category: {category}") + + def decorator(func: Callable[..., Dict[str, Any]]) -> Callable[..., Dict[str, Any]]: + @functools.wraps(func) + def wrapper(*args: Any, **kwargs: Any) -> Dict[str, Any]: + limits = _get_limits(category) + state = _load_state() + ok, reason, retry_after, current = _check_budget(category, limits, state) + if not ok: + logger.warning("Rate limit blocked %s: %s", func.__name__, reason) + return { + "success": False, + "rate_limited": True, + "category": category, + "message": ( + f"RATE LIMIT HIT — STOP HERE. {reason} DO NOT retry this tool. " + "DO NOT try alternative tools to accomplish the same goal. DO NOT " + "search the filesystem or look up the package source. Tell the user " + "the retry-after time in plain English and END the task. This " + "protects the Telegram account from being flagged for spam." + ), + "retry_after_seconds": retry_after, + "limits": {k: limits[k] for k in ("per_minute", "per_hour", "per_day")}, + "current": current, + } + state.setdefault(category, []).append(time.time()) + _save_state(state) + lo, hi = limits["jitter"] + if hi > 0: + time.sleep(random.uniform(lo, hi)) + return func(*args, **kwargs) + return wrapper + return decorator diff --git a/backend/apps/telegram_mcp/server.py b/backend/apps/telegram_mcp/server.py new file mode 100644 index 00000000..b47d0c9e --- /dev/null +++ b/backend/apps/telegram_mcp/server.py @@ -0,0 +1,376 @@ +"""Telegram MCP server (vendored into OpenSwarm). + +MTProto user-account access via Telethon. Auth is driven by the OpenSwarm +backend's /credentials/telegram/* endpoints before this server ever starts; +this process just loads the prebuilt session. + +Tools target four use cases: + - send DMs and channel messages (send_message, send_file, send_voice) + - read and summarize inbox (list_dialogs, get_messages, search_messages) + - forward / filter rules (forward_message) + - diagnostics (get_me, close_session) + +Rate limiting is enforced per-tool via @rate_limited from .rate_limiter so +no single agent run can exceed the daily caps that keep Telegram from +flagging the connected account. + +Env contract (set by OpenSwarm at spawn time): + TELEGRAM_PHONE E.164 phone identifying which session to load + OPENSWARM_TELEGRAM_API_ID app credentials (registered once by the team + OPENSWARM_TELEGRAM_API_HASH at https://my.telegram.org/apps) +""" +from __future__ import annotations + +import asyncio +import logging +import os +import re +import sys +from pathlib import Path +from typing import Any, Dict, List, Optional + +from mcp.server.fastmcp import FastMCP +from telethon import TelegramClient +from telethon.tl.types import Message + +from .rate_limiter import rate_limited + +logger = logging.getLogger(__name__) +logger.setLevel(logging.DEBUG) + +SESSION_DIR = Path.home() / ".telegram_mcp" / "sessions" + +INSTRUCTIONS = """ +Telegram via Telethon (MTProto user account). 9 tools cover send, read, +search, forward, and diagnostics. Per-category rate limits are enforced +server-side to protect the connected account from spam bans. +""" + +mcp = FastMCP(name="Telegram", instructions=INSTRUCTIONS) + + +def _sanitize_phone(phone: str) -> str: + """Phone-as-filename: strip leading + and any non-digit so the SQLite + session file is always a portable bare-digits name.""" + return re.sub(r"\D", "", phone or "") + + +def _session_path(phone: str) -> Path: + return SESSION_DIR / _sanitize_phone(phone) # Telethon appends .session + + +_client: Optional[TelegramClient] = None +_loop: Optional[asyncio.AbstractEventLoop] = None + + +def _get_client() -> TelegramClient: + """Lazy-init the singleton client. Tools run synchronously and share one + background event loop so we don't pay the connect cost on every call.""" + global _client + if _client is not None: + return _client + + phone = os.getenv("TELEGRAM_PHONE", "").strip() + api_id_raw = os.getenv("OPENSWARM_TELEGRAM_API_ID", "").strip() + api_hash = os.getenv("OPENSWARM_TELEGRAM_API_HASH", "").strip() + if not phone or not api_id_raw or not api_hash: + raise RuntimeError( + "Telegram MCP missing env: TELEGRAM_PHONE, OPENSWARM_TELEGRAM_API_ID, OPENSWARM_TELEGRAM_API_HASH " + "must all be set. Connect Telegram via the OpenSwarm Tools page first." + ) + try: + api_id = int(api_id_raw) + except ValueError as exc: + raise RuntimeError(f"OPENSWARM_TELEGRAM_API_ID must be an integer, got {api_id_raw!r}") from exc + + SESSION_DIR.mkdir(parents=True, exist_ok=True) + session_file = _session_path(phone) + if not session_file.with_suffix(".session").exists(): + raise RuntimeError( + f"No Telegram session at {session_file}.session. Connect Telegram via the OpenSwarm Tools page first." + ) + + _client = TelegramClient(str(session_file), api_id, api_hash) + return _client + + +def _run(coro): + """Run an async Telethon coroutine from a sync MCP tool body.""" + global _loop + if _loop is None or _loop.is_closed(): + _loop = asyncio.new_event_loop() + return _loop.run_until_complete(coro) + + +async def _ensure_connected() -> None: + client = _get_client() + if not client.is_connected(): + await client.connect() + if not await client.is_user_authorized(): + raise RuntimeError( + "Telegram session exists but is not authorized. Disconnect and reconnect via the OpenSwarm Tools page." + ) + + +def _message_summary(m: Message) -> Dict[str, Any]: + return { + "id": m.id, + "date": m.date.isoformat() if m.date else None, + "from_id": getattr(m.from_id, "user_id", None) if m.from_id else None, + "text": (m.message or "")[:2000], + "has_media": bool(m.media), + "reply_to": m.reply_to_msg_id, + } + + +def _dialog_summary(d: Any) -> Dict[str, Any]: + entity = d.entity + return { + "id": d.id, + "name": d.name, + "is_user": d.is_user, + "is_group": d.is_group, + "is_channel": d.is_channel, + "unread_count": d.unread_count, + "username": getattr(entity, "username", None), + "last_message": (d.message.message or "")[:500] if d.message and d.message.message else None, + } + + +# ---- send ------------------------------------------------------------------- + + +@mcp.tool() +@rate_limited("send") +def send_message(chat: str, message: str, reply_to: Optional[int] = None) -> Dict[str, Any]: + """Send a text message to a Telegram user, group, or channel. + + Args: + chat: Username (with or without leading @), phone number, or numeric + chat ID. "me" sends to your own Saved Messages. + message: Text body. Telegram-flavored markdown is supported. + reply_to: Optional message ID to reply to in the target chat. + Returns: + Dictionary with success flag and the sent message's id. + """ + async def _do(): + await _ensure_connected() + client = _get_client() + sent = await client.send_message(chat, message, reply_to=reply_to) + return {"success": True, "message_id": sent.id, "chat": chat} + try: + return _run(_do()) + except Exception as e: # noqa: BLE001 + return {"success": False, "message": str(e)} + + +@mcp.tool() +@rate_limited("send") +def send_file(chat: str, file_path: str, caption: Optional[str] = None) -> Dict[str, Any]: + """Send a photo, video, document, or any file to a Telegram chat. + + Args: + chat: Same identifier shape as send_message. + file_path: Absolute path to a file on disk. + caption: Optional caption shown under the media. + Returns: + Dictionary with success flag and the sent message's id. + """ + if not os.path.exists(file_path): + return {"success": False, "message": f"File not found: {file_path}"} + async def _do(): + await _ensure_connected() + client = _get_client() + sent = await client.send_file(chat, file_path, caption=caption) + return {"success": True, "message_id": sent.id, "chat": chat} + try: + return _run(_do()) + except Exception as e: # noqa: BLE001 + return {"success": False, "message": str(e)} + + +@mcp.tool() +@rate_limited("send") +def send_voice(chat: str, file_path: str, caption: Optional[str] = None) -> Dict[str, Any]: + """Send a voice note (.ogg, .opus, .mp3) to a Telegram chat. + + Args: + chat: Same identifier shape as send_message. + file_path: Absolute path to an audio file. + caption: Optional caption. + Returns: + Dictionary with success flag and the sent message's id. + """ + if not os.path.exists(file_path): + return {"success": False, "message": f"File not found: {file_path}"} + async def _do(): + await _ensure_connected() + client = _get_client() + sent = await client.send_file(chat, file_path, caption=caption, voice_note=True) + return {"success": True, "message_id": sent.id, "chat": chat} + try: + return _run(_do()) + except Exception as e: # noqa: BLE001 + return {"success": False, "message": str(e)} + + +# ---- forward --------------------------------------------------------------- + + +@mcp.tool() +@rate_limited("forward") +def forward_message(from_chat: str, message_id: int, to_chat: str) -> Dict[str, Any]: + """Forward a single message from one chat to another. + + Args: + from_chat: Source chat identifier. + message_id: Numeric message ID in the source chat. + to_chat: Destination chat identifier. + Returns: + Dictionary with success flag and the forwarded message id. + """ + async def _do(): + await _ensure_connected() + client = _get_client() + sent = await client.forward_messages(to_chat, message_id, from_chat) + new_id = sent.id if not isinstance(sent, list) else (sent[0].id if sent else None) + return {"success": True, "new_message_id": new_id, "to": to_chat} + try: + return _run(_do()) + except Exception as e: # noqa: BLE001 + return {"success": False, "message": str(e)} + + +# ---- read / inbox ---------------------------------------------------------- + + +@mcp.tool() +@rate_limited("lookup") +def list_dialogs(limit: int = 20, archived: bool = False) -> Dict[str, Any]: + """List your recent Telegram dialogs (chats, groups, channels). + + Args: + limit: Max number of dialogs to return (default 20). + archived: If True, fetch from the archive folder instead of the main inbox. + Returns: + Dictionary with success flag and a list of dialog summaries. + """ + async def _do(): + await _ensure_connected() + client = _get_client() + dialogs = await client.get_dialogs(limit=limit, archived=archived) + return {"success": True, "dialogs": [_dialog_summary(d) for d in dialogs]} + try: + return _run(_do()) + except Exception as e: # noqa: BLE001 + return {"success": False, "message": str(e)} + + +@mcp.tool() +@rate_limited("lookup") +def get_messages(chat: str, limit: int = 20, offset_id: int = 0) -> Dict[str, Any]: + """Read recent messages from a Telegram chat. + + Args: + chat: Same identifier shape as send_message. + limit: Max number of messages (default 20, max ~100 per call). + offset_id: Pagination cursor (message id); 0 means newest. + Returns: + Dictionary with success flag and a list of message summaries + (newest first). + """ + async def _do(): + await _ensure_connected() + client = _get_client() + messages = await client.get_messages(chat, limit=limit, offset_id=offset_id) + return {"success": True, "messages": [_message_summary(m) for m in messages]} + try: + return _run(_do()) + except Exception as e: # noqa: BLE001 + return {"success": False, "message": str(e)} + + +@mcp.tool() +@rate_limited("search") +def search_messages(query: str, chat: Optional[str] = None, limit: int = 20) -> Dict[str, Any]: + """Full-text search across messages. + + Args: + query: Search string (Telegram's server-side text search). + chat: Optional chat to scope the search to. Omit to search global inbox. + limit: Max results (default 20). + Returns: + Dictionary with success flag and a list of matching message summaries. + """ + async def _do(): + await _ensure_connected() + client = _get_client() + messages = await client.get_messages(chat, search=query, limit=limit) + return {"success": True, "messages": [_message_summary(m) for m in messages]} + try: + return _run(_do()) + except Exception as e: # noqa: BLE001 + return {"success": False, "message": str(e)} + + +# ---- plumbing -------------------------------------------------------------- + + +@mcp.tool() +@rate_limited("lookup") +def get_me() -> Dict[str, Any]: + """Return profile info for the currently signed-in Telegram account. + + Useful as a connectivity check before running other tools. + """ + async def _do(): + await _ensure_connected() + client = _get_client() + me = await client.get_me() + return { + "success": True, + "user_id": me.id, + "username": me.username, + "first_name": me.first_name, + "last_name": me.last_name, + "phone": me.phone, + "is_bot": me.bot, + "is_premium": getattr(me, "premium", False), + } + try: + return _run(_do()) + except Exception as e: # noqa: BLE001 + return {"success": False, "message": str(e)} + + +@mcp.tool() +def close_session() -> Dict[str, Any]: + """Cleanly disconnect the Telegram client. Good agent hygiene at end of task.""" + global _client, _loop + try: + if _client and _client.is_connected(): + _run(_client.disconnect()) + _client = None + if _loop and not _loop.is_closed(): + _loop.close() + _loop = None + return {"success": True} + except Exception as e: # noqa: BLE001 + return {"success": False, "message": str(e)} + + +def main() -> None: + """Spawn entrypoint. Validate session presence then start stdio loop.""" + try: + # Trigger lazy-init's preflight checks (env vars + session file) so + # we fail fast at startup instead of mid-tool-call. + _get_client() + except RuntimeError as e: + logger.error(str(e)) + sys.exit(1) + logger.info(f"Telegram MCP ready for phone ending …{(os.getenv('TELEGRAM_PHONE','') or '')[-4:]}") + mcp.run(transport="stdio") + + +if __name__ == "__main__": + main() diff --git a/backend/apps/tools_lib/mcp_rate_limiter.py b/backend/apps/tools_lib/mcp_rate_limiter.py index 8851224d..2ab3faf2 100644 --- a/backend/apps/tools_lib/mcp_rate_limiter.py +++ b/backend/apps/tools_lib/mcp_rate_limiter.py @@ -47,6 +47,23 @@ POLICIES: dict[str, dict] = { ("*", "lookup"), ], }, + # Telegram has its own server-side rate limiter inside the vendored + # package, so this entry is intentionally absent to avoid double-throttling. + # If you want belt-and-suspenders, uncomment the block below: + # "telegram": { + # "categories": { + # "send": {"per_minute": 30, "per_hour": 200, "per_day": 1000, "jitter": (0.5, 2.0)}, + # "forward": {"per_minute": 20, "per_hour": 150, "per_day": 600, "jitter": (0.5, 2.0)}, + # "search": {"per_minute": 60, "per_hour": 500, "per_day": 3000, "jitter": (0.0, 0.5)}, + # "lookup": {"per_minute": 60, "per_hour": 500, "per_day": 3000, "jitter": (0.0, 0.5)}, + # }, + # "tools": [ + # ("send_*", "send"), + # ("forward_*", "forward"), + # ("search_*", "search"), + # ("*", "lookup"), + # ], + # }, } diff --git a/backend/apps/tools_lib/tools_lib.py b/backend/apps/tools_lib/tools_lib.py index b0ae1ab7..b1afc470 100644 --- a/backend/apps/tools_lib/tools_lib.py +++ b/backend/apps/tools_lib/tools_lib.py @@ -343,6 +343,27 @@ def derive_mcp_config(tool: ToolDefinition) -> Optional[dict]: if config.get("command") == "python": config["command"] = _sys.executable + # Telegram MCP: same vendored-package treatment as Instagram. Also + # injects the OpenSwarm app-level Telegram credentials from backend env + # (registered once at https://my.telegram.org/apps) so the server can + # talk to MTProto. End-users never see these values; their own login + # is identified by TELEGRAM_PHONE which comes from tool.credentials. + if tool.name.lower() == "telegram" and config.get("type") == "stdio": + env = config.setdefault("env", {}) + _project_root = os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))) + existing_pp = env.get("PYTHONPATH") or os.environ.get("PYTHONPATH", "") + env["PYTHONPATH"] = (_project_root + os.pathsep + existing_pp) if existing_pp else _project_root + import sys as _sys + if config.get("command") == "python": + config["command"] = _sys.executable + # App-level Telegram credentials — never per-user. + tg_api_id = os.environ.get("OPENSWARM_TELEGRAM_API_ID", "") + tg_api_hash = os.environ.get("OPENSWARM_TELEGRAM_API_HASH", "") + if tg_api_id: + env["OPENSWARM_TELEGRAM_API_ID"] = tg_api_id + if tg_api_hash: + env["OPENSWARM_TELEGRAM_API_HASH"] = tg_api_hash + # Microsoft 365 MCP: use a stable token cache path shared across process spawns if tool.name.lower() == "microsoft 365" and config.get("type") == "stdio": env = config.setdefault("env", {}) @@ -1448,3 +1469,162 @@ async def instagram_from_browser(payload: dict) -> dict: tool.connected_account_email = f"@{username}" _save(tool) return {"ok": True, "username": username, "user_id": user_id_str} + + +# --------------------------------------------------------------------------- +# Telegram: phone + OTP + (optional) 2FA password flow via Telethon. +# +# The vendored MCP server at backend.apps.telegram_mcp loads a Telethon +# session file at ~/.telegram_mcp/sessions/.session. These +# endpoints drive the multi-step auth that builds that session: /start +# requests the SMS code, /verify submits it, /password handles cloud 2FA. +# A short-lived in-memory cache keeps the active Telethon client between +# /start and /verify so we don't lose phone_code_hash state. +# --------------------------------------------------------------------------- +import re as _re_tg +_TG_PENDING: dict[str, dict] = {} + + +def _tg_session_dir() -> "Path": + from pathlib import Path + p = Path.home() / ".telegram_mcp" / "sessions" + p.mkdir(parents=True, exist_ok=True) + return p + + +def _tg_sanitize_phone(phone: str) -> str: + return _re_tg.sub(r"\D", "", phone or "") + + +def _tg_app_creds() -> tuple[int, str]: + api_id_raw = os.environ.get("OPENSWARM_TELEGRAM_API_ID", "").strip() + api_hash = os.environ.get("OPENSWARM_TELEGRAM_API_HASH", "").strip() + if not api_id_raw or not api_hash: + raise ValueError( + "Telegram is not configured on this OpenSwarm install. " + "Set OPENSWARM_TELEGRAM_API_ID and OPENSWARM_TELEGRAM_API_HASH " + "in backend/.env (register an app at https://my.telegram.org/apps)." + ) + return int(api_id_raw), api_hash + + +async def _tg_finalize_after_auth(tool, phone: str, client) -> None: + """Persist the tool config after a successful Telethon sign_in.""" + try: + me = await client.get_me() + except Exception: # noqa: BLE001 + me = None + try: + await client.disconnect() + except Exception: # noqa: BLE001 + pass + + digits = _tg_sanitize_phone(phone) + if me and me.username: + tool.connected_account_email = f"@{me.username}" + elif me and (me.first_name or me.last_name): + tool.connected_account_email = f"{(me.first_name or '').strip()} {(me.last_name or '').strip()}".strip() + else: + tool.connected_account_email = f"+{digits}" + tool.credentials = {"TELEGRAM_PHONE": phone} + tool.auth_type = "env_vars" + tool.auth_status = "connected" + _save(tool) + + +@tools_lib.router.post("/credentials/telegram/start") +async def telegram_start(payload: dict) -> dict: + """Step 1: send the Telegram OTP to the user's phone. + + Body: {tool_id: str, phone: str (E.164, e.g. +15551234567)} + Returns: {ok: true, needs_code: bool} or {ok: false, error: str}. + """ + tool_id = payload.get("tool_id") or "" + phone = (payload.get("phone") or "").strip() + if not tool_id or not phone: + return {"ok": False, "error": "tool_id and phone are required"} + tool = _load(tool_id) + try: + api_id, api_hash = _tg_app_creds() + except ValueError as e: + return {"ok": False, "error": str(e)} + + from telethon import TelegramClient + session_file = _tg_session_dir() / _tg_sanitize_phone(phone) + client = TelegramClient(str(session_file), api_id, api_hash) + try: + await client.connect() + if await client.is_user_authorized(): + # Session already valid from a prior login. + await _tg_finalize_after_auth(tool, phone, client) + return {"ok": True, "needs_code": False, "already_authorized": True} + sent = await client.send_code_request(phone) + except Exception as e: # noqa: BLE001 + try: await client.disconnect() + except Exception: pass + return {"ok": False, "error": (str(e) or type(e).__name__)[:300]} + + _TG_PENDING[tool_id] = { + "phone": phone, + "phone_code_hash": sent.phone_code_hash, + "client": client, + } + return {"ok": True, "needs_code": True} + + +@tools_lib.router.post("/credentials/telegram/verify") +async def telegram_verify(payload: dict) -> dict: + """Step 2: submit the OTP code received in Telegram. + + Body: {tool_id: str, code: str} + Returns: {ok: true} or {ok: false, needs_password: bool, error: str?}. + """ + tool_id = payload.get("tool_id") or "" + code = (payload.get("code") or "").strip() + if not tool_id or not code: + return {"ok": False, "error": "tool_id and code are required"} + pending = _TG_PENDING.get(tool_id) + if not pending: + return {"ok": False, "error": "no pending Telegram sign-in; call /start first"} + tool = _load(tool_id) + + from telethon.errors import SessionPasswordNeededError + client = pending["client"] + phone = pending["phone"] + try: + await client.sign_in(phone, code, phone_code_hash=pending["phone_code_hash"]) + except SessionPasswordNeededError: + return {"ok": False, "needs_password": True} + except Exception as e: # noqa: BLE001 + return {"ok": False, "error": (str(e) or type(e).__name__)[:300]} + + await _tg_finalize_after_auth(tool, phone, client) + _TG_PENDING.pop(tool_id, None) + return {"ok": True} + + +@tools_lib.router.post("/credentials/telegram/password") +async def telegram_password(payload: dict) -> dict: + """Step 3 (only if account has cloud 2FA): submit the 2FA password. + + Body: {tool_id: str, password: str} + """ + tool_id = payload.get("tool_id") or "" + password = payload.get("password") or "" + if not tool_id or not password: + return {"ok": False, "error": "tool_id and password are required"} + pending = _TG_PENDING.get(tool_id) + if not pending: + return {"ok": False, "error": "no pending Telegram sign-in"} + tool = _load(tool_id) + + client = pending["client"] + phone = pending["phone"] + try: + await client.sign_in(password=password) + except Exception as e: # noqa: BLE001 + return {"ok": False, "error": (str(e) or type(e).__name__)[:300]} + + await _tg_finalize_after_auth(tool, phone, client) + _TG_PENDING.pop(tool_id, None) + return {"ok": True} diff --git a/backend/requirements.txt b/backend/requirements.txt index 195b184d..f2528111 100644 --- a/backend/requirements.txt +++ b/backend/requirements.txt @@ -20,6 +20,10 @@ trafilatura # login server-side before persisting credentials (verify-before-save flow, # same shape as the OAuth integrations). instagrapi>=2.0 +# telethon: Telegram MCP backend (MTProto user-account auth + session +# persistence). The vendored Telegram MCP server (backend.apps.telegram_mcp) +# loads sessions via Telethon and runs as a stdio subprocess. +telethon~=1.36 # tzlocal: dev-mode fallback for resolving the user's IANA timezone when # Electron's OPENSWARM_TIMEZONE env var isn't set (i.e. `bash run.sh`). # Packaged builds get the env var directly so this is a safety net. diff --git a/frontend/src/app/pages/Tools/Tools.tsx b/frontend/src/app/pages/Tools/Tools.tsx index bcc25d4b6..4eed8f6c 100644 --- a/frontend/src/app/pages/Tools/Tools.tsx +++ b/frontend/src/app/pages/Tools/Tools.tsx @@ -333,6 +333,30 @@ const INTEGRATIONS: Integration[] = [ ), }, + { + id: 'telegram', + name: 'Telegram', + description: + 'Telegram via MTProto user account. 9 tools: send/receive messages, send files and voice, list dialogs, read history, search, forward, and diagnostics. Per-category rate limits enforced server-side to protect from spam bans. Single sign-on with phone + OTP; works with scheduled tasks for daily digests, drip campaigns, and auto-forwards.', + mcp_config: { + type: 'stdio', + command: 'python', + args: ['-m', 'backend.apps.telegram_mcp'], + }, + color: '#229ED9', + website: 'https://core.telegram.org/api', + connectLabel: 'Connect Telegram', + connectInstructions: 'Enter your Telegram phone (E.164 format, e.g. +15551234567). Telegram will send a 5-digit code to your existing Telegram app — you\'ll be prompted for it next. If you have cloud 2FA enabled, you\'ll be asked for the password after the code. Session is saved at ~/.telegram_mcp/sessions/ and reused on every spawn.', + credentialFields: [ + { key: 'TELEGRAM_PHONE', label: 'Phone (E.164)', placeholder: '+15551234567', type: 'text' }, + ], + icon: ( + + + + + ), + }, { id: 'linkedin', name: 'LinkedIn', @@ -662,7 +686,7 @@ const Tools: React.FC = () => { // browse the long tail. const CURATED_MCP_NAMES = useMemo(() => new Set([ 'google-workspace', 'microsoft-365', 'slack', 'discord', - 'notion', 'airtable', 'hubspot', 'reddit', 'youtube', 'instagram', 'linkedin', 'github', + 'notion', 'airtable', 'hubspot', 'reddit', 'youtube', 'instagram', 'linkedin', 'github', 'telegram', ]), []); const regServers = useMemo(() => { if (regSource !== 'curated') return regServersRaw; @@ -710,6 +734,15 @@ const Tools: React.FC = () => { /** LinkedIn desktop CLI (Electron); null when idle */ const [linkedinConnectBusy, setLinkedinConnectBusy] = useState(null); + /** Telegram multi-step OTP flow (phone -> code -> optional 2FA password). */ + const [telegramFlow, setTelegramFlow] = useState<{ + toolId: string; + phone: string; + step: 'code' | 'password'; + busy: boolean; + } | null>(null); + const [telegramInput, setTelegramInput] = useState(''); + // Full-auth upgrade dialog (password + 2FA) is disabled while we figure out // Instagram's trusted-notification polling endpoint. Connect uses browser-only // sign-in, which unlocks ~8 read tools. See handleInstagramConnect below. @@ -1441,6 +1474,39 @@ const Tools: React.FC = () => { dispatch(discoverTools(credDialogToolId)); return; } + + // Telegram: phone -> /start -> wait for OTP -> /verify -> (optional 2FA) -> /password. + // We keep the credentials dialog open and morph its credentialFields between + // steps so the user never sees a fresh modal between steps. Backend caches + // the Telethon client between calls via _TG_PENDING. + if (credDialogIntegration.id === 'telegram') { + const phone = (credDialogValues['TELEGRAM_PHONE'] || '').trim(); + if (!phone.startsWith('+')) { + setSnackbar({ open: true, message: 'Phone must be E.164 (start with + and country code, e.g. +15551234567)', severity: 'error' }); + return; + } + const startResp = await fetch(`${API_BASE}/tools/credentials/telegram/start`, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ tool_id: credDialogToolId, phone }), + }); + const startData = await startResp.json().catch(() => ({ ok: false, error: `bad response (${startResp.status})` })); + if (!startData.ok) { + setSnackbar({ open: true, message: `Telegram sign-in failed: ${startData.error || 'unknown error'}`, severity: 'error' }); + return; + } + if (!startData.needs_code) { + // Existing valid session was reused. + await dispatch(fetchToolStatus(credDialogToolId)); + setCredDialogOpen(false); + setSnackbar({ open: true, message: 'Telegram connected (existing session)! Re-discovering actions…' }); + dispatch(discoverTools(credDialogToolId)); + return; + } + setTelegramFlow({ toolId: credDialogToolId, phone, step: 'code', busy: false }); + setSnackbar({ open: true, message: 'Telegram sent you a code. Enter it below.' }); + return; + } const result = await dispatch(updateTool({ id: credDialogToolId, credentials: credDialogValues, @@ -1490,6 +1556,75 @@ const Tools: React.FC = () => { } }; + // Telegram OTP step handlers. Each posts to the matching backend endpoint + // and either advances the flow (code → password if 2FA), completes it, or + // surfaces an error inline without closing the dialog. + const handleTelegramVerify = async () => { + if (!telegramFlow) return; + const code = telegramInput.trim(); + if (!code) return; + setTelegramFlow({ ...telegramFlow, busy: true }); + try { + const resp = await fetch(`${API_BASE}/tools/credentials/telegram/verify`, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ tool_id: telegramFlow.toolId, code }), + }); + const data = await resp.json().catch(() => ({ ok: false, error: `bad response (${resp.status})` })); + if (data.needs_password) { + setTelegramInput(''); + setTelegramFlow({ ...telegramFlow, step: 'password', busy: false }); + setSnackbar({ open: true, message: 'Cloud 2FA password required. Enter it below.' }); + return; + } + if (!data.ok) { + setTelegramFlow({ ...telegramFlow, busy: false }); + setSnackbar({ open: true, message: `Telegram verify failed: ${data.error || 'unknown error'}`, severity: 'error' }); + return; + } + await dispatch(fetchToolStatus(telegramFlow.toolId)); + setTelegramFlow(null); + setTelegramInput(''); + setCredDialogOpen(false); + setSnackbar({ open: true, message: 'Telegram connected! Re-discovering actions…' }); + dispatch(discoverTools(telegramFlow.toolId)); + } catch (err: unknown) { + setTelegramFlow({ ...telegramFlow, busy: false }); + const msg = err instanceof Error ? err.message : String(err); + setSnackbar({ open: true, message: msg, severity: 'error' }); + } + }; + + const handleTelegramPassword = async () => { + if (!telegramFlow) return; + const password = telegramInput; + if (!password) return; + setTelegramFlow({ ...telegramFlow, busy: true }); + try { + const resp = await fetch(`${API_BASE}/tools/credentials/telegram/password`, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ tool_id: telegramFlow.toolId, password }), + }); + const data = await resp.json().catch(() => ({ ok: false, error: `bad response (${resp.status})` })); + if (!data.ok) { + setTelegramFlow({ ...telegramFlow, busy: false }); + setSnackbar({ open: true, message: `Telegram 2FA failed: ${data.error || 'unknown error'}`, severity: 'error' }); + return; + } + await dispatch(fetchToolStatus(telegramFlow.toolId)); + setTelegramFlow(null); + setTelegramInput(''); + setCredDialogOpen(false); + setSnackbar({ open: true, message: 'Telegram connected! Re-discovering actions…' }); + dispatch(discoverTools(telegramFlow.toolId)); + } catch (err: unknown) { + setTelegramFlow({ ...telegramFlow, busy: false }); + const msg = err instanceof Error ? err.message : String(err); + setSnackbar({ open: true, message: msg, severity: 'error' }); + } + }; + const handleDisconnectIntegration = async (toolId: string, integration: Integration) => { if (integration.authType === 'oauth2') { // Revoke the token on Google's side (fire-and-forget) @@ -3009,6 +3144,33 @@ const Tools: React.FC = () => { Click Sign in with Slack below — a Slack window will open. Sign in normally and the window will close automatically once you reach your workspace. + ) : credDialogIntegration?.id === 'telegram' && telegramFlow ? ( + <> + + {telegramFlow.step === 'code' + ? `Telegram sent a 5-digit code to ${telegramFlow.phone}. Check your existing Telegram app and paste it below.` + : `Your Telegram account has cloud 2FA enabled. Enter the password you set up to complete sign-in.`} + + setTelegramInput(e.target.value)} + onKeyDown={(e) => { + if (e.key === 'Enter' && telegramInput.trim() && !telegramFlow.busy) { + e.preventDefault(); + if (telegramFlow.step === 'code') void handleTelegramVerify(); + else void handleTelegramPassword(); + } + }} + fullWidth + size="small" + disabled={telegramFlow.busy} + sx={{ '& .MuiOutlinedInput-root': { bgcolor: c.bg.page, fontFamily: c.font.mono, fontSize: '0.85rem' } }} + /> + ) : ( <> {credDialogIntegration?.connectInstructions && ( @@ -3034,15 +3196,38 @@ const Tools: React.FC = () => { )} - +