Telegram MCP Server

MCP сервер для интеграции AI-агентов с Telegram. Позволяет отправлять сообщения пользователю и получать ответы в синхронном режиме.

Особенности

• Интеграция с AI-агентами через протокол MCP
• Отправка сообщений в Telegram
• Ожидание ответа пользователя с настраиваемым таймаутом
• Синхронный режим работы (агент ждет ответа пользователя)
• Безопасное хранение токенов через переменные окружения
• Готовый Docker-контейнер для простого развертывания

Быстрый старт

Предварительные требования

1. Создайте Telegram бота через @BotFather и получите токен
2. Получите ID вашего чата через @userinfobot
3. Установите Docker

Варианты использования

Вариант 1: Использование в сервисах с поддержкой MCP (Cursor, WindSurf, Anthropic, VSCode с расширениями)

1. Добавьте следующую конфигурацию в ваш файл mcp_config.json:

{
  "mcpServers": {
    "telegram-docker": {
      "command": "docker",
      "args": [
        "run",
        "-i",
        "--rm",
        "-e", "TELEGRAM_BOT_TOKEN=your_bot_token_here",
        "-e", "TELEGRAM_CHAT_ID=your_chat_id_here",
        "bmwdroch/telegram-mcp-server:latest"
      ]
    }
  }
}

2. Замените your_bot_token_here и your_chat_id_here на ваши значения
3. Перезапустите IDE для применения изменений
4. Теперь вы можете использовать инструмент sendMessage в ваших AI-агентах

Пример использования инструмента в AI-агенте

# Пример вызова инструмента в Cursor, WindSurf, Anthropic или других агентах с поддержкой MCP
result = await mcp4_sendMessage(
    agentMessage="Привет! Это тестовое сообщение от AI-агента.",
    parseMode="HTML",
    timeout=3600  # Таймаут ожидания ответа в секундах (1 час)
)

# Обработка ответа пользователя
user_response = result.get("userResponse")
print(f"Пользователь ответил: {user_response}")

Вариант 2: Запуск контейнера напрямую

# Создайте файл .env с вашими данными
echo "TELEGRAM_BOT_TOKEN=your_bot_token_here" > .env
echo "TELEGRAM_CHAT_ID=your_chat_id_here" >> .env

# Запустите контейнер
docker run -i --rm --env-file .env bmwdroch/telegram-mcp-server:latest

Вариант 2: Сборка из исходного кода

# Клонируйте репозиторий
git clone https://github.com/username/telegram-mcp-server.git
cd telegram-mcp-server

# Создайте файл .env с вашими данными
echo "TELEGRAM_BOT_TOKEN=your_bot_token_here" > .env
echo "TELEGRAM_CHAT_ID=your_chat_id_here" >> .env

# Соберите и запустите через Docker Compose
docker-compose up --build

Подключение к Cursor/Cline

Добавьте в файл конфигурации MCP серверов (~/.codeium/windsurf/mcp_config.json):

{
  "mcpServers": {
    "telegram": {
      "command": "docker",
      "args": [
        "run",
        "-i",
        "--rm",
        "--env-file",
        "/path/to/.env",
        "username/telegram-mcp-server:latest"
      ]
    }
  }
}

Или с прямым указанием переменных окружения:

{
  "mcpServers": {
    "telegram": {
      "command": "docker",
      "args": [
        "run",
        "-i",
        "--rm",
        "username/telegram-mcp-server:latest"
      ],
      "env": {
        "TELEGRAM_BOT_TOKEN": "your_bot_token_here",
        "TELEGRAM_CHAT_ID": "your_chat_id_here"
      }
    }
  }
}

Доступные инструменты

sendMessage

Отправляет сообщение пользователю в Telegram и ожидает ответа.

Параметры:

• agentMessage (обязательный) - сообщение от агента для пользователя
• timeout (опциональный) - таймаут ожидания ответа в секундах (от 600 до 3600 секунд)
• parseMode (опциональный) - режим форматирования сообщения (HTML, Markdown, или null)

Возвращает: JSON-объект со следующими полями:

• status: "success" или "timeout" или "error"
• userResponse: текст ответа пользователя (при status="success")
• error: описание ошибки (при status="error")
• timestamp: время выполнения запроса

Примеры использования

В Python

import json

# В коде агента
response = await call_tool("sendMessage", {
    "agentMessage": "Нужно ваше решение: продолжить обработку файла или остановиться?",
    "timeout": 600,  # 10 минут
    "parseMode": "Markdown"
})

user_decision = json.loads(response)
if user_decision["status"] == "success":
    print(f"Пользователь ответил: {user_decision['userResponse']}")
elif user_decision["status"] == "timeout":
    print("Пользователь не ответил вовремя")
else:
    print(f"Ошибка: {user_decision['error']}")