Telegram bot on aiogram 3 with Claude through an API gateway

A Telegram bot backed by an AI model is a common task: answering customer questions, a team assistant, processing text right inside the messenger. Below is a Telegram bot on aiogram 3 that sends messages to Claude, remembers the conversation, shows "typing…" and reports errors clearly. The full code is at the end in a single file.

What you need

• Python 3.10 or newer.
• A bot token from @BotFather↗.
• A Zukko key of the form fetch_..., created in the dashboard↗.

Zukko accepts requests in the OpenAI format, so the official openai package with its async AsyncOpenAI client is enough to talk to Claude. The base URL is https://api.zukko.pro/v1.

pip install aiogram openai python-dotenv

You don't need to install asyncio: it is part of the Python standard library.

Create a .env file next to the code:

How a Telegram bot on aiogram 3 is structured

In aiogram 3, handlers are attached to a Router, and the Dispatcher receives updates from Telegram and passes them to the included routers. Filters are decorator arguments: CommandStart(), Command("clear"), the "magic" filter F.text (a message with text). The Bot object is created separately and passed to dp.start_polling(bot). The example in the aiogram documentation↗ follows the same structure.

The skeleton looks like this:

router = Router()

@router.message(CommandStart())
async def cmd_start(message: Message) -> None:
    await message.answer("Hi! Send me a question. /clear starts over.")

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

Handler order matters: aiogram checks them top to bottom. That is why /start and /clear are registered before the general text handler; otherwise it would swallow the commands too.

Conversation history

The model API remembers nothing between requests: for the bot to understand "tell me more", you have to send previous turns every time. In this example the history lives in a dict user_id -> list of messages. It is lost when the process restarts; for production, move it to Redis or a database.

Two rules that save trouble:

1. Limit the length. The whole history is billed as input tokens on every message, so we keep only the last MAX_HISTORY turns.
2. Start the history with a user turn. After trimming, the first item may be an assistant turn; the Anthropic format expects a conversation to start with user, so leading extras are dropped.

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

If the model request fails, the user turn is not saved to history; otherwise the context would contain two questions in a row with no answer.

Calling Claude with a typing indicator

A model answer can take several seconds. Telegram's "typing" status disappears after about 5 seconds, so a single send_chat_action call is not enough for a long answer. aiogram 3 has ChatActionSender in aiogram.utils.chat_action for this: inside async with it repeats the action every 5 seconds until the block finishes.

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,
    )

aiogram injects the bot object into the handler if you declare a bot: Bot parameter.

One more Telegram limit: a single message can be at most 4096 characters. Long answers are split into parts.

Handling API errors

The openai SDK raises every error with an HTTP status as APIStatusError (and its subclasses), and network failures as APIConnectionError. It is convenient to map Zukko status codes to messages a user understands:

More on keys and limits: key limits↗.

Full bot code

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"]

Run it with python bot.py. The bot uses long polling, so you don't need a public address or a webhook to start.

Messages without text (stickers, photos) are skipped by the F.text filter, so the bot simply doesn't answer them. parse_mode is intentionally not set: models often answer in Markdown, and Telegram's strict parser may reject such text, while plain text always goes through.

Which model to pick and what it costs

The model is set by the ZUKKO_MODEL variable, no code changes needed. Options from the current catalog:

The current list is at GET https://api.zukko.pro/v1/models or on the "Models" page in the dashboard. Test which model suits your bot on your own tasks: run 20–30 typical user questions through two or three models and compare the answers.

An example calculation. Say one user message plus history is 2,000 input tokens and the answer is 500 output tokens. On claude-sonnet-5: 2,000 × $0.30 / 1,000,000 + 500 × $0.30 / 1,000,000 = $0.0006 + $0.00015 = $0.00075. A thousand such messages cost $0.75. On glm-5.3-flash the same thousand: 2,500,000 × $0.04 / 1,000,000 = $0.10. Keep in mind that input tokens grow as the history grows, which is exactly why it should be trimmed.

FAQ

Why is the bot silent?

Check the logs: on an API error the code writes the status and response body via logging.warning. Common causes are a wrong key in .env (401), zero balance (402) or a model not allowed for the key (403). If the logs don't show even incoming updates, check the bot token.

Can I stream the answer into Telegram as it is generated?

Yes, with stream=True and periodic message edits (message.edit_text). Telegram limits how often you can edit, so update the text no more than once every second or two. In Zukko streaming the last chunk carries usage and an empty choices, so check if chunk.choices: before chunk.choices[0].

How do I keep history across restarts?

Replace the histories dict with storage: a Redis list keyed by user_id or a database table. The trim() logic stays the same.

Can I call Claude in the Anthropic format?

Yes. The gateway accepts both formats, and any model can be called in either. With the official anthropic SDK the base URL has no /v1: https://api.zukko.pro. The difference between formats is explained in choosing a format↗.