Управляю VDS с телефона: Telegram-бот + Claude Code CLI

Я не devops, поэтому хотел получать ответы на человеческом языке в любое время. Ты в дороге, приходит алерт, нужно срочно посмотреть логи или проверить статус сервиса. Достаёшь телефон, открываешь SSH-клиент, набираешь команды…

В итоге, я написал Telegram-бота, который принимает запросы на человеческом языке и выполняет их через Claude Code CLI. Теперь вместо journalctl -u nginx --since "1 hour ago" | grep error я просто пишу в Telegram: «Покажи ошибки ^[1] nginx за последний час». Выложил в opensource.

В статье расскажу про архитектуру и примеры.

Claude Code CLI

Консольный инструмент от Anthropic, который даёт Claude доступ к файловой системе и терминалу. AI-агент, который может читать файлы, выполнять bash-команды и анализировать результаты.Внутри него можно создавать еще агентов, можно помещать его работать в конкретный проект…

Я использую подписку Claude Max ($200/месяц), в которую входит почти безлимитный доступ к Claude Code CLI. Щедро, в отличии от OpenAI. Токены не считаю, подписки хватает, поэтому извините – в комментариях не отвечу на этот вопрос.

Архитектура

┌─────────────┐     ┌─────────────────┐     ┌─────────────────┐
│  Telegram   │────▶│   Python Bot    │────▶│ Claude Code CLI │
│   Client    │◀────│  (handlers +    │◀────│   (subprocess)  │
└─────────────┘     │   services)     │     └─────────────────┘
                    └────────┬────────┘              │
                             │                       ▼
                    ┌────────▼────────┐     ┌───────────────┐
                    │   PostgreSQL    │     │    Server     │
                    │   (sessions)    │     │  (filesystem) │
                    └─────────────────┘     └───────────────┘

Бот состоит из нескольких слоёв:

bot/
├── main.py              # Точка входа, инициализация
├── config.py            # Конфигурация из ENV
├── handlers/
│   ├── commands.py      # /start, /reset, /status, /cancel
│   ├── messages.py      # Обработка текстовых сообщений
│   └── files.py         # Загрузка и анализ файлов
├── services/
│   ├── claude.py        # Вызов Claude CLI
│   ├── session.py       # Управление сессиями
│   └── formatter.py     # Форматирование для Telegram
└── database/
    └── pool.py          # Connection pool PostgreSQL

Разделение на handlers и services. Handlers знают про Telegram (Update, Context), services нет. Это позволяет тестировать логику ^[2] отдельно от Telegram API и переиспользовать services, если понадобится другой интерфейс.

Вызов Claude CLI: subprocess + stdin

Важный для продакшен сервера был вопрос – как безопасно передать пользовательский ввод в Claude CLI.

Плохо shell=True:

# ОПАСНО! Не делайте так
subprocess.run(f'claude "{user_message}"', shell=True)

Если user_message содержит "; rm -rf / #, может пойти что-то не так.

Subprocess.PIPE + stdin:

process = await asyncio.create_subprocess_exec(
    CLAUDE_CLI_PATH,
    stdin=asyncio.subprocess.PIPE,
    stdout=asyncio.subprocess.PIPE,
    stderr=asyncio.subprocess.PIPE,
    cwd=working_dir,
)

stdout, stderr = await process.communicate(
    input=full_prompt.encode('utf-8')
)

Здесь create_subprocess_exec запускает процесс напрямую, без shell. Промпт передаётся через stdin как данные, а не как часть команды. Даже если в сообщении будут спецсимволы shel, они не интерпретируются.

Дополнительно использую asyncio. Бот не блокируется, пока Claude что-то делает. Можно обрабатывать команды от нескольких пользователей параллельно (если настроить whitelist на несколько ID – про это дальше будет).

Safety Prompt: как ограничить от безумия

Claude Code по умолчанию может делать что угодно: редактировать файлы, запускать команды, устанавливать пакеты.

Safety instructions, которые добавляются к каждому запросу:

safety_instructions = """IMPORTANT CONTEXT: You are being accessed through Telegram bot.
The user is writing from their phone via Telegram messenger.
Responses will be shown in Telegram chat, so keep them concise.

CRITICAL RULES - YOU MUST FOLLOW:
1. DO NOT execute ANY system commands unless the message explicitly contains 
   trigger words: "выполни", "сделай", "запусти", "исправь", "создай", 
   "удали", "restart", "перезапусти"
2. If user just asks about status - ONLY provide information, don't modify
3. NEVER run systemctl, apt, rm, or modifying commands without explicit request
4. Default mode is READ-ONLY - only analyze and inform
"""

if needs_execution:
    safety_instructions += "CURRENT MODE: Execution allowedn"
