Блог
Telegram бот Kaspi Pay API aiogram Python

Как принимать оплату через Kaspi прямо в Telegram-боте

Полный гайд по приёму платежей через Kaspi Pay API в Telegram-боте на Python. С нуля до рабочего бота: код, webhook, обработка статусов, готовые шаблоны.

Pay Bot 18 апреля 2026 г. 3 мин чтения

В Казахстане Telegram — основной канал для продаж услуг, инфопродуктов, цифровых товаров. Если у вас бот, в нём логично принимать оплату прямо в чате — без редиректов на сайт. Эта статья — про то, как сделать это через Kaspi Pay API на Python + aiogram. Все примеры боевые, копируются и работают.

Архитектура

Что происходит, когда клиент жмёт «Оплатить» в боте:

  1. Бот через Pay Bot API создаёт QR-операцию.
  2. Бот отправляет клиенту QR-картинку и deep-link «Открыть Kaspi».
  3. Клиент платит в приложении Kaspi (либо сканирует QR с другого устройства, либо тапает по deep-link на телефоне).
  4. Pay Bot получает уведомление от Kaspi и отправляет webhook вашему серверу.
  5. Сервер обновляет состояние и шлёт клиенту через бот сообщение «Оплачено, спасибо!».

Никаких терминалов, никаких карт, никакого 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 за вас.