如何在 Aiogram 3 的机器人中通过 CryptoBot 接入支付功能(第 3 部分):带代码的 Crypto Pay API
在 aiogram 3 上通过 CryptoBot(Crypto Pay API)在 Telegram 机器人中接收加密货币:代币、账户创建、按钮验证支付以及带签名验证的 webhook。完整代码。
CryptoBot (@CryptoBot) —— Telegram 内广受欢迎的加密货币钱包。 通过其支付服务Crypto Pay,该机器人可以开具USDT、TON、BTC及其他币种的账单:买家可直接在Telegram中通过自己的CryptoBot钱包付款,款项将直接进入您的应用余额。
下文介绍的是基于 aiogram 3 开发的机器人,它能够生成账单、通过按钮验证付款,并通过 webhook 接收即时付款通知。
Crypto Pay 的支付流程
- 机器人通过 createInvoice 方法生成账单,并获取支付链接。
- 用户点击链接,在 @CryptoBot 中完成支付。
- 机器人通过以下两种方式之一获知付款情况:用户点击“我已付款”,机器人查询账单状态(getInvoices);或者 CryptoBot 主动向您的服务器发送通知(Webhook)。
- 当账单状态变为“已支付”(paid)时,机器人将发放商品。
步骤 1. 获取 Crypto Pay 令牌
- 打开 @CryptoBot → Crypto Pay → 创建应用。
- 设置名称——机器人将生成该应用的 API 令牌。
- 将令牌保存到环境变量 CRYPTO_PAY_TOKEN 中。请勿公开该令牌:它可用于管理您应用的账单。
步骤 2. 安装 aiogram
pip install -U aiogramaiohttp(我们将通过它调用 API)会随 aiogram 一起安装。
第 3 步. Crypto Pay 客户端
所有发往 Crypto Pay 的请求均发送至 https://pay.crypt.bot/api/<方法>,并在请求头中包含 Crypto-Pay-API-Token。将这个小型客户端保存到 cryptopay.py 文件中:
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第4步:带支付功能和验证按钮的机器人
bot.py 文件:
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向机器人发送 /buy,在 @CryptoBot 中支付账单,然后点击“我已支付”。
第 5 步:通过 Webhook 接收即时通知
“我已支付”按钮始终有效,但用户可能不会点击它。 为了让机器人自动检测到付款,请启用 Webhook:@CryptoBot → Crypto Pay → 您的应用 → Webhooks → 填写类似 https://ваш-домен/webhooks/cryptobot 的地址。需要配备域名和 HTTPS 的服务器。
CryptoBot 会对每条通知进行签名:在 crypto-pay-api-signature 头部中包含请求正文的 HMAC-SHA256 值,其密钥即为您令牌的 SHA-256 值。 没有正确签名的通知可能由任何人发送,因此我们会拒绝此类请求。
请在 bot.py 中添加 Webhook 处理程序,并替换 main 函数——通知 Web 服务器将随机器人一同启动。请将新的导入语句移至文件开头:
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()Web 服务器仅监听 127.0.0.1:8080,而对外访问则由配备 HTTPS 证书的 nginx(或 Caddy)负责——请将路径 /webhooks/cryptobot 代理到该地址。 deliver 函数对按钮和 Webhook 均适用,且不会重复发送商品。
以卢布或美元支付
如果机器人中的价格以法币显示,请创建一个 currency_type="fiat" 的账单:CryptoBot 会根据汇率自动换算金额,用户可选择支付的币种。
invoice = await crypto.call(
"createInvoice",
currency_type="fiat",
fiat="RUB",
amount="500",
accepted_assets="USDT,TON",
description="Доступ к курсу",
payload=str(message.from_user.id),
)在此账单中,请检查“fiat”和“amount”字段——金额应以卢布为单位,而非以代币为单位。
安全性:运行前需检查的事项
- 令牌 — 仅存放在环境变量或 .env 文件中,切勿在代码和代码库中直接暴露。
- paid 状态 — 这是已付款的唯一标志。切勿在未通过 API 验证状态的情况下,仅凭点击按钮就发货。
- 金额和币种 — 务必与订单核对:这样可以防止有人将金额较低的已付款账单冒充成应付账单。
- 账单归属 — payload 必须与买家或其订单号一致。
- 仅限一次出货 — 将已处理的 invoice_id 存储在数据库中。示例中内存中的集合会在机器人重启时被清空。
- Webhook签名 — 务必进行验证,比较时请使用 hmac.compare_digest。
常见错误
- UNAUTHORIZED — 令牌无效,或将测试网令牌发送到了主网 API 地址(反之亦然)。
- 金额为数字。 请以字符串形式传递 amount:"5"、"2.5"。
- 未收到 webhook 通知。 该地址必须通过 HTTPS 对外开放,且服务器应返回 200 状态码。
- 每个请求仅允许一个 ClientSession。 对于教程示例来说这是正常的,但在高负载情况下,请在机器人启动时创建一个会话,并在停止时关闭它。
其他支付方式包括 Telegram Stars 和 xRocket; 所有选项的对比请参阅文章 《如何在 Telegram 机器人中接收付款》。
常见问题
通过CryptoBot接收付款是否需要验证?
在Crypto Pay中创建应用只需拥有@CryptoBot账户即可。具体要求和限额请查阅该服务本身——这些内容可能会有所变动。
可以使用哪些币种开具账单?
CryptoBot支持的币种:USDT、TON、BTC、ETH等。可通过getCurrencies方法获取可用币种列表。
如何免费测试支付功能?
通过测试网:@CryptoTestnetBot 机器人及地址 https://testnet-pay.crypt.bot/api。该环境使用专属代币和虚拟币。
是否可以不使用 webhook?
可以。仅需“我已付款”按钮即可触发交易,而添加 webhook 的目的是让机器人即使在用户忘记点击按钮时也能立即交付商品。
款项会打到哪里?
会打入您在 @CryptoBot 中的应用余额。从那里可以将资金提现到自己的钱包。



