Building a Telegram lead-intake bot with aiogram 3 and SQLite: a form, an admin inbox and no lost requests

Written with AI assistance; the template itself was built and tested by the author's AI assistant.

Many small businesses take requests in Telegram DMs: "how much is a haircut?", "call me back", a phone number buried in a voice message. Some of those get lost between chats. This article shows a small structure that fixes it: a bot that asks a fixed set of questions, validates the answers, writes them to SQLite before doing anything else, and pushes a card with status buttons to the admins.

Everything below is adapted from a small aiogram 3 project. You can rebuild the ideas yourself from the snippets.

The shape of the thing

• A client presses /start and walks through a form: service, name, phone, comment, confirmation.
• The lead is saved to SQLite and gets a number.
• Every admin receives a card with buttons: In work / Done / Rejected.
• When an admin changes the status, the client gets a message about it.
• A per-user limit stops someone from flooding the inbox.

1. The form as an FSM

aiogram's finite state machine is a good fit: one state per question.

from aiogram.fsm.state import State, StatesGroup


class LeadForm(StatesGroup):
    service = State()
    name = State()
    phone = State()
    comment = State()
    confirm = State()

Handlers are filtered by state, so a message only reaches the handler that expects it. Moving forward is a lookup table, and an editing flag lets the user fix one field from the confirmation screen and jump straight back:

NEXT_STEP = {
    LeadForm.service: LeadForm.name,
    LeadForm.name: LeadForm.phone,
    LeadForm.phone: LeadForm.comment,
}


async def advance(current: State, message: Message, state: FSMContext, config: Config) -> None:
    """After a field is entered: next step, or back to the confirmation screen."""
    data = await state.get_data()
    if data.get("editing") or current == LeadForm.comment:
        await show_confirm(message, state, config)
    else:
        await ask(NEXT_STEP[current], message, state, config)

Two catch-all handlers matter more than they look: one for users with no state (StateFilter(None) → "Press /start") and one for any message inside the form that no step handler accepted, such as a sticker instead of a name. Without them the bot just goes silent, and silence reads as "broken".

(advance is from app/handlers/client.py; LeadForm from app/states.py.)

2. Validate before you store

A lead with a bad phone number is a lost lead. The bot accepts both Telegram's "share contact" button and typed text, strips spaces, dashes, dots and brackets, and normalizes to +digits. The English variant of the check:

def _normalize_international(cleaned: str) -> str | None:
    """Optional "+" followed by 8-15 digits, returned as +XXXXXXXX.

    Local numbers with a leading trunk "0" are rejected (the country code is unknown);
    the "00" international prefix is treated as "+".
    """
    digits = cleaned.lstrip("+")
    if not cleaned.startswith("+"):
        if digits.startswith("00"):
            digits = digits[2:]
        elif digits.startswith("0"):
            return None
    if not 8 <= len(digits) <= 15 or digits.startswith("0"):
        return None
    return "+" + digits

Rejecting a local 07700… is a deliberate trade-off: without a country code the number is not callable from elsewhere, so the bot asks again with an example. The Russian variant additionally turns 8XXXXXXXXXX and 9XXXXXXXXX into +7…. Names are trimmed, whitespace-collapsed and limited to 2–60 characters; comments to 500. User text is passed through html.escape before it is shown in the HTML-formatted confirmation. (From app/validators.py.)

3. Save first, notify second

Two tables, users and leads, with a status column and indexes for the queries the admin screens use:

