diff --git a/backend/apps/telegram_bot/__init__.py b/backend/apps/telegram_bot/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/backend/apps/telegram_bot/listener.py b/backend/apps/telegram_bot/listener.py new file mode 100644 index 00000000..45e3f7d1 --- /dev/null +++ b/backend/apps/telegram_bot/listener.py @@ -0,0 +1,314 @@ +"""Telegram-as-control-plane listener. + +Sister of backend.apps.telegram_mcp. Where the MCP server is the +agent's *outbound* channel to Telegram, this listener is the +*inbound* channel: you DM yourself in Saved Messages with a command +prefix (/task ...), the backend dispatches it to an agent, and the +final response comes back to the same chat. + +Design tenets: + - Only `from_users='me'` in Saved Messages is honored. Nobody else + can drive your agent, not even someone with your phone in their + contacts. + - The listener uses a *copy* of the Telethon session + (.listener.session) so it can stay connected without + fighting the MCP server's SQLite lock when the agent calls a + Telegram tool mid-task. + - Lifespan is a SubApp so the listener starts with the backend and + shuts down cleanly on Ctrl+C. If Telegram isn't connected, the + listener idles instead of erroring. + - Long-lived dispatch is poll-based (no WS subscription yet); we + just check session.status every 2s up to a 30-minute cap. Good + enough for the MVP. +""" +from __future__ import annotations + +import asyncio +import logging +import os +import re +import shutil +from contextlib import asynccontextmanager +from pathlib import Path +from typing import Optional + +from telethon import TelegramClient, events + +from backend.config.Apps import SubApp + +logger = logging.getLogger(__name__) + +_SESSION_DIR = Path.home() / ".telegram_mcp" / "sessions" +_MAX_REPLY_CHARS = 3500 +_POLL_INTERVAL_S = 2.0 +_TASK_TIMEOUT_S = 30 * 60 + +_listener_task: Optional[asyncio.Task] = None +_client: Optional[TelegramClient] = None + + +def _connected_phone() -> Optional[str]: + """Read which Telegram is currently connected, from the tool config.""" + try: + from backend.apps.tools_lib.tools_lib import _load_all + for tool in _load_all(): + if (tool.name or "").lower() == "telegram" and tool.auth_status == "connected": + phone = (tool.credentials or {}).get("TELEGRAM_PHONE", "").strip() + if phone: + return phone + except Exception as exc: # noqa: BLE001 + logger.warning(f"telegram-bot: could not read tool config: {exc}") + return None + + +def _api_creds() -> tuple[Optional[int], Optional[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: + return (None, None) + try: + return (int(api_id_raw), api_hash) + except ValueError: + return (None, None) + + +def _digits(phone: str) -> str: + return re.sub(r"\D", "", phone or "") + + +def _ensure_listener_session(phone: str) -> Path: + """Copy the main MCP session file to a side .listener.session path so the + two processes don't fight over the SQLite lock. Re-copy if main is newer.""" + _SESSION_DIR.mkdir(parents=True, exist_ok=True) + main = _SESSION_DIR / f"{_digits(phone)}.session" + side = _SESSION_DIR / f"{_digits(phone)}.listener.session" + if main.exists() and (not side.exists() or main.stat().st_mtime > side.stat().st_mtime + 1): + try: + shutil.copy2(main, side) + logger.info(f"telegram-bot: copied main session to {side.name}") + except Exception as exc: # noqa: BLE001 + logger.warning(f"telegram-bot: session copy failed: {exc}") + # Telethon takes the path without .session suffix. + return side.with_suffix("") + + +HELP_TEXT = ( + "*OpenSwarm Telegram bot*\n\n" + "Commands:\n" + " `/task ` — run an agent task\n" + " `/status` — list running sessions\n" + " `/help` — this message\n\n" + "Messages without a `/` prefix are ignored, so you can keep using " + "Saved Messages normally." +) + + +async def _route(event, text: str) -> None: + """Dispatch a recognized command.""" + if text.startswith("/help"): + await event.respond(HELP_TEXT) + return + if text.startswith("/status"): + await _handle_status(event) + return + if text.startswith("/task "): + prompt = text[len("/task "):].strip() + if not prompt: + await event.respond("Usage: `/task `") + return + await _handle_task(event, prompt) + return + if text.startswith("/"): + await event.respond(f"Unknown command. /help for the list.") + + +async def _handle_status(event) -> None: + try: + from backend.apps.agents.agent_manager import agent_manager + sessions = [s for s in agent_manager.get_all_sessions() if s.status == "running"] + except Exception as exc: # noqa: BLE001 + await event.respond(f"Could not read sessions: {exc}") + return + if not sessions: + await event.respond("No running sessions.") + return + lines = [f"• `{s.id[:8]}` — {s.name or '(no name)'}" for s in sessions[:10]] + await event.respond("Running:\n" + "\n".join(lines)) + + +async def _handle_task(event, prompt: str) -> None: + """Spawn an agent, poll until it stops, send the final assistant reply.""" + from backend.apps.agents.agent_manager import agent_manager + from backend.apps.agents.models import AgentConfig + + config = AgentConfig( + name=f"telegram: {prompt[:48]}", + mode="agent", + # Default allowed_tools (Read/Edit/Write/Bash/Glob/Grep/AskUserQuestion) + # plus the connected MCPs so the agent can use Telegram, Instagram, + # LinkedIn, GitHub as needed when invoked from this entry point. + allowed_tools=[ + "Read", "Edit", "Write", "Bash", "Glob", "Grep", "AskUserQuestion", + "mcp:Telegram", "mcp:Instagram", "mcp:LinkedIn", "mcp:GitHub", + ], + ) + try: + session = await agent_manager.launch_agent(config) + except Exception as exc: # noqa: BLE001 + await event.respond(f"Could not launch agent: {exc}") + return + + await event.respond(f"⏳ Starting task — session `{session.id[:8]}`") + + try: + await agent_manager.send_message(session.id, prompt) + except Exception as exc: # noqa: BLE001 + await event.respond(f"Send failed: {exc}") + return + + # Poll until the session reports a terminal status, with a hard cap. + deadline = asyncio.get_event_loop().time() + _TASK_TIMEOUT_S + final = None + while asyncio.get_event_loop().time() < deadline: + await asyncio.sleep(_POLL_INTERVAL_S) + s = agent_manager.get_session(session.id) + if s is None: + break + if s.status in ("completed", "stopped", "error"): + final = s + break + + if final is None: + await event.respond( + f"Task `{session.id[:8]}` is still running after {_TASK_TIMEOUT_S // 60}m. " + f"Check OpenSwarm UI for progress." + ) + return + + reply = _extract_last_assistant_text(final.id) or f"Task `{session.id[:8]}` finished with no text reply." + if final.status == "error": + reply = f"❌ Task errored.\n\n{reply}" + await event.respond(reply[:_MAX_REPLY_CHARS]) + + +def _extract_last_assistant_text(session_id: str) -> str: + """Pull the last assistant text from the persisted session transcript. + + Sessions are stored at backend/data/sessions/.json. Each line of the + transcript is a Message; we want the last role=assistant message whose + content is a plain string or has a text-content block. + """ + try: + from backend.config.paths import SESSIONS_DIR + import json + path = Path(SESSIONS_DIR) / f"{session_id}.json" + if not path.exists(): + return "" + with open(path) as f: + data = json.load(f) + messages = data.get("messages") or data.get("transcript") or [] + for msg in reversed(messages): + if msg.get("role") != "assistant": + continue + content = msg.get("content") + if isinstance(content, str) and content.strip(): + return content.strip() + if isinstance(content, list): + texts = [b.get("text", "") for b in content if isinstance(b, dict) and b.get("type") == "text"] + joined = "\n".join(t for t in texts if t).strip() + if joined: + return joined + return "" + except Exception as exc: # noqa: BLE001 + logger.warning(f"telegram-bot: could not extract assistant text: {exc}") + return "" + + +async def _listener_main() -> None: + """Run-forever loop. Tolerates the no-Telegram-connected case by idling.""" + global _client + + phone = _connected_phone() + if not phone: + logger.info("telegram-bot: no connected Telegram — listener idle.") + return + + api_id, api_hash = _api_creds() + if api_id is None: + logger.warning( + "telegram-bot: OPENSWARM_TELEGRAM_API_ID/_API_HASH not set in backend env — listener idle." + ) + return + + session_base = _ensure_listener_session(phone) + _client = TelegramClient(str(session_base), api_id, api_hash) + try: + await _client.connect() + except Exception as exc: # noqa: BLE001 + logger.warning(f"telegram-bot: connect failed, listener idle: {exc}") + return + + if not await _client.is_user_authorized(): + logger.warning( + "telegram-bot: listener session not authorized — disconnect and reconnect Telegram in OpenSwarm. Idling." + ) + try: await _client.disconnect() + except Exception: pass + return + + me = await _client.get_me() + my_id = me.id + logger.info( + f"telegram-bot: listener active for @{me.username or me.phone} (id={my_id}). " + f"DM yourself in Saved Messages with /help to start." + ) + + @_client.on(events.NewMessage(from_users="me")) + async def _on_message(event): + # Hard authorization: only your own Saved Messages chat. event.chat_id + # for Saved Messages equals your own user id, and from_users='me' is + # already enforced by the decorator. Defense in depth here. + if event.chat_id != my_id: + return + text = (event.message.message or "").strip() + if not text or not text.startswith("/"): + return + try: + await _route(event, text) + except Exception as exc: # noqa: BLE001 + logger.exception(f"telegram-bot: route failed: {exc}") + try: await event.respond(f"Listener error: {exc}") + except Exception: pass + + try: + await _client.run_until_disconnected() + except asyncio.CancelledError: + raise + except Exception as exc: # noqa: BLE001 + logger.exception(f"telegram-bot: listener crashed: {exc}") + + +@asynccontextmanager +async def telegram_bot_lifespan(): + """Boot the listener when the backend starts; tear it down on shutdown.""" + global _listener_task + _listener_task = asyncio.create_task(_listener_main(), name="telegram-bot-listener") + try: + yield + finally: + if _listener_task and not _listener_task.done(): + _listener_task.cancel() + try: + await _listener_task + except asyncio.CancelledError: + pass + except Exception as exc: # noqa: BLE001 + logger.warning(f"telegram-bot: listener task error during shutdown: {exc}") + if _client is not None: + try: + await _client.disconnect() + except Exception as exc: # noqa: BLE001 + logger.warning(f"telegram-bot: client disconnect error: {exc}") + + +telegram_bot = SubApp("telegram-bot", telegram_bot_lifespan) diff --git a/backend/apps/tools_lib/tools_lib.py b/backend/apps/tools_lib/tools_lib.py index b1afc470..96c318fa 100644 --- a/backend/apps/tools_lib/tools_lib.py +++ b/backend/apps/tools_lib/tools_lib.py @@ -1520,6 +1520,21 @@ async def _tg_finalize_after_auth(tool, phone: str, client) -> None: pass digits = _tg_sanitize_phone(phone) + + # Snapshot the freshly-authenticated SQLite session to a side path used + # by the long-running bot listener (backend.apps.telegram_bot). Without + # this copy, listener + MCP-server would fight over the same file lock + # whenever an agent invokes a Telegram tool mid-listener-session. + try: + import shutil as _sh + session_dir = _tg_session_dir() + main_session = session_dir / f"{digits}.session" + listener_session = session_dir / f"{digits}.listener.session" + if main_session.exists(): + _sh.copy2(main_session, listener_session) + except Exception as exc: # noqa: BLE001 + logger.warning(f"telegram: could not snapshot listener session: {exc}") + if me and me.username: tool.connected_account_email = f"@{me.username}" elif me and (me.first_name or me.last_name): diff --git a/backend/main.py b/backend/main.py index 1341252b..822e0fbd 100644 --- a/backend/main.py +++ b/backend/main.py @@ -44,11 +44,12 @@ from backend.apps.subscription.router import subscription from backend.apps.auth.router import auth from backend.apps.web.web import web from backend.apps.agents.anthropic_proxy import anthropic_proxy +from backend.apps.telegram_bot.listener import telegram_bot from fastapi.middleware.cors import CORSMiddleware from fastapi import WebSocket, WebSocketDisconnect import json -main_app = MainApp([health, agents, skills, tools_lib, modes, settings, mcp_registry, skill_registry, outputs, dashboards, service, subscription, auth, web, anthropic_proxy]) +main_app = MainApp([health, agents, skills, tools_lib, modes, settings, mcp_registry, skill_registry, outputs, dashboards, service, subscription, auth, web, anthropic_proxy, telegram_bot]) app = main_app.app # Generate per-install auth token BEFORE we bind the HTTP port. By the