Telegram-бот на aiogram 3 с Claude через API-шлюз

Telegram-бот с нейросетью — частая задача: ответы на вопросы клиентов, помощник для команды, обработка текстов прямо в мессенджере. Ниже — Telegram-бот на aiogram 3, который отправляет сообщения в Claude, помнит контекст разговора, показывает «печатает…» и понятно сообщает об ошибках. В конце — полный код одним файлом.

Что понадобится

• Python 3.10 или новее.
• Токен бота от @BotFather↗.
• Ключ Zukko вида fetch_... — создаётся в кабинете↗.

Zukko принимает запросы в OpenAI-формате, поэтому для связи с Claude хватит официального пакета openai и его асинхронного клиента AsyncOpenAI. Base URL — https://api.zukko.pro/v1.

pip install aiogram openai python-dotenv

Пакет asyncio ставить не нужно — он входит в стандартную библиотеку Python.

Создайте рядом с кодом файл .env:

Как устроен Telegram-бот на aiogram 3

В aiogram 3 обработчики вешаются на Router, а Dispatcher получает обновления от Telegram и раздаёт их подключённым роутерам. Фильтры задаются аргументами декоратора: CommandStart(), Command("clear"), «магический» фильтр F.text (сообщение с текстом). Объект Bot создаётся отдельно и передаётся в dp.start_polling(bot). Так устроен и пример из документации aiogram↗.

Каркас выглядит так:

router = Router()

@router.message(CommandStart())
async def cmd_start(message: Message) -> None:
    await message.answer("Привет! Напишите вопрос. /clear — начать заново.")

async def main() -> None:
    bot = Bot(token=BOT_TOKEN)
    dp = Dispatcher()
    dp.include_router(router)
    await dp.start_polling(bot)

Порядок обработчиков важен: aiogram проверяет их сверху вниз. Поэтому команды /start и /clear регистрируем раньше общего обработчика текста, иначе он перехватит и команды.

История диалога

API модели ничего не помнит между запросами: чтобы бот понимал «а подробнее?», нужно каждый раз отправлять предыдущие реплики. Для примера храним историю в словаре user_id -> список сообщений. После перезапуска процесса она пропадёт — для продакшена переносите её в Redis или базу.

Два правила, которые избавят от проблем:

1. Ограничивайте длину. Вся история уходит во входные токены при каждом сообщении, поэтому оставляем последние MAX_HISTORY реплик.
2. Начинайте историю с реплики пользователя. После обрезки первой может оказаться реплика ассистента; Anthropic-формат ожидает, что диалог начинается с user, поэтому лишнее в начале отбрасываем.

def trim(history: list[dict]) -> list[dict]:
    history = history[-MAX_HISTORY:]
    while history and history[0]["role"] != "user":
        history.pop(0)
    return history

Если запрос к модели упал, реплику пользователя в историю не сохраняем — иначе в контексте появятся два вопроса подряд без ответа.

Запрос к Claude и индикатор «печатает»

Ответ модели может идти несколько секунд. Статус «печатает» в Telegram гаснет примерно через 5 секунд, поэтому один вызов send_chat_action для долгого ответа не годится. В aiogram 3 для этого есть ChatActionSender из aiogram.utils.chat_action: внутри async with он повторяет действие каждые 5 секунд, пока блок не завершится.

async with ChatActionSender.typing(bot=bot, chat_id=message.chat.id):
    response = await client.chat.completions.create(
        model=MODEL,
        messages=[{"role": "system", "content": SYSTEM_PROMPT}, *history],
        max_tokens=1024,
    )

Объект bot aiogram сам передаёт в обработчик, если указать параметр bot: Bot.

Ещё одно ограничение Telegram: одно сообщение — не длиннее 4096 символов. Длинный ответ режем на части.

Обработка ошибок API

Все ошибки с HTTP-статусом SDK openai поднимает как APIStatusError (и его подклассы), а сетевые сбои — как APIConnectionError. Удобно сопоставить коды Zukko с понятными пользователю сообщениями:

Подробности о ключах и лимитах — в разделе лимиты ключей↗.

Полный код бота

import asyncio
import logging
import os

from aiogram import Bot, Dispatcher, F, Router
from aiogram.filters import Command, CommandStart
from aiogram.types import Message
from aiogram.utils.chat_action import ChatActionSender
from dotenv import load_dotenv
from openai import APIConnectionError, APIStatusError, AsyncOpenAI

load_dotenv()

BOT_TOKEN = os.environ["TELEGRAM_BOT_TOKEN"]

Запуск: python bot.py. Бот работает через long polling, внешний адрес и вебхук для начала не нужны.

Сообщения без текста (стикеры, фото) фильтр F.text пропускает мимо — бот на них просто не отвечает. parse_mode не задан намеренно: модель часто отвечает в Markdown, и строгий парсер Telegram может отклонить такой текст, а простой текст отправится всегда.

Какую модель выбрать и сколько это стоит

Модель задаётся переменной ZUKKO_MODEL, код менять не нужно. Варианты из текущего каталога:

Актуальный список — GET https://api.zukko.pro/v1/models или страница «Модели» в кабинете. Какая модель лучше подходит вашему боту, проверьте на своих задачах: прогоните 20–30 типичных вопросов пользователей через две-три модели и сравните ответы.

Пример расчёта. Допустим, одно сообщение пользователя вместе с историей — 2 000 входных токенов, ответ — 500 выходных. На claude-sonnet-5: 2 000 × $0.30 / 1 000 000 + 500 × $0.30 / 1 000 000 = $0.0006 + $0.00015 = $0.00075. Тысяча таких сообщений — $0.75. На glm-5.3-flash та же тысяча: 2 500 000 × $0.04 / 1 000 000 = $0.10. Учтите, что с ростом истории растут и входные токены, — поэтому её и стоит обрезать.

Частые вопросы

Почему бот молчит?

Проверьте логи: при ошибке API код пишет статус и тело ответа через logging.warning. Частые причины — неверный ключ в .env (401), нулевой баланс (402) или модель, не разрешённая для ключа (403). Если в логах нет даже входящих обновлений — проверьте токен бота.

Можно ли стримить ответ в Telegram по мере генерации?

Можно, через stream=True и периодическое редактирование сообщения (message.edit_text). Но у Telegram есть лимиты на частоту редактирования, поэтому обновляйте текст не чаще раза в секунду-две. В стриминге Zukko последний чанк приходит с usage и пустым choices, так что перед chunk.choices[0] проверяйте if chunk.choices:.

Как сделать, чтобы история не терялась после перезапуска?

Замените словарь histories на хранилище: Redis со списком по ключу user_id или таблицу в базе. Логика trim() остаётся той же.

Можно ли вызывать Claude в Anthropic-формате?

Да. Шлюз принимает оба формата, и любую модель можно звать в любом из них. С официальным SDK anthropic base URL указывается без /v1: https://api.zukko.pro. Разница форматов описана в статье какой формат выбрать↗.