CREATE TABLE IF NOT EXISTS leads (
    id         INTEGER PRIMARY KEY AUTOINCREMENT,
    user_id    INTEGER NOT NULL,
    username   TEXT,
    service    TEXT NOT NULL,
    name       TEXT NOT NULL,
    phone      TEXT NOT NULL,
    comment    TEXT NOT NULL DEFAULT '',
    status     TEXT NOT NULL DEFAULT 'new',
    created_at TEXT NOT NULL,
    updated_at TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_leads_user_created ON leads (user_id, created_at);
CREATE INDEX IF NOT EXISTS idx_leads_status ON leads (status, id);

Timestamps are UTC strings in %Y-%m-%d %H:%M:%S, which sort lexicographically, so created_at >= ? just works. The connection uses WAL mode (PRAGMA journal_mode=WAL) via aiosqlite.

The "send" handler is ordered on purpose: clear the FSM state first (a double tap on the button cannot create a duplicate), check the limit, then insert, answer the client, and only then notify admins. If Telegram fails to deliver to an admin, the lead is already in the database and visible in the admin list. That is the "no lost requests" part. (From app/db.py, confirm_send in app/handlers/client.py.)

4. The admin inbox

Admins get the lead card with one button per other status:

def lead_status_kb(lead_id: int, current: str, language: str = "ru") -> InlineKeyboardMarkup:
    """Status change buttons; the current status is not shown."""
    row = [
        InlineKeyboardButton(
            text=tr(language, f"btn_{status}"), callback_data=f"lead:{lead_id}:{status}"
        )
        for status in STATUS_BUTTONS
        if status != current
    ]
    return InlineKeyboardMarkup(inline_keyboard=[row] if row else [])

Sending goes through a wrapper that survives the things that actually happen: a flood limit (one retry after the retry_after delay), a user who blocked the bot (marked blocked in users), and other API errors (logged, not raised). Admins are notified concurrently:

async def notify_admins(bot: Bot, config: Config, db: Database, lead: dict) -> int:
    """Sends the lead card to all admins. Returns the number delivered."""
    text = config.tr("new_lead") + format_lead(lead, config.tz, config.language)
    markup = lead_status_kb(lead["id"], lead["status"])
    results = await asyncio.gather(
        *(safe_send(bot, admin_id, text, reply_markup=markup) for admin_id in config.admin_ids)
    )
    return sum(results)

The callback handler for the buttons validates the status against a whitelist, ignores a repeated press of the same status, updates the row, edits the card in place and tells the client:

@router.callback_query(F.data.startswith("lead:"))
async def change_status(cb: CallbackQuery, db: Database, config: Config) -> None:
    _, lead_id_raw, status = cb.data.split(":")
    if status not in STATUSES:
        await cb.answer()
        return
    # ... "lead not found" and "already has this status" checks omitted ...
    lead = await db.set_status(existing["id"], status)
    await cb.answer(status_label(status, config.language))
    # ... the card is edited in place here ...
    key = CLIENT_STATUS_TEXT.get(status)
    if key:  # the "new" status is not sent to the client
        await safe_send(cb.bot, lead["user_id"], config.text(key, number=lead["id"]), db=db)

Admin handlers sit on a router with a filter that checks the user ID against ADMIN_IDS, so everyone else's taps are never routed there. The admin menu (/admin) also has paginated lists of new and in-work leads, today/7-day counts, a CSV export (cells starting with =, +, -, @ are prefixed to defuse spreadsheet formula injection) and a broadcast to everyone who has not blocked the bot. (From app/keyboards.py, app/notify.py, app/handlers/admin.py.)

5. Anti-flood without extra infrastructure

You do not need Redis for this. The leads table already knows how many requests a user created recently:

WINDOW = timedelta(hours=1)


async def can_create_lead(
    db: Database, user_id: int, limit: int, now: datetime | None = None
) -> bool:
    since = (now or utcnow()) - WINDOW
    return await db.count_recent_leads(user_id, since) < limit

The check runs on /start and again on confirm. The limit comes from config, so a salon can set 3 per hour and a busier shop something else. (From app/throttle.py.)

6. Config and language

Texts, services and limits live in YAML, not in code, so the owner changes wording without touching Python. One setting switches the built-in buttons, admin panel, date format and phone rules between ru and en:

language: en
timezone: Europe/London

working_hours:
  text: "Mon–Sat, 10:00–20:00"
  start: "10:00"
  end: "20:00"
  days: [1, 2, 3, 4, 5, 6]       # 1 = Monday ... 7 = Sunday

antispam:
  max_leads_per_hour: 3

services:
  - "Haircut"
  - "Colouring"
  - "Other"

Outside working hours the client still gets the confirmation, plus a line saying when you will reply. Secrets (BOT_TOKEN, ADMIN_IDS) stay in .env, not in this file. (From config.en.yaml, trimmed.)

7. Running it

The container is plain python:3.12-slim, runs as a non-root user, and uses long polling, so there is no webhook, domain or TLS to set up:

services:
  bot:
    build: .
    container_name: leads-bot
    restart: unless-stopped
    env_file: .env
    environment:
      DB_PATH: data/leads.db
    volumes:
      - ./data:/app/data                         # the leads database
      - ./config.yaml:/app/config.yaml:ro        # texts and services, editable without a rebuild
      - ./config.en.yaml:/app/config.en.yaml:ro  # English edition (CONFIG_PATH=config.en.yaml)

Put BOT_TOKEN, ADMIN_IDS and, for English, CONFIG_PATH=config.en.yaml in .env, then docker compose up -d --build. Every admin must press /start in the bot once, otherwise Telegram will not let the bot message them. (From docker-compose.yml.)

What this is not

• Not a CRM. There are four statuses, a list and a CSV export. No pipelines, reminders, assignments or history of changes.
• One SQLite file. Simple and enough for a small inbox, but it is your only copy. Back up data/leads.db (or use the CSV export) on a schedule.
• A polling bot on one process. Run exactly one instance per token; a second one will fight over updates. FSM state is not meant to be shared between processes.
• No payments. It collects requests; it does not take money.

Wrapping up

The ideas that carry over to any bot: one FSM state per question, normalize input before you store it, write to the database before notifying anyone, keep status changes behind a whitelist, and let the database enforce your rate limit. That is enough to stop losing requests.

If you would rather not assemble it yourself: the full template (source, tests, Docker setup, config in English and Russian, README in both languages) is sold for 9 USDT via the Telegram shop. There is also a free monitoring kit for a single VPS if you want to know when the bot's server goes down.