如何在 Aiogram 3 的机器人中通过 xRocket 接入支付功能：Rocket Pay API 及代码示例

通过 xRocket（Rocket Pay）在 aiogram 3 的 Telegram 机器人中接收加密货币：API 密钥、收款账户、状态查询和安全的 Webhook。附带说明的完整可运行代码。

更新于 2026/11/02阅读 5 分钟

xRocket (@xRocket) —— 是一款集成在Telegram内的加密货币钱包和交易所。其支付服务Rocket Pay允许机器人开具TON、USDT及生态系统内其他代币的账单：买家在@xRocket上支付账单，款项将直接进入您的应用余额。

下文展示了一个基于 aiogram 3 的机器人，它能够生成账单、通过按钮验证付款，并通过 webhook 接收来自 xRocket 的通知。

Rocket Pay 支付流程

机器人通过 POST /tg-invoices 请求生成账单，并获取支付链接。
用户点击链接，在 @xRocket 完成支付。
机器人通过 GET /tg-invoices/<id> 请求查询账单状态——这可能是在用户点击“我已付款”按钮后，或在收到 xRocket 的通知后触发。
账单已付清——机器人交付商品。

步骤 1. 获取 API 密钥

打开 @xRocket → Rocket Pay → 创建应用（机器人菜单中的选项名称可能略有不同）。
在应用设置中获取 API 密钥。
将密钥保存到环境变量 XROCKET_API_KEY 中。该密钥可用于管理应用的账户和资金——请像保管密码一样妥善保管。

步骤 2. 安装 aiogram

pip install -U aiogram

第3步. Rocket Pay客户端

API可通过地址 https://pay.xrocket.exchange 访问，密钥通过 Rocket-Pay-Key 头部传递。响应中始终包含 success 字段，数据则位于 data 字段中。xrocket.py 文件：

import aiohttp

API_URL = "https://pay.xrocket.exchange"


class XRocketPay:
    # Минимальный клиент Rocket Pay API (@xRocket) — счета tg-invoices

    def __init__(self, api_key: str, api_url: str = API_URL) -> None:
        self.headers = {"Rocket-Pay-Key": api_key}
        self.api_url = api_url

    async def _request(self, method: str, path: str, body: dict | None = None) -> dict:
        async with aiohttp.ClientSession(headers=self.headers) as session:
            async with session.request(method, f"{self.api_url}{path}", json=body) as resp:
                data = await resp.json()
        if not data.get("success"):
            raise RuntimeError(f"xRocket: {data.get('message') or data.get('errors')}")
        return data["data"]

    async def create_invoice(self, amount: float, currency: str, description: str, payload: str) -> dict:
        return await self._request("POST", "/tg-invoices", {
            "amount": amount,
            "currency": currency,     # монета: USDT, TONCOIN...
            "numPayments": 1,         # счёт можно оплатить один раз
            "description": description,
            "payload": payload,       # ваша метка заказа, вернётся вместе со счётом
            "expiredIn": 3600,        # счёт действует 1 час
        })

    async def get_invoice(self, invoice_id: int) -> dict:
        return await self._request("GET", f"/tg-invoices/{invoice_id}")

第4步：带支付功能和验证按钮的机器人

bot.py 文件：

import asyncio
import os

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

from xrocket import XRocketPay

PRICE = 5
CURRENCY = "USDT"
PAID_STATUSES = {"paid", "completed"}  # счёт с numPayments=1 после оплаты получает статус completed

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


def is_paid(invoice: dict) -> bool:
    return str(invoice.get("status", "")).lower() in PAID_STATUSES


