В Казахстане Telegram — основной канал для продаж услуг, инфопродуктов, цифровых товаров. Если у вас бот, в нём логично принимать оплату прямо в чате — без редиректов на сайт. Эта статья — про то, как сделать это через Kaspi Pay API на Python + aiogram. Все примеры боевые, копируются и работают.
Архитектура
Что происходит, когда клиент жмёт «Оплатить» в боте:
- Бот через Pay Bot API создаёт QR-операцию.
- Бот отправляет клиенту QR-картинку и deep-link «Открыть Kaspi».
- Клиент платит в приложении Kaspi (либо сканирует QR с другого устройства, либо тапает по deep-link на телефоне).
- Pay Bot получает уведомление от Kaspi и отправляет webhook вашему серверу.
- Сервер обновляет состояние и шлёт клиенту через бот сообщение «Оплачено, спасибо!».
Никаких терминалов, никаких карт, никакого 3-D Secure SMS. Полностью внутри Telegram-флоу.
Что понадобится
- Telegram-бот (создаётся через @BotFather).
- Аккаунт в Pay Bot и API-ключ
kp_live_...илиkp_test_.... - Сервер с публичным URL для webhook'ов (Render, Railway, fly.io, VPS).
- Python 3.11+ и
aiogram 3.
Минимальный код бота
Стартовый каркас:
import asyncio
import os
import uuid
import aiohttp
from aiogram import Bot, Dispatcher, F
from aiogram.types import Message, CallbackQuery, InlineKeyboardButton, InlineKeyboardMarkup
from aiogram.filters import CommandStart
BOT_TOKEN = os.environ["BOT_TOKEN"]
KASPI_KEY = os.environ["KASPI_KEY"]
API = "https://api.paybot.kz"
bot = Bot(BOT_TOKEN)
dp = Dispatcher()
# Простой in-memory storage. В проде — Redis или БД.
ORDERS: dict[str, dict] = {}
@dp.message(CommandStart())
async def start(m: Message):
kb = InlineKeyboardMarkup(inline_keyboard=[
[InlineKeyboardButton(text="Курс «Python для начинающих» — 15 000 ₸", callback_data="buy:course1")],
[InlineKeyboardButton(text="Подписка на месяц — 5 000 ₸", callback_data="buy:sub1")],
])
await m.answer("Что покупаем?", reply_markup=kb)
@dp.callback_query(F.data.startswith("buy:"))
async def buy(cb: CallbackQuery):
product = cb.data.split(":", 1)[1]
prices = {"course1": 15000, "sub1": 5000}
amount = prices[product]
async with aiohttp.ClientSession() as s:
resp = await s.post(
f"{API}/v1/qr/create",
headers={
"X-API-Key": KASPI_KEY,
"Idempotency-Key": f"tg:{cb.from_user.id}:{product}:{uuid.uuid4()}",
},
json={"amount": amount},
)
data = await resp.json()
op_id = data["operation_id"]
ORDERS[op_id] = {
"user_id": cb.from_user.id,
"product": product,
"amount": amount,
}
kb = InlineKeyboardMarkup(inline_keyboard=[
[InlineKeyboardButton(text="Открыть Kaspi", url=data["deep_link"])],
[InlineKeyboardButton(text="Проверить статус", callback_data=f"check:{op_id}")],
])
await cb.message.answer_photo(
data["qr_url"],
caption=f"Сумма: {amount} ₸\nОтсканируйте QR в Kaspi или нажмите «Открыть Kaspi».",
reply_markup=kb,
)
@dp.callback_query(F.data.startswith("check:"))
async def check(cb: CallbackQuery):
op_id = cb.data.split(":", 1)[1]
async with aiohttp.ClientSession() as s:
resp = await s.get(
f"{API}/v1/qr/status/{op_id}",
headers={"X-API-Key": KASPI_KEY},
)
data = await resp.json()
status = data["status"]
if status == "paid":
await cb.message.answer("Оплачено! Спасибо. Доступ к товару отправлен в личку.")
# выдача товара здесь
elif status == "pending":
await cb.answer("Платёж ещё не пришёл. Подождите несколько секунд.", show_alert=True)
else:
await cb.message.answer(f"Статус: {status}. Попробуйте создать платёж заново.")
async def main():
await dp.start_polling(bot)
if __name__ == "__main__":
asyncio.run(main())
Запускаете — бот работает. Клиент может покупать, опрос статуса работает. Но это медленно и неаккуратно. Дальше — улучшения.
Подключаем webhook вместо опроса
Опрос статуса работает, но клиент должен жать «Проверить» сам. С webhook'ом сообщение прилетит автоматически.
Сервер на FastAPI:
import os
import hmac
import hashlib
import asyncio
from fastapi import FastAPI, Request, HTTPException
from aiogram import Bot
BOT_TOKEN = os.environ["BOT_TOKEN"]
WEBHOOK_SECRET = os.environ["KASPI_WEBHOOK_SECRET"].encode()
app = FastAPI()
bot = Bot(BOT_TOKEN)
# Та же in-memory структура, что и в боте. В проде — общее хранилище (Redis).
from main import ORDERS # упрощение для примера
@app.post("/kaspi-webhook")
async def kaspi(request: Request):
raw = await request.body()
sig = request.headers.get("X-Kaspi-Signature", "")
expected = hmac.new(WEBHOOK_SECRET, raw, hashlib.sha256).hexdigest()
if not hmac.compare_digest(f"sha256={expected}", sig):
raise HTTPException(401, "invalid signature")
event = await request.json()
if event["event"] == "payment.completed":
op_id = event["data"]["operation_id"]
order = ORDERS.get(op_id)
if order:
await bot.send_message(
order["user_id"],
f"Платёж на {order['amount']} ₸ получен. Высылаю товар.",
)
# выдача товара здесь
return {"ok": True}
В кабинете Pay Bot подписываемся на payment.completed с URL https://your-server.com/kaspi-webhook. Готово.
Теперь после оплаты пользователь получает сообщение через 3–5 секунд, без необходимости нажимать «Проверить».
Тонкости, на которые стоит обратить внимание
Persistence
ORDERS в памяти умрёт при рестарте. В проде используйте Redis или PostgreSQL:
import redis.asyncio as redis
r = redis.from_url(os.environ["REDIS_URL"])
async def save_order(op_id, data):
await r.hset(f"order:{op_id}", mapping=data)
async def get_order(op_id):
return await r.hgetall(f"order:{op_id}")
Защита от двойных оплат
Если пользователь нажал «Купить» два раза подряд — у вас два QR на одну и ту же покупку. Решение:
# Перед созданием платежа
existing = await r.get(f"pending:{user_id}:{product}")
if existing:
return await show_existing_qr(cb, existing)
Или просто блокируйте кнопку после первого клика средствами aiogram (cb.message.edit_reply_markup(reply_markup=None)).
Idempotency-Key
В моём примере выше я использовал uuid.uuid4() в ключе:
f"tg:{cb.from_user.id}:{product}:{uuid.uuid4()}"
Это неправильно для ретраев — каждый вызов даст новый ключ. Если хотите идемпотентность, ключ должен быть детерминирован:
# Один платёж на пользователя+продукт+сутки
day = datetime.utcnow().date().isoformat()
key = f"tg:{cb.from_user.id}:{product}:{day}"
Или сгенерируйте UUID при первом нажатии и сохраните его в БД до отправки.
Webhook security
Кроме HMAC обязательно проверяйте timestamp (защита от replay) и логируйте все входящие — это спасёт от голословных «я заплатил, а товар не пришёл».
Подробно — в статье о webhooks.
Готовый чек-лист продакшна
Перед запуском бота с реальными платежами проверьте:
- [ ] Хранилище заказов — Redis или БД, не in-memory.
- [ ] HMAC-подпись на webhook'ах проверяется.
- [ ]
event.idдедуплицируется. - [ ]
Idempotency-Keyдетерминирован. - [ ] Используется live-ключ только в проде; в dev — Sandbox.
- [ ] Логи всех платежей записываются хотя бы в файл.
- [ ] Есть
/adminкоманда для ручной проверки статуса через/v1/qr/status/{op_id}. - [ ] Если 5xx в обработчике — бот не падает, ошибка отправляется в Sentry.
Альтернатива: hosted checkout
Если не хочется писать обработчик QR, есть проще — Hosted Checkout. Бот создаёт «платёжную ссылку» через API:
resp = await s.post(
f"{API}/v2/payment-links",
headers={"Authorization": f"Bearer {jwt}"},
json={
"amount": 15000,
"description": "Курс Python",
"success_url": "https://t.me/yourbot?start=paid",
},
)
link = (await resp.json())["url"]
await cb.message.answer(f"Оплатить: {link}")
Pay Bot сам показывает страницу с QR, обрабатывает оплату, шлёт webhook и редиректит в ваш бот по success_url. Подходит, если не хочется возиться с inline-кнопками и картинками.
Дальше
Бот, который принимает оплату — это базовая логика. Развивать можно в разные стороны:
- Каталог через FSM — пользователь листает товары, добавляет в корзину, потом оплачивает одной операцией.
- Подписки — фоновый job шлёт счета за неделю до истечения.
- Партнёрская программа — реферальный код в
metadata, начисление бонусов через webhook.
API Pay Bot тут не ограничивает — все эти схемы делаются на уровне вашего бота. Если нужна консультация под конкретный сценарий — пишите в @paybotkz_bot или оставляйте заявку, разберём вашу задачу за один звонок.
Готовы попробовать Pay Bot?
Зарегистрируйтесь в кабинете — Sandbox доступен сразу. Или оставьте заявку, и менеджер настроит подключение Kaspi за вас.