How to Set Up CryptoBot Payments in an Aiogram 3 Bot: Crypto Pay API with Code

pip install -U aiogram

import aiohttp

API_URL = "https://pay.crypt.bot/api"  # тестовая сеть: https://testnet-pay.crypt.bot/api


class CryptoPay:
    # Минимальный клиент Crypto Pay API (@CryptoBot)

    def __init__(self, token: str, api_url: str = API_URL) -> None:
        self.token = token
        self.api_url = api_url

    async def call(self, method: str, **params) -> dict:
        headers = {"Crypto-Pay-API-Token": self.token}
        async with aiohttp.ClientSession() as session:
            async with session.post(f"{self.api_url}/{method}", json=params, headers=headers) as resp:
                data = await resp.json()
        if not data.get("ok"):
            raise RuntimeError(f"Crypto Pay: {data.get('error')}")
        return data["result"]

    async def create_invoice(self, amount: str, asset: str, description: str, payload: str) -> dict:
        return await self.call(
            "createInvoice",
            asset=asset,          # монета: USDT, TON, BTC...
            amount=amount,        # сумма строкой, например "5" или "2.5"
            description=description,
            payload=payload,      # ваша метка заказа, вернётся вместе со счётом
            expires_in=3600,      # счёт действует 1 час
        )

    async def get_invoice(self, invoice_id: int) -> dict | None:
        result = await self.call("getInvoices", invoice_ids=str(invoice_id))
        items = result.get("items", [])
        return items[0] if items else None

import asyncio
import os
from decimal import Decimal

from aiogram import Bot, Dispatcher, F, Router
from aiogram.filters import Command
from aiogram.types import CallbackQuery, InlineKeyboardButton, InlineKeyboardMarkup, Message

from cryptopay import CryptoPay

PRICE = Decimal("5")
ASSET = "USDT"

router = Router()
crypto = CryptoPay(os.environ["CRYPTO_PAY_TOKEN"])
delivered: set[int] = set()  # счета, по которым товар уже выдан (в реальном боте — база данных)


async def deliver(bot: Bot, invoice: dict) -> bool:
    # Выдаёт товар по оплаченному счёту. Вызывается и из кнопки, и из вебхука, поэтому выдача — строго один раз
    if invoice.get("status") != "paid" or invoice["invoice_id"] in delivered:
        return False
    if invoice.get("asset") != ASSET or Decimal(invoice["amount"]) != PRICE:
        return False
    delivered.add(invoice["invoice_id"])
    user_id = int(invoice["payload"])
    # Здесь выдайте товар: откройте доступ, отправьте файл, начислите баланс
    await bot.send_message(user_id, "Оплата получена! Доступ открыт ✅")
    return True


@router.message(Command("buy"))
async def buy(message: Message) -> None:
    invoice = await crypto.create_invoice(
        amount=str(PRICE), asset=ASSET,
        description="Доступ к курсу",
        payload=str(message.from_user.id),  # кому выдать товар после оплаты
    )
    keyboard = InlineKeyboardMarkup(inline_keyboard=[
        [InlineKeyboardButton(text=f"💳 Оплатить {PRICE} {ASSET}", url=invoice["bot_invoice_url"])],
        [InlineKeyboardButton(text="✅ Я оплатил", callback_data=f"check:{invoice['invoice_id']}")],
    ])
    await message.answer("Счёт на оплату действует 1 час:", reply_markup=keyboard)


@router.callback_query(F.data.startswith("check:"))
async def check(callback: CallbackQuery, bot: Bot) -> None:
    invoice_id = int(callback.data.split(":")[1])
    invoice = await crypto.get_invoice(invoice_id)
    # Счёт должен принадлежать тому, кто нажал кнопку
    if not invoice or invoice.get("payload") != str(callback.from_user.id):
        await callback.answer("Счёт не найден", show_alert=True)
        return
    if invoice["status"] == "expired":
        await callback.answer("Счёт просрочен. Создайте новый: /buy", show_alert=True)
        return
    if invoice["status"] != "paid":
        await callback.answer("Оплата пока не поступила", show_alert=True)
        return
    await deliver(bot, invoice)
    await callback.message.edit_text("Счёт оплачен ✅")
    await callback.answer()


async def main() -> None:
    bot = Bot(token=os.environ["BOT_TOKEN"])
    dp = Dispatcher()
    dp.include_router(router)
    await dp.start_polling(bot)


if __name__ == "__main__":
    asyncio.run(main())

export BOT_TOKEN="токен_от_BotFather"
export CRYPTO_PAY_TOKEN="токен_Crypto_Pay"
python bot.py

import hashlib
import hmac
import json

from aiohttp import web

TOKEN = os.environ["CRYPTO_PAY_TOKEN"]


def check_signature(body: bytes, signature: str) -> bool:
    secret = hashlib.sha256(TOKEN.encode()).digest()  # ключ подписи — SHA-256 от токена
    expected = hmac.new(secret, body, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, signature)


async def cryptobot_webhook(request: web.Request) -> web.Response:
    body = await request.read()
    if not check_signature(body, request.headers.get("crypto-pay-api-signature", "")):
        return web.Response(status=401)
    update = json.loads(body)
    if update.get("update_type") == "invoice_paid":
        await deliver(request.app["bot"], update["payload"])  # payload здесь — сам счёт
    return web.Response(text="ok")  # отвечаем 200, иначе CryptoBot повторит уведомление


async def main() -> None:
    bot = Bot(token=os.environ["BOT_TOKEN"])
    dp = Dispatcher()
    dp.include_router(router)

    app = web.Application()
    app["bot"] = bot
    app.router.add_post("/webhooks/cryptobot", cryptobot_webhook)
    runner = web.AppRunner(app)
    await runner.setup()
    await web.TCPSite(runner, "127.0.0.1", 8080).start()  # снаружи — через nginx с HTTPS

    try:
        await dp.start_polling(bot)
    finally:
        await runner.cleanup()

invoice = await crypto.call(
    "createInvoice",
    currency_type="fiat",
    fiat="RUB",
    amount="500",
    accepted_assets="USDT,TON",
    description="Доступ к курсу",
    payload=str(message.from_user.id),
)