async def deliver(bot: Bot, invoice: dict) -> bool:
    # Выдаёт товар по оплаченному счёту — строго один раз
    if not is_paid(invoice) or invoice["id"] in delivered:
        return False
    if invoice.get("currency") != CURRENCY or float(invoice.get("amount", 0)) != PRICE:
        return False
    delivered.add(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 rocket.create_invoice(
        amount=PRICE, currency=CURRENCY,
        description="Доступ к курсу",
        payload=str(message.from_user.id),  # кому выдать товар после оплаты
    )
    keyboard = InlineKeyboardMarkup(inline_keyboard=[
        [InlineKeyboardButton(text=f"💳 Оплатить {PRICE} {CURRENCY}", url=invoice["link"])],
        [InlineKeyboardButton(text="✅ Я оплатил", callback_data=f"xr:{invoice['id']}")],
    ])
    await message.answer("Счёт на оплату действует 1 час:", reply_markup=keyboard)


@router.callback_query(F.data.startswith("xr:"))
async def check(callback: CallbackQuery, bot: Bot) -> None:
    invoice_id = int(callback.data.split(":")[1])
    try:
        invoice = await rocket.get_invoice(invoice_id)
    except RuntimeError:
        invoice = None
    # Счёт должен принадлежать тому, кто нажал кнопку
    if not invoice or str(invoice.get("payload")) != str(callback.from_user.id):
        await callback.answer("Счёт не найден", show_alert=True)
        return
    if not is_paid(invoice):
        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 XROCKET_API_KEY="ключ_Rocket_Pay"
python bot.py

向机器人发送 /buy 指令，在 @xRocket 支付账单，然后点击“我已支付”。

第 5 步. 支付通知（Webhook）

为了让机器人能在付款后立即发放商品，请在 Rocket Pay 应用的设置中指定 Webhook 地址，例如 https://ваш-домен/webhooks/xrocket。需要一个带有域名和 HTTPS 的服务器。

主要原则：不要轻信通知内容。我们仅从通知中提取账单编号，而状态、金额和有效载荷则需使用自己的 API 密钥向 xRocket 查询。伪造的请求不会有任何效果——API 会返回“账单未支付”的响应。

在 bot.py 中添加一个处理程序，并替换 main 函数。将导入语句移至文件开头：

from aiohttp import web


async def xrocket_webhook(request: web.Request) -> web.Response:
    try:
        update = await request.json()
        invoice_id = int(update["data"]["id"])
        # Статус узнаём у xRocket своим ключом, а не из тела уведомления
        invoice = await rocket.get_invoice(invoice_id)
    except (ValueError, KeyError, TypeError, RuntimeError):
        return web.Response(status=400)
    await deliver(request.app["bot"], invoice)
    return web.Response(text="ok")


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/xrocket", xrocket_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，而对外访问则由 nginx（或 Caddy）通过 HTTPS 开放——请将路径 /webhooks/xrocket 代理到该地址。 “我已付款”按钮和 Webhook 都使用同一个 deliver 函数，因此商品不会被重复发放。

安全性：运行前需检查的事项

API密钥 — 仅存放在环境变量或 .env 文件中，切勿写入代码或存入代码库。
状态 — 请通过API查询。无论是点击按钮还是webhook正文，均不能作为付款凭证。
金额和币种 — 请与订单信息核对。
账户归属 — payload 必须与买家或其订单号一致。
一次性发放 — 将已处理的账户 ID 存储在数据库中：内存中的集合在重启时会被清空。

常见错误

代币名称错误。 在 xRocket 中，TON 被传递为 TONCOIN。币种列表及最低金额请参阅 Rocket Pay API 文档。
金额低于最低限额。 每种币种都有对应的最低账单金额——API 将返回错误。
预期状态为 paid，但实际返回 completed。 单笔支付的账单在付款后可能会转为“completed”状态——请像示例中那样同时检查这两个状态。
未收到 Webhook 通知。 该地址必须通过 HTTPS 在互联网上可访问，且服务器应返回 200 状态码。

其他支付方式包括 Telegram Stars 和 CryptoBot；所有选项的对比请参阅文章《如何在 Telegram 机器人中接收付款》。

常见问题

xRocket 与 CryptoBot 有什么区别？

两者都是 Telegram 内的加密货币钱包，均提供用于接收支付的 API。它们在支持的币种、手续费和界面方面有所不同。许多商家同时接入这两款服务，以便买家能通过自己习惯使用的钱包进行支付。

应使用哪种币种开具账单？

通常使用USDT——其汇率较为稳定。TON应标注为TONCOIN。

xRocket 的 Webhook 可靠吗？

请仅将其作为提示：获取账单编号后，使用您的密钥通过 API 查询状态。这样，虚假通知就无法造成任何影响。

款项会打到哪里？

会进入 @xRocket 中的 Rocket Pay 应用余额。从那里可以将资金提现到自己的钱包。

如何在 Aiogram 3 的机器人中通过 xRocket 接入支付功能：Rocket Pay API 及代码示例

Rocket Pay 支付流程

步骤 1. 获取 API 密钥

步骤 2. 安装 aiogram

第3步. Rocket Pay客户端

第4步：带支付功能和验证按钮的机器人

第 5 步. 支付通知（Webhook）

安全性：运行前需检查的事项

常见错误

常见问题

相关现成解决方案

⭐️ “自动购买星星”网页应用

📲 AI视频剪辑

✍🏻 AI自动评论 2.0

🎯 狙击手礼物