Building Reliable Telegram AI Workflows

Telegram is a good front door for AI workflows. It is already on your phone, it handles authentication for you, and its Bot API is simple and well documented. The hard part is not getting a bot to answer. It is getting it to answer every message exactly once, even when the model is slow, the server restarts or Telegram retries a delivery.

This piece describes a shape that holds up: a webhook that does almost nothing, a queue, and a worker that does the slow work.

The shape of a reliable bot

1. Telegram calls your webhook with an update.
2. The webhook verifies the request, stores the update and answers immediately.
3. A worker picks the job up, calls the model, and replies through the Bot API.

The split matters because the webhook and the model have very different timing. Telegram expects a quick answer to its request. A model call can take many seconds. If the webhook waits for the model, slow responses turn into timeouts, timeouts turn into redelivered updates, and redelivered updates turn into duplicate replies.

Verify every request

When you register the webhook, pass a secret_token. Telegram then sends it back on every request in the X-Telegram-Bot-Api-Secret-Token header, which lets you reject anything that did not come from Telegram.

curl -s "https://api.telegram.org/bot${TELEGRAM_BOT_TOKEN}/setWebhook" \
  -d "url=https://bots.example.com/telegram/webhook" \
  -d "secret_token=${TELEGRAM_WEBHOOK_SECRET}" \
  -d 'allowed_updates=["message"]' \
  -d "drop_pending_updates=true"

allowed_updates keeps the bot from receiving update types it does not handle. The secret may only contain letters, digits, underscores and hyphens.

Store first, then answer

Every update carries an update_id. Make it the primary key of your job table and the database does the deduplication for you:

CREATE TABLE IF NOT EXISTS jobs (
    update_id  INTEGER PRIMARY KEY,
    payload    TEXT    NOT NULL,
    status     TEXT    NOT NULL DEFAULT 'queued',
    attempts   INTEGER NOT NULL DEFAULT 0,
    locked_at  TEXT,
    last_error TEXT
);

The webhook itself stays tiny. It checks the secret, inserts the update with INSERT OR IGNORE, and returns:

import hmac
import json
import os
import sqlite3

from fastapi import FastAPI, Header, HTTPException, Request

SECRET = os.environ["TELEGRAM_WEBHOOK_SECRET"]
DB_PATH = os.environ.get("JOBS_DB", "jobs.db")

app = FastAPI()


@app.post("/telegram/webhook")
async def telegram_webhook(
    request: Request,
    x_telegram_bot_api_secret_token: str | None = Header(default=None),
):
    if not x_telegram_bot_api_secret_token or not hmac.compare_digest(
        x_telegram_bot_api_secret_token, SECRET
    ):
        raise HTTPException(status_code=401)

    update = await request.json()

    conn = sqlite3.connect(DB_PATH, timeout=5)
    try:
        with conn:
            conn.execute(
                "INSERT OR IGNORE INTO jobs (update_id, payload) VALUES (?, ?)",
                (update["update_id"], json.dumps(update)),
            )
    finally:
        conn.close()

    return {"ok": True}

If Telegram delivers the same update twice, the second insert is ignored and the user still gets one reply.

Do the slow work in a worker

The worker claims one job at a time. A single UPDATE ... RETURNING statement marks the job as taken and hands it back, and jobs left in working for too long, after a crash for example, become claimable again:

def claim_next(conn: sqlite3.Connection):
    row = conn.execute(
        """
        UPDATE jobs
           SET status = 'working',
               attempts = attempts + 1,
               locked_at = datetime('now')
         WHERE update_id = (
               SELECT update_id FROM jobs
                WHERE status = 'queued'
                   OR (status = 'working' AND locked_at < datetime('now', '-5 minutes'))
                ORDER BY update_id
                LIMIT 1)
        RETURNING update_id, payload, attempts
        """
    ).fetchone()
    conn.commit()
    return row

RETURNING needs SQLite 3.35 or newer, which Ubuntu 24.04 ships. With more than one worker, wrap the claim in BEGIN IMMEDIATE so two workers never take the same job.

Give every external call a timeout, cap the number of attempts, and record the error when a job finally fails. A job that fails quietly is worse than one that fails loudly.

Respect Telegram's limits

The Bot API documents a few limits that matter for AI replies:

• A single message holds at most 4,096 characters, so long answers need splitting.
• Bots should stay around one message per second in a single chat.
• When you go too fast, the API answers with HTTP 429 and a retry_after value in seconds. Wait that long, then try again.

import os
import time

import httpx

API = f"https://api.telegram.org/bot{os.environ['TELEGRAM_BOT_TOKEN']}"


def split_message(text: str, limit: int = 4096) -> list[str]:
    chunks = []
    while len(text) > limit:
        cut = text.rfind("\n", 0, limit)
        if cut <= 0:
            cut = limit
        chunks.append(text[:cut])
        text = text[cut:].lstrip("\n")
    chunks.append(text)
    return chunks


def send_message(chat_id: int, text: str, attempts: int = 5) -> None:
    for chunk in split_message(text):
        for _ in range(attempts):
            response = httpx.post(
                f"{API}/sendMessage",
                json={"chat_id": chat_id, "text": chunk},
                timeout=15,
            )
            data = response.json()
            if data.get("ok"):
                break
            retry_after = data.get("parameters", {}).get("retry_after")
            if response.status_code == 429 and retry_after:
                time.sleep(retry_after)
                continue
            raise RuntimeError(data.get("description", "sendMessage failed"))
        else:
            raise RuntimeError("sendMessage kept hitting the rate limit")

Telegram counts length slightly differently from Python, so leave a little headroom if your answers often sit near the limit.

Show that something is happening

A model call that takes ten seconds feels broken if nothing happens on screen. Send a typing action with sendChatAction when the worker starts. It shows for about five seconds or until your reply arrives, so repeat it for longer jobs.

Checklist

• The webhook verifies the secret token and answers immediately.
• update_id is a primary key, so duplicates are ignored.
• Slow work runs in a worker, with timeouts and a retry limit.
• Long replies are split, and 429 responses are honoured.
• Failed jobs keep their error, so you can see what went wrong and replay them.

None of this is specific to AI. It is ordinary queue discipline. The model just makes the slow part slower, which is why the discipline matters more.