else:
    safety_instructions += "CURRENT MODE: Information only (no execution)n"

Флаг needs_execution определяется простой проверкой:

execute_keywords = [
    "выполни", "сделай", "запусти", "исправь", 
    "создай", "удали", "restart", "перезапусти"
]
needs_execution = any(kw in user_message.lower() for kw in execute_keywords)

Если ключевых слов нет, к сообщению добавляется префикс:

if not needs_execution:
    user_message = f"[ТОЛЬКО ИНФОРМАЦИЯ, НЕ ВЫПОЛНЯТЬ КОМАНДЫ] {user_message}"

Это не 100% защита. Claude может ошибиться (но пока не было). На практике работает хорошо. Хотя иногда переживаю.

Управление процессами

Claude может думать долго, особенно если анализирует большие логи. Нужны таймауты и возможность отмены.

Таймаут:

try:
    stdout, stderr = await asyncio.wait_for(
        process.communicate(input=prompt.encode('utf-8')),
        timeout=CLAUDE_TIMEOUT  # 300 секунд по умолчанию
    )
except asyncio.TimeoutError:
    process.terminate()
    try:
        await asyncio.wait_for(process.wait(), timeout=5)
    except asyncio.TimeoutError:
        process.kill()  # SIGKILL если не завершился по SIGTERM

Отмена по команде /cancel:

Активные процессы хранятся в словаре по user_id:

_active_processes: Dict[int, asyncio.subprocess.Process] = {}

При запуске Claude процесс регистрируется:

_active_processes[user_id] = process

Команда /cancel находит и завершает процесс:

async def cancel_process(user_id: int) -> bool:
    process = _active_processes.get(user_id)
    if process and process.returncode is None:
        process.terminate()
        try:
            await asyncio.wait_for(process.wait(), timeout=5)
        except asyncio.TimeoutError:
            process.kill()
        return True
    return False

Graceful shutdown:

При остановке бота (SIGTERM/SIGINT) нужно корректно завершить все процессы:

async def terminate_all_processes():
    for user_id, process in list(_active_processes.items()):
        if process.returncode is None:
            process.terminate()
            try:
                await asyncio.wait_for(process.wait(), timeout=5)
            except asyncio.TimeoutError:
                process.kill()
    _active_processes.clear()

Это вызывается в post_shutdown хуке python-telegram-bot.

Сессии и контекст

Claude должен помнить контекст разговора. Если я спросил про nginx, а потом написал «А что с ошибками?» он должен понять, что речь про nginx.

Структура сессии:

@dataclass
class Session:
    user_id: int
    context: List[Dict]        # История сообщений
    working_dir: str           # Текущая директория
    message_count: int         # Счётчик сообщений
    created_at: datetime
    last_activity: datetime

Хранение:

Двухуровневое: in-memory cache + PostgreSQL.

class SessionManager:
    def __init__(self):
        self._cache: Dict[int, Session] = {}
    
    def get_session(self, user_id: int) -> Session:
        # Сначала проверяем кэш
        if user_id in self._cache:
            return self._cache[user_id]
        
        # Потом БД
        row = execute_one(
            "SELECT * FROM sessions WHERE user_id = %s", 
            (user_id,)
        )
        if row:
            session = Session.from_dict(dict(row))
            self._cache[user_id] = session
            return session
        
        # Создаём новую
        # ...

Кэш нужен для скорости. Не дёргать БД на каждое сообщение. PostgreSQL для персистентности между перезапусками бота.

Контекст в промпте:

При формировании запроса к Claude добавляю последние N сообщений из контекста:

def build_prompt(user_message: str, context: List[Dict]) -> str:
    context_text = ""
    if context:
        recent = context[-CLAUDE_MAX_CONTEXT_MESSAGES:]  # последние 10
        for msg in recent:
            context_text += f"User: {msg['user']}n"
            # Обрезаем длинные ответы в контексте
            assistant_msg = msg['assistant'][:500] + "..." 
                if len(msg['assistant']) > 500 else msg['assistant']
            context_text += f"Assistant: {assistant_msg}n"
    
    return safety_instructions + context_text + f"Current request: {user_message}"

Обрезка ответов в контексте важна, иначе промпт раздувается и Claude начинает работать медленнее.

Форматирование для Telegram

Telegram поддерживает HTML-разметку. Claude возвращает Markdown. Нужен конвертер.

Проблема 1: Экранирование

В Telegram HTML символы <, >, & нужно экранировать, но только вне тегов:

def escape_html(text: str) -> str:
    return text.replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;")

Проблема 2: Code blocks

Claude возвращает код в тройных бэктиках. Telegram использует <pre> и <code>.

