Usage
Quick Start
Import the client, core models, and error types:
import os
from cryptobot import CryptoBotClient
from cryptobot.errors import CryptoBotError
from cryptobot.models import Asset, ButtonName, Status
Create a client instance with your API token:
client = CryptoBotClient(
api_token=os.environ["CRYPTOBOT_API_TOKEN"],
is_mainnet=True,
timeout=5.0,
)
Async quick start:
from cryptobot import AsyncCryptoBotClient
async def run():
async with AsyncCryptoBotClient(api_token=os.environ["CRYPTOBOT_API_TOKEN"], max_retries=2) as client:
app = await client.get_me()
print(app.name)
Basic Operations
Getting App Information
Get basic information about your app:
app = client.get_me()
print(f"App ID: {app.app_id}")
print(f"App Name: {app.name}")
print(f"Bot Username: {app.payment_processing_bot_username}")
Creating an Invoice
Create a simple invoice for cryptocurrency payment:
invoice = client.create_invoice(
asset=Asset.USDT,
amount=5.25,
description="Coffee order #42"
)
print(f"Invoice URL: {invoice.bot_invoice_url}")
print(f"Invoice ID: {invoice.invoice_id}")
Creating an Invoice with Custom Options
Create an invoice with additional parameters:
invoice = client.create_invoice(
asset=Asset.BTC,
amount=0.001,
description="Premium service subscription",
paid_btn_name=ButtonName.callback, # Enum value for paid invoice button
paid_btn_url="https://example.com/success", # URL for success callback
payload="user_123_premium", # Custom payload for tracking
allow_comments=True, # Allow user comments
allow_anonymous=False, # Require user identification
expires_in=3600 # Expire after 1 hour
)
Making Transfers
Transfer cryptocurrency to another user:
try:
transfer = client.transfer(
user_id=12345,
asset=Asset.TON,
amount=0.5,
spend_id="transfer_001", # Unique spend ID to prevent duplicates
comment="Payment for services"
)
print(f"Transfer completed: {transfer.transfer_id}")
except CryptoBotError as e:
print(f"Transfer failed: {e.code} - {e.name}")
Checking Balances
Get your current balances:
balances = client.get_balances()
for balance in balances:
print(f"{balance.currency_code}: {balance.available}")
Getting Exchange Rates
Retrieve current exchange rates:
rates = client.get_exchange_rates()
for rate in rates:
print(f"1 {rate.source} = {rate.rate} {rate.target}")
You can also filter for specific currencies:
# Find USD exchange rate for Bitcoin
btc_usd_rate = next((r for r in rates if r.source == "BTC" and r.target == "USD"), None)
if btc_usd_rate:
print(f"1 BTC = ${btc_usd_rate.rate} USD")
Getting Supported Currencies
Get information about all supported currencies:
currencies = client.get_currencies()
for currency in currencies:
print(f"{currency.name} ({currency.code})")
print(f" Blockchain: {currency.is_blockchain}")
print(f" Stablecoin: {currency.is_stablecoin}")
print(f" Decimals: {currency.decimals}")
if currency.url:
print(f" URL: {currency.url}")
Filter by type:
# Get only blockchain currencies
blockchain_currencies = [c for c in currencies if c.is_blockchain]
# Get only stablecoins
stablecoins = [c for c in currencies if c.is_stablecoin]
# Get only fiat currencies
fiat_currencies = [c for c in currencies if c.is_fiat]
Getting Invoice Information
Retrieve information about existing invoices:
# Get all invoices
invoices = client.get_invoices()
# Get invoices with filters
paid_invoices = client.get_invoices(
asset=Asset.USDT,
invoice_ids=[123, 456],
status=Status.paid,
offset=0,
count=50
)
invoice_ids accepts either a comma-separated string ("123,456") or list[int] ([123, 456]).
count is validated between 1 and 1000.
Async equivalent:
import os
from cryptobot import AsyncCryptoBotClient
from cryptobot.models import Asset, Status
async with AsyncCryptoBotClient(api_token=os.environ["CRYPTOBOT_API_TOKEN"]) as async_client:
paid_invoices = await async_client.get_invoices(
asset=Asset.USDT,
invoice_ids=[123, 456],
status=Status.paid,
count=50,
)
Deleting an Invoice
Delete an active invoice that is no longer needed:
deleted = client.delete_invoice(invoice_id=12345)
print("Deleted:", deleted)
Retrieving Transfers
List outgoing transfers with optional filters:
# Get all transfers
transfers = client.get_transfers()
# Filter by asset or specific IDs
transfers = client.get_transfers(asset=Asset.TON, count=50)
transfers = client.get_transfers(transfer_ids=[100, 101, 102])
# Look up by spend_id
transfers = client.get_transfers(spend_id="reward_2026_02_10_user_123456789")
Crypto Checks
Create a check that any Telegram user (or a pinned user) can activate:
from cryptobot.models import Asset
# Create an open check
check = client.create_check(asset=Asset.USDT, amount=1.0)
print(check.check_id, check.bot_check_url)
# Pin a check to a specific user
check = client.create_check(asset=Asset.TON, amount=0.25, pin_to_user_id=123456789)
Retrieve and manage checks:
# List active checks
checks = client.get_checks(asset=Asset.USDT, status="active")
# Get specific checks by ID
checks = client.get_checks(check_ids=[10, 11, 12])
# Delete a check
client.delete_check(check_id=checks[0].check_id)
App Statistics
Get aggregated statistics for your app:
stats = client.get_stats(
start_at="2026-01-01T00:00:00Z",
end_at="2026-03-01T00:00:00Z",
)
print(f"Volume: {stats.volume}")
print(f"Unique users: {stats.unique_users_count}")
print(f"Paid invoices: {stats.paid_invoice_count}")
print(f"Conversion: {stats.conversion}")
Both start_at and end_at are optional ISO 8601 strings. When omitted, start_at defaults to 24 hours ago and end_at defaults to now.
Environment Configuration
Testnet vs Mainnet
By default, the client uses the mainnet environment. To target the testnet (with a testnet token), set is_mainnet=False:
client = CryptoBotClient(os.environ["CRYPTOBOT_TESTNET_TOKEN"], is_mainnet=False)
Custom Timeout
Configure request timeout (default is 5 seconds):
client = CryptoBotClient(os.environ["CRYPTOBOT_API_TOKEN"], timeout=60)
Built-in Retries
Retry behavior is optional and disabled by default (max_retries=0).
client = CryptoBotClient(
api_token=os.environ["CRYPTOBOT_API_TOKEN"],
timeout=30.0,
max_retries=3,
retry_backoff=0.5,
retryable_status_codes={429, 500, 502, 503, 504},
)
When present, the Retry-After response header is respected automatically before the next attempt.
Swap Incoming Payments
Automatically convert payments to a different asset by using swap_to:
invoice = client.create_invoice(
asset=Asset.TON,
amount=20,
description="Swap TON to USDT on receipt",
swap_to="USDT",
)
print(invoice.bot_invoice_url)
Error Handling
All API errors raise cryptobot.errors.CryptoBotError:
from cryptobot.errors import CryptoBotError
try:
client.transfer(user_id=12345, asset=Asset.BTC, amount=10, spend_id="test")
except CryptoBotError as exc:
print(f"Error code: {exc.code}")
print(f"Error name: {exc.name}")
Webhook Integration
Using the Built-in Webhook Server
CryptoBot Python includes a FastAPI-based webhook listener for handling payment notifications:
import os
from cryptobot.webhook import InMemoryReplayKeyStore, Listener
def handle_webhook(headers, data):
if data.get("update_type") == "invoice_paid":
payload = data.get("payload", {})
print("Paid invoice:", payload.get("invoice_id"))
listener = Listener(
host="0.0.0.0",
callback=handle_webhook,
api_token=os.environ["CRYPTOBOT_API_TOKEN"],
replay_store=InMemoryReplayKeyStore(),
replay_ttl_seconds=3600,
port=2203,
url="/webhook",
log_level="info",
)
listener.listen()
The listener automatically verifies signatures, can reject duplicate replayed webhook payloads, and provides a startup banner.
callback can be synchronous (def) or asynchronous (async def).
For custom replay de-duplication keys, provide replay_key_resolver:
def replay_key_resolver(data, raw_body, headers):
payload = data.get("payload", {})
invoice_id = payload.get("invoice_id")
if invoice_id is not None:
return f"invoice_paid:{invoice_id}"
return None
Custom Webhook Handler
You can also create your own webhook handler and verify signatures with check_signature:
import json
import os
from cryptobot.webhook import check_signature
from fastapi import FastAPI, Request, HTTPException
app = FastAPI()
api_token = os.environ["CRYPTOBOT_API_TOKEN"]
@app.post("/webhook")
async def webhook_handler(request: Request):
# Read the raw body first (required for signature verification)
raw_body = await request.body()
raw_body_str = raw_body.decode("utf-8")
if not check_signature(api_token, raw_body_str, request.headers):
raise HTTPException(status_code=400, detail="Invalid signature")
data = json.loads(raw_body_str)
# Process the verified webhook data
# ...
return {"ok": True}
Available Assets
The library supports these assets (see cryptobot.models.Asset for the canonical list):
from cryptobot.models import Asset
# Available assets
Asset.BTC # Bitcoin
Asset.TON # Toncoin
Asset.ETH # Ethereum
Asset.USDT # Tether
Asset.USDC # USD Coin
Asset.BNB # Binance Coin
Asset.TRX # TRON
Advanced Patterns
Pagination with iterators
When dealing with many records, prefer the built-in paginated iterator helpers. They are available for invoices, transfers, and checks.
Invoices:
# Iterate by page
for page in client.iter_invoice_pages(asset=Asset.USDT, status=Status.paid, page_size=100):
print("page size:", len(page))
# Iterate flattened invoices
for invoice in client.iter_invoices(asset=Asset.USDT, status=Status.paid, page_size=100):
print(invoice.invoice_id, invoice.status)
Transfers:
for transfer in client.iter_transfers(asset=Asset.TON, page_size=100):
print(transfer.transfer_id, transfer.amount)
Checks:
for check in client.iter_checks(asset=Asset.USDT, status="active", page_size=100):
print(check.check_id, check.bot_check_url)
Async pagination works with async for:
import os
from cryptobot import AsyncCryptoBotClient
from cryptobot.models import Asset, Status
async with AsyncCryptoBotClient(api_token=os.environ["CRYPTOBOT_API_TOKEN"]) as async_client:
async for page in async_client.iter_invoice_pages(
asset=Asset.USDT,
status=Status.paid,
page_size=100,
start_offset=0,
):
print("page size:", len(page))
async for invoice in async_client.iter_invoices(
asset=Asset.USDT,
status=Status.paid,
page_size=100,
start_offset=0,
):
print(invoice.invoice_id, invoice.status)
All iterator variants support start_offset and validate page_size in the 1..1000 range.
Invoice Status Checking
Create a helper function to check invoice status:
def wait_for_payment(client, invoice_id, max_attempts=60, delay=5):
"""Poll invoice status until paid or timeout"""
import time
for attempt in range(max_attempts):
invoices = client.get_invoices(invoice_ids=str(invoice_id))
if invoices and invoices[0].status == Status.paid:
return invoices[0]
if invoices and invoices[0].status == Status.expired:
raise Exception("Invoice expired")
time.sleep(delay)
raise TimeoutError("Payment timeout")
# Usage
invoice = client.create_invoice(asset=Asset.USDT, amount=10)
print(f"Waiting for payment: {invoice.bot_invoice_url}")
try:
paid_invoice = wait_for_payment(client, invoice.invoice_id)
print(f"Payment received! Amount: {paid_invoice.paid_amount}")
except Exception as e:
print(f"Payment failed: {e}")
Calculating Total Revenue
Calculate total revenue from paid invoices:
from decimal import Decimal
def calculate_revenue(client, asset=None):
"""Calculate total revenue from paid invoices"""
invoices = client.get_invoices(status=Status.paid, asset=asset)
revenue_by_asset = {}
for invoice in invoices:
asset_name = invoice.paid_asset or invoice.asset.name
amount = Decimal(invoice.paid_amount or invoice.amount)
if asset_name in revenue_by_asset:
revenue_by_asset[asset_name] += amount
else:
revenue_by_asset[asset_name] = amount
return revenue_by_asset
# Usage
revenue = calculate_revenue(client)
for asset, total in revenue.items():
print(f"{asset}: {total}")
Multi-Currency Pricing
Create invoices that allow users to pay in different currencies:
def create_flexible_invoice(client, amount_usd, description):
"""Create an invoice with multiple payment options"""
# Get exchange rates
rates = client.get_exchange_rates()
# Find cryptocurrencies and their rates
available_assets = [Asset.USDT, Asset.BTC, Asset.ETH, Asset.TON]
# Create invoice (USDT is 1:1 with USD for simplicity)
invoice = client.create_invoice(
asset=Asset.USDT,
amount=amount_usd,
description=description,
allow_comments=True,
expires_in=3600 # 1 hour
)
# Show equivalent amounts in other currencies
print(f"Invoice created: {invoice.bot_invoice_url}")
print(f"\nPayment options:")
print(f" USDT: ${amount_usd}")
for rate in rates:
if rate.target == "USD" and rate.source in [a.name for a in available_assets]:
crypto_amount = amount_usd / float(rate.rate)
print(f" {rate.source}: {crypto_amount:.8f}")
return invoice
# Usage
invoice = create_flexible_invoice(client, 50.0, "Premium subscription")
Best Practices
Always handle exceptions when making API calls
Use unique spend_ids for transfers to prevent duplicates
Validate user input before creating invoices or transfers
Store API tokens securely using environment variables
Use a dedicated testnet token with
is_mainnet=Falseduring developmentImplement proper webhook signature verification for security
Set appropriate invoice expiration times to avoid stale invoices
Use pagination when retrieving large numbers of invoices
Check invoice status before processing orders
Log all transactions for audit trails and debugging