MCP Telegram Bot Server

Этот проект реализует Model Context Protocol (MCP) сервер, предоставляющий LLM доступ к Telegram Bot API. Реализован на Python 3.12+ с FastMCP, aiosqlite и async-клиентом httpx.

Позволяет LLM (например, Claude Desktop или через Inspector) читать недавние сообщения и отвечать пользователям в Telegram напрямую через MCP-тулзы.

🧠 Ключевая архитектурная идея

Telegram Bot API не предоставляет ни произвольного чтения истории чата, ни списка чатов бота — это принципиальное ограничение платформы, а не недоработка. Поэтому сервер не «оборачивает» API один-в-один, а строит собственное персистентное состояние из входящих getUpdates: фоновый listener непрерывно накапливает сообщения в локальную БД, а MCP-тулзы выполняют discovery и чтение поверх этого хранилища.

Это осознанный ответ на constraints платформы: getUpdates даёт только поток новых апдейтов — значит, «память» и «поиск чатов» нужно реализовать на своей стороне.

Related MCP server: telegram-mcp

🚀 Установка и запуск

1. Python 3.12+ и Node.js.

2. Клонируйте репозиторий и создайте venv:

   python -m venv venv
   .\venv\Scripts\activate  # Windows (Linux: source venv/bin/activate)

3. Зависимости:

   pip install -r requirements.txt

4. Конфигурация:

   • cp .env.example .env

   • впишите токен бота (см. «Переменные окружения»).

5. Запуск через Inspector (интерактивное тестирование в браузере):

   npx @modelcontextprotocol/inspector .\venv\Scripts\python src/main.py

⚙️ Переменные окружения (.env)

Токен вынесен в env, а не в код — сервер запускается с чужим токеном «из коробки», без правок исходников.

🛠️ Доступные MCP-тулзы

Сервер предоставляет следующие инструменты для LLM:

• get_known_chats() — Возвращает список всех чатов (и их ID), которые когда-либо писали боту с момента его первого запуска.

• get_recent_messages(chat_id, limit) — Читает историю сообщений конкретного чата из локальной БД (в хронологическом порядке).

• send_message(chat_id, text) — Отправляет сообщение в чат. Требует ручного подтверждения (Human-in-the-Loop) через интерфейс, если не включен AUTO_APPROVE.

• get_chat_info(chat_id) — Запрашивает детальную информацию о чате (тип, название, описание, количество участников) напрямую через API Telegram.

🏛️ Архитектурные решения

• SQLite + WAL, а не PostgreSQL/Redis. Задача — локальное персистентное состояние одного процесса. SQLite даёт персистентность между рестартами без внешних зависимостей, а режим WAL разводит параллельные чтения (MCP-тулзы) и записи (listener) без блокировок. Внешняя СУБД здесь решала бы несуществующую проблему.

• Составной ключ PRIMARY KEY (chat_id, message_id). Telegram message_id уникален только внутри чата, а не глобально. Одиночный PK по message_id привёл бы к коллизиям между разными чатами; составной ключ заодно бесплатно обеспечивает идемпотентность вставок.

• Отдельная таблица chats, а не SELECT DISTINCT по сообщениям. Имя чата меняется со временем; хранение снимка в каждом сообщении давало бы дубли в discovery. UPSERT в справочник держит актуальное имя и делает get_known_chats дешёвым независимо от объёма истории.

• edited_message обновляет текст, но НЕ time. Время редактирования приходит в edit_date. Если писать его в поле сортировки, отредактированное старое сообщение «всплыло» бы в конец истории (ORDER BY time DESC). Поэтому редактирование меняет только текст — хронология сохраняется. Тонкое место, которое легко упустить.

• Plain text вывод истории, а не JSON. Экономит токены (нет служебных кавычек/ключей), формат «Имя (ID): текст» модели парсят нативно (обучены на логах диалогов), и нет риска сломать вывод на неэкранированных смайлах/переносах строк из реальных сообщений. ID в скобках снимает неоднозначность одинаковых имён.

• Разделение восстановления: listener + watchdog. Listener сам переживает сетевые сбои через try/except + экспоненциальный backoff. Watchdog отвечает только за «смерть самой корутины» и воскрешает её — но не трогает listener, отменённый штатно при shutdown. Чёткое разделение ответственности вместо одной перегруженной retry-обёртки.

• get_chat_info — живой запрос, а не БД. Информация о конкретном чате берётся напрямую через getChat/getChatMemberCount, поэтому тул работает по любому chat_id, даже если чата ещё нет в локальной памяти. Локальное хранилище используется только там, где Bot API бессилен — история и discovery.

• Human-in-the-Loop для send_message (+ AUTO_APPROVE). Отправка — действие с побочным эффектом на реальных людей, поэтому по умолчанию требует подтверждения. Флаг AUTO_APPROVE — осознанный tradeoff для автопрогонов/демо.

• Что намеренно НЕ добавлено: SQLAlchemy, Redis, DI-фреймворк, Celery, message broker, Docker. Для задачи такого масштаба это дало бы больше кода, чем инженерной ценности. Стек сознательно минимален: FastMCP + httpx + aiosqlite + asyncio.

⚠️ Важные настройки в Telegram (ОБЯЗАТЕЛЬНО)

Чтобы бот работал в группах, необходимо отключить Privacy Mode — иначе бот видит только сообщения со слеша (/) и не соберёт нормальную историю.

1. Откройте @BotFather.

2. /setprivacy → выберите бота → Disable.

3. (Опционально, надёжнее) сделайте бота администратором группы — тогда он видит 100% сообщений.

🛑 Ограничения (следствия платформы)

• Single-Bot. Используется Long Polling (getUpdates). Нельзя запускать несколько инстансов или совмещать с webhook — Telegram вернёт 409 Conflict (у бота может быть только один потребитель апдейтов). На старте сервер вызывает deleteWebhook.

• Нет истории до запуска. getUpdates отдаёт только новые апдейты (и хранит недоставленные ~24 ч). Сообщения до первого запуска бот не увидит.

• Discovery только по активным чатам. get_known_chats вернёт чат, только если в нём была активность после запуска бота.

🛡️ Гарантии доставки

• At-least-once. Трекинг update_id в таблице state (offset): после падения/рестарта сбор продолжается с последней подтверждённой точки, сообщения не теряются.

• Идемпотентное хранение. ON CONFLICT DO NOTHING гарантирует, что повторно доставленный апдейт (сетевой сбой, рестарт до сохранения offset) не создаёт дубль в БД.

Формулировка намеренно точная: доставка Telegram — at-least-once, а exactly-once достигается не на уровне доставки, а за счёт идемпотентного хранения.