Решение — сначала извлечь все code blocks, заменить на плейсхолдеры, отформатировать остальной текст, потом вернуть code blocks:

def format_code_blocks(text: str) -> Tuple[str, List[str]]:
    code_blocks = []
    pattern = r"```(w*)n?(.*?)```"
    
    def replacer(match):
        lang = match.group(1) or ""
        code = match.group(2).strip()
        idx = len(code_blocks)
        code_blocks.append((lang, code))
        return f"<<<CODE_BLOCK_{idx}>>>"
    
    text = re.sub(pattern, replacer, text, flags=re.DOTALL)
    return text, code_blocks

def restore_code_blocks(text: str, code_blocks: List) -> str:
    for idx, (lang, code) in enumerate(code_blocks):
        placeholder = f"<<<CODE_BLOCK_{idx}>>>"
        formatted = f"<pre>{escape_html(code)}</pre>"
        text = text.replace(placeholder, formatted)
    return text

Проблема 3: Лимит 4096 символов

Telegram не принимает сообщения длиннее 4096 символов. Нужно разбивать, но аккуратно:

def split_message(text: str, max_length: int = 4096) -> List[str]:
    parts = []
    while len(text) > max_length:
        # Ищем хорошую точку разбиения
        split_idx = max_length
        
        # Пробуем разбить по параграфу
        para_idx = text.rfind("nn", 0, max_length)
        if para_idx > max_length // 2:
            split_idx = para_idx
        # Или по переносу строки
        elif (nl_idx := text.rfind("n", 0, max_length)) > max_length // 2:
            split_idx = nl_idx
        
        part = text[:split_idx]
        
        # Проверяем незакрытые теги
        open_tags = re.findall(r"<(b|i|code|pre)>", part)
        close_tags = re.findall(r"</(b|i|code|pre)>", part)
        
        # Закрываем незакрытые теги в конце части
        for tag in reversed(open_tags):
            if open_tags.count(tag) > close_tags.count(tag):
                part += f"</{tag}>"
        
        parts.append(part.strip())
        text = text[split_idx:].strip()
    
    if text:
        parts.append(text)
    return parts

База данных

Две таблицы:

-- Сессии пользователей
CREATE TABLE sessions (
    user_id BIGINT PRIMARY KEY,
    context JSONB DEFAULT '[]'::jsonb,
    working_dir TEXT DEFAULT '/root',
    message_count INT DEFAULT 0,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    last_activity TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

-- Логи команд (для истории и отладки)
CREATE TABLE command_logs (
    id SERIAL PRIMARY KEY,
    user_id BIGINT,
    command TEXT,
    response TEXT,
    execution_time_ms INT,
    error TEXT,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

Контекст хранится как JSONB, PostgreSQL умеет с ним работать эффективно.

Connection pool настроен через psycopg2:

_pool = pool.ThreadedConnectionPool(
    minconn=1, 
    maxconn=10, 
    **DB_CONFIG
)

Почему PostgreSQL, а не SQLite? Claude Code сам предложил при проектировании. Для однопользовательского бота SQLite хватило бы, но PostgreSQL надёжнее при конкурентных запросах и проще масштабировать если понадобится.

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

Реальные задачи, которые решаю через бота:

Мониторинг:

Анализ логов:

Работа с базой:

Ограничения

Скорость. Claude думает 3-10 секунд. Для простого ls это долго.

Стоимость. Нужна подписка Claude или оплата API.

Не для критичных операций. Ну тут сами решайте.

Не 100% защита. Safety prompt просто инструкция, а гарантия. Claude обычно следует правилам, но бывает и нет.

Как попробовать

Репозиторий: github.com/gmen1057/claude-cli-telegrambot ^[3]

Требования:

• VPS с Linux

• Python 3.9+

• PostgreSQL

• Claude Code CLI

• Telegram Bot Token

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

git clone https://github.com/gmen1057/claude-cli-telegrambot.git
cd claude-cli-telegrambot
python -m venv venv && source venv/bin/activate
pip install -r requirements.txt
cp .env.example .env  # настроить переменные
python -m bot.main

Есть Docker-вариант с docker-compose.

Итого

• asyncio.create_subprocess_exec + stdin для безопасного вызова CLI

• Safety prompt с ключевыми словами для разделения read/write операций

• Двухуровневое хранение сессий: memory cache + PostgreSQL

• Форматтер с защитой code blocks и разбиением длинных сообщений

• Таймауты и graceful shutdown для стабильности

Проект open-source, буду рад issues и PR.

Ссылки:

• Репозиторий: github.com/gmen1057/claude-cli-telegrambot ^[3]

• Claude Code: docs.anthropic.com/en/docs/claude-code ^[4]