Я не devops, поэтому хотел получать ответы на человеческом языке в любое время. Ты в дороге, приходит алерт, нужно срочно посмотреть логи или проверить статус сервиса. Достаёшь телефон, открываешь SSH-клиент, набираешь команды...
В итоге, я написал Telegram-бота, который принимает запросы на человеческом языке и выполняет их через Claude Code CLI. Теперь вместо journalctl -u nginx --since "1 hour ago" | grep error я просто пишу в Telegram: «Покажи ошибки 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 нет. Это позволяет тестировать логику отдельно от 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 allowed\n" 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("&", "&").replace("<", "<").replace(">", ">")
Проблема 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("\n\n", 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
Требования:
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 для безопасного вызова CLISafety prompt с ключевыми словами для разделения read/write операций
Двухуровневое хранение сессий: memory cache + PostgreSQL
Форматтер с защитой code blocks и разбиением длинных сообщений
Таймауты и graceful shutdown для стабильности
Проект open-source, буду рад issues и PR.
Ссылки:
Репозиторий: github.com/gmen1057/claude-cli-telegrambot
Claude Code: docs.anthropic.com/en/docs/claude-code
