Kaspi в Telegram-боте: QR, вебхук и сверка без скриншотов (Python, aiogram 3)
В Казахстане типичная продажа через Telegram выглядит так: бот или менеджер пишет «переведите 4 990 ₸ на Kaspi», клиент присылает скриншот, человек открывает приложение и ищет платёж глазами. Это медленно, это не работает ночью, и это ломается о поддельные чеки.
Ниже — бот, который сам выставляет QR Kaspi Pay, узнаёт об оплате за несколько секунд и выдаёт товар без участия человека. Код рабочий, около 200 строк на Python. Разберём по частям всё, на чём такие интеграции обычно ломаются: подпись вебхука, повторные доставки, идемпотентность, истёкший QR и «тихую» смерть сессии Kaspi.
Схема (QR → вебхук → сверка) от конкретного провайдера не зависит. Чтобы код был рабочим, а не псевдокодом, в примере используется REST API Aitole — сервиса, который выставляет счета через Kaspi-аккаунт самого продавца (автор участвует в его разработке; это независимый сервис, не банк и не партнёр АО «Kaspi Bank»). Если у вас другой посредник, поменяются URL и названия полей, но не грабли.
Почему не встроенные платежи Telegram
Bot Payments в Telegram работают через подключённых платёжных провайдеров — это эквайринг: договор, проверка, процент с оборота. У большинства небольших продавцов в Казахстане эквайринга нет, зато есть Kaspi Pay для бизнеса, и покупатели платят через Kaspi по привычке. Задача — автоматизировать именно этот Kaspi, а не подключать новый способ оплаты.
Как это устроено
Покупатель ──/start──▶ Бот ──POST /v2/qr──▶ Aitole ──▶ Kaspi-касса продавца
▲ │ │
│ QR + кнопка │ │ поллер видит оплату
└───────────────────┘ ▼
◀── вебхук payment.paid (HMAC-SHA256)
◀── страховка: GET /v2/operations/{id} раз в минуту
Деньги идут напрямую на Kaspi-счёт продавца. Посредник работает через штатную роль «Кассир» в Kaspi Pay: она умеет выставлять счета, но не умеет выводить деньги. С точки зрения бота это просто HTTP API с тремя действиями: создать счёт, узнать статус, получить событие.
Что понадобится
Python 3.10+,
aiogram3,aiohttp,httpx,qrcode[pil].Токен бота от @BotFather.
Тестовый API-ключ
sk_test_…. Он работает только с песочницей: тестовый ключ с настоящим Kaspi API не пустит (403 key_mode_mismatch), так что случайно списать живые деньги нельзя. Это полезное свойство, которое стоит требовать от любого платёжного API.Публичный HTTPS-адрес для вебхука.
localhostи внутренние адреса как адрес вебхука не принимаются, поэтому для локальной разработки нужен туннель (ngrok, cloudflared и т. п.).Справочник API, по которому написан код: aitole.kz/docs.
pip install aiogram aiohttp httpx "qrcode[pil]"
export BOT_TOKEN=... AITOLE_KEY=sk_test_... AITOLE_WEBHOOK_SECRET=whsec_... ADMIN_CHAT_ID=...
Подписка на вебхук создаётся в личном кабинете сервиса (не по API-ключу — и правильно: утёкший ключ не должен позволять перенаправить ваши события на чужой адрес). Адрес https://ваш-домен/aitole/webhook, событие payment.paid (плюс payment.expired и connection.session_expired — ниже станет ясно зачем). Секрет whsec_… показывается один раз — сохраните сразу.
Шаг 1. Выставляем QR
async def create_qr(order: dict) -> dict:
order["attempt"] += 1
r = await http.post(
"/v2/qr",
headers={"Idempotency-Key": f"tgbot-{order['id']}-{order['attempt']}"},
json={"amount": order["amount"],
"comment": order["title"],
"metadata": {"order_id": order["id"]}},
)
body = r.json()
if r.status_code >= 400:
err = body["error"]
raise AitoleError(f"{err['code']}: {err['message']}")
# Провайдер может отказать с HTTP 200 — смотрим статус, а не код ответа
if body["status"] == "failed":
raise AitoleError(body.get("failure_message") or "Kaspi отказал")
order["op_id"] = body["id"]
by_op[body["id"]] = order["id"]
return body
Три детали, которые здесь важнее, чем кажется.
Idempotency-Key обязателен. Сеть рвётся, httpx бросает таймаут, пользователь дважды жмёт кнопку. Повтор с тем же ключом вернёт тот же QR, а не второй счёт. Ключ уникален в пределах аккаунта и живёт 24 часа, поэтому ставим префикс (tgbot-): если у вас ещё и сайт нумерует заказы с единицы, ключи не пересекутся. Номер попытки в ключе нужен, чтобы после истечения QR выставить новый — с тем же ключом вернулся бы старый.
HTTP 200 ещё не успех. Если Kaspi отказал, ответ придёт с кодом 200, status: "failed" и причиной в failure_message. Проверять только raise_for_status() недостаточно.
Ответ на создание — не подтверждение оплаты. Он возвращает операцию в статусе pending, ссылку qr_token и expires_at. Об оплате скажет событие.
Ответ выглядит так:
{
"id": "op_d7834b…",
"type": "qr",
"status": "pending",
"amount": 4990,
"qr_token": "https://pay.kaspi.kz/…",
"expires_at": "2026-09-08T00:37:24Z",
"created_at": "2026-09-08T00:32:24Z"
}
Шаг 2. Показываем QR так, чтобы было удобно с любого устройства
qr_token — обычная ссылка. С телефона удобнее нажать кнопку: откроется Kaspi. С Telegram Desktop удобнее отсканировать картинку. Отправляем оба варианта одним сообщением:
kb = InlineKeyboardMarkup(inline_keyboard=[
[InlineKeyboardButton(text="Оплатить в Kaspi", url=op["qr_token"])]])
await bot.send_photo(chat_id, qr_png(op["qr_token"]), reply_markup=kb,
caption="С телефона нажмите кнопку, с компьютера "
"отсканируйте QR в Kaspi. QR действует 5 минут.")
QR Kaspi живёт 5 минут — об этом стоит написать покупателю прямо в подписи.
Ещё одна особенность Kaspi: по QR он не сообщает, кто заплатил. В боте это не проблема — мы сами связали операцию op_… с chat_id в момент выставления. Поэтому храните op_id рядом с заказом: по нему вы потом найдёте, кому выдавать товар.
Шаг 3. Принимаем вебхук и проверяем подпись
Вебхук приходит с заголовками X-Webhook-Timestamp и X-Webhook-Signature: sha256=<hex>. Подпись — HMAC-SHA256 от строки "{timestamp}.{сырое тело}" секретом эндпоинта.
def signature_ok(raw: bytes, ts: str, got: str) -> bool:
mine = "sha256=" + hmac.new(WEBHOOK_SECRET, ts.encode() + b"." + raw,
hashlib.sha256).hexdigest()
if not hmac.compare_digest(mine, got):
return False
try:
return abs(time.time() - int(ts)) <= 300 # защита от повтора
except ValueError:
return False
async def aitole_webhook(request: web.Request) -> web.Response:
raw = await request.read() # сырые байты, до json.loads
if not signature_ok(raw,
request.headers.get("X-Webhook-Timestamp", ""),
request.headers.get("X-Webhook-Signature", "")):
return web.Response(status=400)
event = json.loads(raw)
if event["id"] not in seen_events: # доставка бывает двойной
seen_events.add(event["id"])
asyncio.create_task(handle_event(event)) # отвечаем сразу, работаем потом
return web.Response(status=200)
Что здесь можно сделать неправильно — и что делают чаще всего:
Считать подпись по распарсенному JSON.
json.dumps(json.loads(raw))меняет пробелы и порядок ключей, подпись не сойдётся. Только сырые байты.Сравнивать через
==. Обычное сравнение строк прерывается на первом отличии, и по времени ответа можно подбирать подпись.hmac.compare_digestработает за постоянное время.Не проверять свежесть. Без окна в 300 секунд перехваченный запрос «оплачено» можно прислать ещё раз через неделю.
Долго работать внутри запроса. У отправителя таймаут 10 секунд; если за это время вы ходите в базу, выдаёте товар и шлёте три сообщения, успешная доставка превращается в повтор. Отвечаем
200сразу, обработку уводим в фоновую задачу (в бою — в очередь).Не ждать дублей и перестановок. Неудачная доставка повторяется через 1 мин, 5 мин, 30 мин, 2 ч, 6 ч и 24 ч, а порядок не гарантирован: повтор
payment.createdможет прийти послеpayment.paid. Поэтому дедуплицируем поidконверта и смотрим наdata.status, а не на очерёдность.
Шаг 4. Обрабатываем события
async def handle_event(event: dict) -> None:
kind, data = event["type"], event.get("data") or {}
if kind == "connection.session_expired":
await bot.send_message(ADMIN_CHAT_ID,
"Сессия Kaspi слетела: приём платежей стоит. "
"Войдите заново по SMS в кабинете Aitole.")
return
order = orders.get(by_op.get(data.get("id", "")))
if not order or data.get("id") != order["op_id"]:
return # чужая или старая попытка
if kind == "payment.paid" and data.get("status") == "paid":
await on_paid(order)
elif kind == "payment.expired":
await on_expired(order)
on_paid первым делом проверяет, не выдан ли уже товар: оплату может принести и вебхук, и сверка из следующего шага. on_expired предлагает кнопку «Новый QR» — это новая попытка с новым Idempotency-Key.
Отдельно про connection.session_expired. У Kaspi одна активная сессия на номер. Если кто-то войдёт в приложение Kaspi Pay под номером кассира, отданным платформе, сессия платформы слетит, и новые счета начнут отвергаться с 409 no_active_connection. Бот при этом выглядит живым, но денег не принимает. Поэтому номер кассира должен быть выделенным, а событие — уходить админу немедленно. Это тот случай, когда мониторинг «бот отвечает» ничего не показывает: нужен мониторинг «деньги принимаются».
Шаг 5. Страховка: сверка, если вебхук не дошёл
Ваш сервер перезагружался, туннель упал, сертификат истёк — вебхук не пришёл. Покупатель заплатил и ждёт. Раз в минуту проходим по зависшим заказам и спрашиваем статус напрямую:
async def reconcile() -> None:
while True:
await asyncio.sleep(60)
for order in list(orders.values()):
if order["status"] != "pending" or not order["op_id"]:
continue
if time.time() - order["created"] < 60:
continue
r = await http.get(f"/v2/operations/{order['op_id']}")
if r.status_code != 200:
continue
status = r.json()["status"]
if status == "paid":
await on_paid(order)
elif status == "expired":
await on_expired(order)
Для большого потока удобнее читать журнал GET /v2/events с курсором: он хранит всю историю и переживает любой сбой доставки. Правило простое: вебхук — для скорости, журнал или опрос — для надёжности. Лучше оба.
Шаг 6. Проверяем в песочнице
С тестовым ключом qr_token заведомо ненастоящий (mock_qr_…), оплатить его нельзя. Оплату изображаем вручную:
# «покупатель оплатил»
curl -X POST https://api.aitole.kz/v2/sandbox/operations/op_…/pay -H "X-API-Key: sk_test_…"
# «QR истёк»
curl -X POST https://api.aitole.kz/v2/sandbox/operations/op_…/expire -H "X-API-Key: sk_test_…"
# «клиент отклонил счёт»
curl -X POST https://api.aitole.kz/v2/sandbox/operations/op_…/reject -H "X-API-Key: sk_test_…"
Эти ручки меняют состояние у провайдера, а статус операции двигает тот же поллер, что и в бою, с той же задержкой в несколько секунд. Через пару секунд прилетит настоящий подписанный вебхук. Отдельно проверьте:
повтор нажатия «Купить» при искусственном таймауте — второго счёта быть не должно;
вебхук с испорченной подписью и со старым timestamp — ответ
400;выключенный вебхук — заказ всё равно закрывается сверкой через минуту;
expire→ кнопка «Новый QR» → оплата новой попытки.
Кнопка «Проверить» в кабинете сервиса шлёт webhook.test с настоящей подписью — удобно, чтобы отладить приём до первого платежа.
Хранение: две таблицы и одно правило
Словари в примере — для наглядности. После рестарта бот должен помнить, какие заказы открыты и какие события уже обработаны, иначе один и тот же платёж выдаст товар дважды или не выдаст вовсе. Минимальная схема для Postgres:
CREATE TABLE orders (
id text PRIMARY KEY,
chat_id bigint NOT NULL,
amount integer NOT NULL CHECK (amount > 0), -- целые тенге
status text NOT NULL DEFAULT 'pending', -- pending | paid | expired
attempt integer NOT NULL DEFAULT 0,
op_id text UNIQUE, -- текущая попытка оплаты
paid_at timestamptz,
created_at timestamptz NOT NULL DEFAULT now()
);
CREATE TABLE processed_events (
event_id text PRIMARY KEY, -- id конверта вебхука
created_at timestamptz NOT NULL DEFAULT now()
);
Правило одно: переход в paid делается одним условным апдейтом, и товар выдаётся, только если апдейт что-то изменил.
UPDATE orders SET status = 'paid', paid_at = now()
WHERE op_id = $1 AND status = 'pending'
RETURNING chat_id;
Если вебхук и сверка пришли одновременно, строку обновит только один из них, и второй получит пустой результат. Дедупликация событий — INSERT ... ON CONFLICT (event_id) DO NOTHING в той же транзакции. Это надёжнее любых проверок if order["status"] == "paid" в памяти процесса.
Перед запуском на живых деньгах
Боевой ключ — с минимальными правами: в scopes оставьте только
create_qrи задайте потолок суммы. Тогда утёкший ключ не выставит счёт на миллион.Номер кассира — выделенный, в приложение Kaspi Pay под ним не входить.
Лимит боевых ключей — 200 запросов в минуту; при
429ждитеRetry-After.Суммы — целые тенге, никаких float.
Фискальный чек — отдельная история: его пробивает ваша онлайн-касса, а не платёжный API.
Итого
Приём Kaspi в боте без ручной сверки сводится к пяти вещам: свой QR на каждый заказ, Idempotency-Key на каждую попытку, проверка подписи по сырому телу, идемпотентный обработчик событий и сверка по таймеру на случай потерянного вебхука. Остальное — детали конкретного API.
Если вы принимаете Kaspi в боте иначе — расскажите в комментариях, как решаете сверку. Особенно интересны случаи с несколькими менеджерами на одном номере и с оплатой по счёту на телефон вместо QR.
Полный код бота (bot.py)
import asyncio
import hashlib
import hmac
import io
import json
import os
import time
import uuid
import httpx
import qrcode
from aiogram import Bot, Dispatcher, F
from aiogram.filters import Command
from aiogram.types import (BufferedInputFile, CallbackQuery,
InlineKeyboardButton, InlineKeyboardMarkup, Message)
from aiohttp import web
API = "https://api.aitole.kz"
AITOLE_KEY = os.environ["AITOLE_KEY"] # sk_test_… или sk_live_…
WEBHOOK_SECRET = os.environ["AITOLE_WEBHOOK_SECRET"].encode()
ADMIN_CHAT_ID = int(os.environ["ADMIN_CHAT_ID"]) # куда слать тревоги
bot = Bot(os.environ["BOT_TOKEN"])
dp = Dispatcher()
http = httpx.AsyncClient(base_url=API, timeout=15,
headers={"X-API-Key": AITOLE_KEY})
PRODUCTS = {"course": ("Мини-курс по SMM", 4990),
"consult": ("Консультация 30 минут", 9000)}
# Вместо базы — словари. В бою замените на Postgres/Redis.
orders: dict[str, dict] = {} # order_id -> заказ
by_op: dict[str, str] = {} # op_id -> order_id
seen_events: set[str] = set() # id конвертов, которые уже обработали
class AitoleError(Exception):
pass
async def create_qr(order: dict) -> dict:
"""Выставляет QR. Idempotency-Key = заказ + попытка: повтор запроса
при таймауте вернёт тот же QR, а не второй счёт."""
order["attempt"] += 1
r = await http.post(
"/v2/qr",
headers={"Idempotency-Key": f"tgbot-{order['id']}-{order['attempt']}"},
json={"amount": order["amount"],
"comment": order["title"],
"metadata": {"order_id": order["id"]}},
)
body = r.json()
if r.status_code >= 400:
err = body["error"]
raise AitoleError(f"{err['code']}: {err['message']}")
# Провайдер может отказать с HTTP 200 — смотрим статус, а не код ответа
if body["status"] == "failed":
raise AitoleError(body.get("failure_message") or "Kaspi отказал")
order["op_id"] = body["id"]
by_op[body["id"]] = order["id"]
return body
def qr_png(link: str) -> BufferedInputFile:
buf = io.BytesIO()
qrcode.make(link).save(buf, format="PNG")
return BufferedInputFile(buf.getvalue(), filename="kaspi-qr.png")
async def send_payment(chat_id: int, order: dict) -> None:
try:
op = await create_qr(order)
except AitoleError as e:
await bot.send_message(chat_id, "Не получилось выставить счёт, попробуйте позже.")
await bot.send_message(ADMIN_CHAT_ID, f"Заказ {order['id']}: {e}")
return
kb = InlineKeyboardMarkup(inline_keyboard=[
[InlineKeyboardButton(text="Оплатить в Kaspi", url=op["qr_token"])]])
await bot.send_photo(
chat_id, qr_png(op["qr_token"]), reply_markup=kb,
caption=(f"{order['title']} — {order['amount']:,} ₸\n\n"
"С телефона нажмите кнопку, с компьютера отсканируйте QR "
"в приложении Kaspi. QR действует 5 минут.").replace(",", " "))
@dp.message(Command("start"))
async def start(message: Message) -> None:
kb = InlineKeyboardMarkup(inline_keyboard=[
[InlineKeyboardButton(text=f"{t} — {p:,} ₸".replace(",", " "),
callback_data=f"buy:{code}")]
for code, (t, p) in PRODUCTS.items()])
await message.answer("Что берёте?", reply_markup=kb)
@dp.callback_query(F.data.startswith("buy:"))
async def buy(cb: CallbackQuery) -> None:
title, amount = PRODUCTS[cb.data.split(":", 1)[1]]
order = {"id": uuid.uuid4().hex[:12], "chat_id": cb.message.chat.id,
"title": title, "amount": amount, "status": "pending",
"attempt": 0, "op_id": None, "created": time.time()}
orders[order["id"]] = order
await cb.answer()
await send_payment(cb.message.chat.id, order)
@dp.callback_query(F.data.startswith("retry:"))
async def retry(cb: CallbackQuery) -> None:
order = orders.get(cb.data.split(":", 1)[1])
await cb.answer()
if order and order["status"] == "expired":
order["status"] = "pending"
await send_payment(order["chat_id"], order)
async def on_paid(order: dict) -> None:
if order["status"] == "paid": # вебхук и сверка могут прийти оба
return
order["status"] = "paid"
await bot.send_message(order["chat_id"],
f"Оплата получена: {order['title']}. Спасибо!")
# здесь выдаём товар: ссылку на курс, запись в календарь и т. п.
async def on_expired(order: dict) -> None:
if order["status"] != "pending":
return
order["status"] = "expired"
kb = InlineKeyboardMarkup(inline_keyboard=[
[InlineKeyboardButton(text="Новый QR", callback_data=f"retry:{order['id']}")]])
await bot.send_message(order["chat_id"], "Время QR вышло.", reply_markup=kb)
async def handle_event(event: dict) -> None:
kind, data = event["type"], event.get("data") or {}
if kind == "connection.session_expired":
await bot.send_message(ADMIN_CHAT_ID,
"Сессия Kaspi слетела: приём платежей стоит. "
"Войдите заново по SMS в кабинете Aitole.")
return
order = orders.get(by_op.get(data.get("id", "")))
if not order or data.get("id") != order["op_id"]:
return # чужая или старая попытка
if kind == "payment.paid" and data.get("status") == "paid":
await on_paid(order)
elif kind == "payment.expired":
await on_expired(order)
def signature_ok(raw: bytes, ts: str, got: str) -> bool:
mine = "sha256=" + hmac.new(WEBHOOK_SECRET, ts.encode() + b"." + raw,
hashlib.sha256).hexdigest()
if not hmac.compare_digest(mine, got):
return False
try:
return abs(time.time() - int(ts)) <= 300 # защита от повтора
except ValueError:
return False
async def aitole_webhook(request: web.Request) -> web.Response:
raw = await request.read() # сырые байты, до json.loads
if not signature_ok(raw,
request.headers.get("X-Webhook-Timestamp", ""),
request.headers.get("X-Webhook-Signature", "")):
return web.Response(status=400)
event = json.loads(raw)
if event["id"] not in seen_events: # доставка бывает двойной
seen_events.add(event["id"])
asyncio.create_task(handle_event(event)) # отвечаем сразу, работаем потом
return web.Response(status=200)
async def reconcile() -> None:
"""Страховка на случай, если вебхук не дошёл: раз в минуту
сверяем зависшие заказы по GET /v2/operations/{id}."""
while True:
await asyncio.sleep(60)
for order in list(orders.values()):
if order["status"] != "pending" or not order["op_id"]:
continue
if time.time() - order["created"] < 60:
continue
r = await http.get(f"/v2/operations/{order['op_id']}")
if r.status_code != 200:
continue
status = r.json()["status"]
if status == "paid":
await on_paid(order)
elif status == "expired":
await on_expired(order)
async def main() -> None:
app = web.Application()
app.router.add_post("/aitole/webhook", aitole_webhook)
runner = web.AppRunner(app)
await runner.setup()
await web.TCPSite(runner, "0.0.0.0", 8080).start()
asyncio.create_task(reconcile())
await dp.start_polling(bot)
if __name__ == "__main__":
asyncio.run(main())
KioskNews shows a cleaned-up reading view extracted from the publisher’s page — the original always lives on their site, not ours.