Telegram Bot Payments Integration: Step-by-Step Guide with Code

 1# Install required libraries
 2pip install aiogram stripe aiohttp

 1# stripe_config.py — Stripe configuration
 2# Store keys in environment variables, NOT in code!
 3
 4import os
 5import stripe
 6
 7# Test keys (for development)
 8# Find them in Stripe Dashboard → Developers → API keys
 9stripe.api_key = os.getenv("STRIPE_SECRET_KEY")  # sk_test_...
10STRIPE_WEBHOOK_SECRET = os.getenv("STRIPE_WEBHOOK_SECRET")  # whsec_...
11
12# Public URL of your server (needed for webhook)
13# In production replace with your domain
14WEBHOOK_URL = os.getenv("WEBHOOK_URL", "https://your-server.com/webhook/stripe")

 1# payments.py — creating Stripe payment session
 2
 3import stripe
 4from aiogram import Router, F
 5from aiogram.types import CallbackQuery, InlineKeyboardButton
 6from aiogram.utils.keyboard import InlineKeyboardBuilder
 7
 8router = Router()
 9
10async def create_checkout_session(
11    user_id: int,
12    order_id: str,
13    items: list[dict],
14    bot_username: str,
15) -> str:
16    """
17    Creates Stripe Checkout Session for order payment.
18    Returns URL to redirect user to payment.
19
20    Parameters:
21    - user_id: Telegram ID of the buyer
22    - order_id: order number in our system
23    - items: list of items [{name, price, quantity}]
24    - bot_username: bot name (for return after payment)
25    """
26    # Build product list for Stripe
27    line_items = []
28    for item in items:
29        line_items.append({
30            "price_data": {
31                "currency": "usd",  # Payment currency
32                "product_data": {
33                    "name": item["name"],  # Product name
34                },
35                # Stripe accepts price in cents (100 = $1.00)
36                "unit_amount": int(item["price"] * 100),
37            },
38            "quantity": item["quantity"],  # Quantity
39        })
40
41    # Create payment session
42    session = stripe.checkout.Session.create(
43        payment_method_types=["card"],  # Accept cards
44        line_items=line_items,
45        mode="payment",  # One-time payment (not subscription)
46        # Return URL after successful payment
47        success_url=f"https://t.me/{bot_username}?start=payment_success_{order_id}",
48        # Return URL on payment cancellation
49        cancel_url=f"https://t.me/{bot_username}?start=payment_cancel_{order_id}",
50        # Metadata — link payment to our order
51        metadata={
52            "telegram_user_id": str(user_id),
53            "order_id": order_id,
54        },
55    )
56
57    return session.url  # Payment URL
58
59
60@router.callback_query(F.data.startswith("pay:"))
61async def handle_payment(callback: CallbackQuery):
62    """
63    Handler for "Pay" button.
64    Callback data format: "pay:order_id"
65    """
66    order_id = callback.data.split(":")[1]
67
68    # TODO: get order items from database
69    # items = await db.get_order_items(order_id)
70    # Using test data for now:
71    items = [
72        {"name": "T-Shirt (Black, M)", "price": 29.99, "quantity": 1},
73        {"name": "Shipping", "price": 5.99, "quantity": 1},
74    ]
75
76    # Create payment session
77    payment_url = await create_checkout_session(
78        user_id=callback.from_user.id,
79        order_id=order_id,
80        items=items,
81        bot_username=callback.bot.username,
82    )
83
84    # Send button with payment link
85    builder = InlineKeyboardBuilder()
86    builder.row(InlineKeyboardButton(
87        text="💳 Pay with Card",
88        url=payment_url,  # Stripe Checkout link
89    ))
90
91    await callback.message.edit_text(
92        text=f"🛒 <b>Order #{order_id}</b>\n\n"
93             f"Total: $35.98\n\n"
94             f"Tap the button below to pay securely via Stripe:",
95        reply_markup=builder.as_markup(),
96    )
97    await callback.answer()

 1# webhook.py — handling webhook from Stripe
 2
 3from aiohttp import web
 4import stripe
 5import json
 6
 7# Webhook secret key (from Stripe Dashboard → Webhooks)
 8WEBHOOK_SECRET = "whsec_your_webhook_secret_here"
 9
10async def stripe_webhook_handler(request: web.Request) -> web.Response:
11    """
12    Receives webhook events from Stripe.
13    Stripe sends confirmation here after each payment.
14    """
15    # Get request body and signature
16    payload = await request.body()
17    sig_header = request.headers.get("Stripe-Signature")
18
19    try:
20        # Verify signature — ensure request is from Stripe
21        event = stripe.Webhook.construct_event(
22            payload, sig_header, WEBHOOK_SECRET
23        )
24    except ValueError:
25        # Invalid payload
26        return web.Response(status=400)
27    except stripe.error.SignatureVerificationError:
28        # Invalid signature — possibly fake request
29        return web.Response(status=400)
30
31    # Handle successful payment event
32    if event["type"] == "checkout.session.completed":
33        session = event["data"]["object"]
34
35        # Extract metadata from session
36        user_id = int(session["metadata"]["telegram_user_id"])
37        order_id = session["metadata"]["order_id"]
38        amount = session["amount_total"] / 100  # Convert from cents
39
40        # Update order status in database
41        # await db.update_order_status(order_id, "paid")
42
43        # Send confirmation to buyer via Telegram
44        # await bot.send_message(
45        #     chat_id=user_id,
46        #     text=f"✅ Payment received! Order #{order_id}\n"
47        #          f"Amount: ${amount:.2f}\n"
48        #          f"We will ship your order soon!"
49        # )
50
51        print(f"✅ Payment confirmed: order={order_id}, user={user_id}, amount=${amount}")
52
53    return web.Response(status=200)
54
55
56def setup_webhook_app(bot) -> web.Application:
57    """
58    Configures aiohttp application for receiving webhook.
59    Runs on the same server as the bot.
60    """
61    app = web.Application()
62    # Register webhook handler at /webhook/stripe
63    app.router.add_post("/webhook/stripe", stripe_webhook_handler)
64    return app

 1# paypal_payments.py — creating order via PayPal API
 2
 3import aiohttp
 4import os
 5
 6# PayPal keys (from Developer Dashboard)
 7PAYPAL_CLIENT_ID = os.getenv("PAYPAL_CLIENT_ID")
 8PAYPAL_SECRET = os.getenv("PAYPAL_SECRET")
 9
10# Base URL (sandbox for testing, api-m for production)
11PAYPAL_BASE = "https://api-m.sandbox.paypal.com"  # Test mode
12# PAYPAL_BASE = "https://api-m.paypal.com"  # Production
13
14async def get_paypal_access_token() -> str:
15    """
16    Gets PayPal API access token.
17    Token lives 8 hours, needs refreshing.
18    """
19    async with aiohttp.ClientSession() as session:
20        # Authorization via Basic Auth
21        auth = aiohttp.BasicAuth(PAYPAL_CLIENT_ID, PAYPAL_SECRET)
22        async with session.post(
23            f"{PAYPAL_BASE}/v1/oauth2/token",
24            auth=auth,
25            data={"grant_type": "client_credentials"},
26        ) as resp:
27            data = await resp.json()
28            return data["access_token"]
29
30
31async def create_paypal_order(
32    order_id: str,
33    items: list[dict],
34    total_amount: float,
35) -> dict:
36    """
37    Creates order in PayPal and returns payment link.
38
39    Parameters:
40    - order_id: order number in our system
41    - items: list of items [{name, price, quantity}]
42    - total_amount: total order amount in dollars
43    """
44    token = await get_paypal_access_token()
45
46    # Build request body for PayPal
47    order_body = {
48        "intent": "CAPTURE",  # Capture payment immediately
49        "purchase_units": [{
50            "reference_id": order_id,  # Our order number
51            "amount": {
52                "currency_code": "USD",
53                # Split amount into items and shipping
54                "value": f"{total_amount:.2f}",
55                "breakdown": {
56                    "item_total": {
57                        "currency_code": "USD",
58                        "value": f"{total_amount:.2f}",
59                    }
60                }
61            },
62            # List of items in order
63            "items": [
64                {
65                    "name": item["name"],
66                    "unit_amount": {
67                        "currency_code": "USD",
68                        "value": f"{item['price']:.2f}",
69                    },
70                    "quantity": str(item["quantity"]),
71                }
72                for item in items
73            ],
74        }],
75        # Where to return user after payment
76        "application_context": {
77            "return_url": f"https://your-server.com/paypal/success?order={order_id}",
78            "cancel_url": f"https://your-server.com/paypal/cancel?order={order_id}",
79        }
80    }
81
82    async with aiohttp.ClientSession() as session:
83        async with session.post(
84            f"{PAYPAL_BASE}/v2/checkout/orders",
85            headers={
86                "Authorization": f"Bearer {token}",
87                "Content-Type": "application/json",
88            },
89            json=order_body,
90        ) as resp:
91            data = await resp.json()
92
93            # Look for payment link in response
94            for link in data.get("links", []):
95                if link["rel"] == "approve":
96                    return {
97                        "order_id": data["id"],  # Order ID in PayPal
98                        "approval_url": link["href"],  # Payment link
99                    }
100
101            raise Exception(f"PayPal error: {data}")

 1# paypal_capture.py — capturing payment after buyer approval
 2
 3async def capture_paypal_order(paypal_order_id: str) -> dict:
 4    """
 5    Captures payment after buyer approved it on PayPal page.
 6    Called from webhook or callback.
 7    """
 8    token = await get_paypal_access_token()
 9
10    async with aiohttp.ClientSession() as session:
11        async with session.post(
12            f"{PAYPAL_BASE}/v2/checkout/orders/{paypal_order_id}/capture",
13            headers={
14                "Authorization": f"Bearer {token}",
15                "Content-Type": "application/json",
16            },
17        ) as resp:
18            data = await resp.json()
19
20            # Check capture status
21            if data["status"] == "COMPLETED":
22                # Payment successful!
23                payer_email = data["payer"]["email_address"]
24                amount = data["purchase_units"][0]["payments"]["captures"][0]["amount"]["value"]
25
26                return {
27                    "status": "completed",
28                    "payer_email": payer_email,
29                    "amount": float(amount),
30                }
31            else:
32                return {"status": data["status"], "error": "Payment not completed"}

 1# telegram_payments.py — sending invoice via Telegram Payments API
 2
 3from aiogram import Router, F
 4from aiogram.types import (
 5    LabeledPrice,       # For specifying price
 6    InlineKeyboardButton,
 7)
 8from aiogram.utils.keyboard import InlineKeyboardBuilder
 9
10router = Router()
11
12# Provider token — obtained from @BotFather
13# For Stripe: token from Stripe Dashboard → Settings → Payment methods
14PROVIDER_TOKEN = "YOUR_PROVIDER_TOKEN_FROM_BOTFATHER"
15
16@router.callback_query(F.data.startswith("invoice:"))
17async def send_invoice(callback):
18    """
19    Sends invoice to buyer via Telegram Payments.
20    Callback data format: "invoice:order_id"
21    """
22    order_id = callback.data.split(":")[1]
23
24    # Build price list (items + shipping)
25    prices = [
26        LabeledPrice(label="T-Shirt (Black, M)", amount=2999),  # $29.99 in cents
27        LabeledPrice(label="Shipping", amount=599),              # $5.99 in cents
28    ]
29
30    # Send invoice — Telegram will show native payment form
31    await callback.bot.send_invoice(
32        chat_id=callback.from_user.chat.id,
33
34        # Main product information
35        title="Order #1234",                    # Invoice title
36        description="T-Shirt (Black, M) + Shipping",  # Description
37        payload=f"order_{order_id}",            # Internal identifier
38        provider_token=PROVIDER_TOKEN,          # Payment provider token
39        currency="usd",                         # Currency
40        prices=prices,                          # List of items
41
42        # Additional parameters (optional)
43        need_name=False,          # Request name?
44        need_phone_number=False,  # Request phone?
45        need_email=False,         # Request email?
46        need_shipping_address=False,  # Request shipping address?
47        is_flexible=False,        # Shipping price depends on address?
48        start_parameter=f"pay_{order_id}",  # Deep link parameter
49    )
50
51    await callback.answer("Invoice sent! Check your messages.")
52
53
54@router.pre_checkout_query()
55async def process_pre_checkout(pre_checkout_query):
56    """
57    Pre-checkout handler — Telegram asks "all ok?"
58    before charging.
59    Here you can verify stock, amount, etc.
60    """
61    # Check that item is in stock
62    # If all ok — confirm
63    await pre_checkout_query.answer(ok=True)
64
65    # If something is wrong — reject with error:
66    # await pre_checkout_query.answer(
67    #     ok=False,
68    #     error_message="Sorry, this item is out of stock."
69    # )
70
71
72@router.message(F.successful_payment)
73async def process_successful_payment(message):
74    """
75    Successful payment handler.
76    Called AFTER money is charged.
77    """
78    payment = message.successful_payment
79
80    # Extract payment data
81    order_payload = payment.invoice_payload    # Our payload ("order_1234")
82    total_amount = payment.total_amount / 100  # Amount in dollars
83    currency = payment.currency                # Currency
84    telegram_charge_id = payment.telegram_payment_id  # Transaction ID in Telegram
85    provider_charge_id = payment.provider_payment_id  # Transaction ID at provider
86
87    # Update order status in database
88    # await db.update_order_status(order_payload, "paid")
89
90    # Send confirmation to buyer
91    await message.answer(
92        f"✅ <b>Payment successful!</b>\n\n"
93        f"Order: {order_payload}\n"
94        f"Amount: ${total_amount:.2f} {currency}\n\n"
95        f"We will process your order right away! 🚀"
96    )

 1# Secure webhook verification from Stripe
 2# ALWAYS verify signature before processing payment
 3
 4import stripe
 5
 6def verify_stripe_webhook(payload: bytes, sig_header: str, secret: str) -> dict:
 7    """
 8    Verifies webhook signature from Stripe.
 9    If signature is invalid — raises exception.
10    Never process webhook without signature verification!
11    """
12    try:
13        event = stripe.Webhook.construct_event(
14            payload, sig_header, secret
15        )
16        return event
17    except stripe.error.SignatureVerificationError:
18        # Signature mismatch — possible forgery
19        raise ValueError("Invalid webhook signature — possible forgery!")

 1# Server-side order amount validation
 2# Never trust amount from client — recalculate yourself
 3
 4async def validate_order_amount(order_id: str, claimed_total: float) -> bool:
 5    """
 6    Recalculates order amount on server and compares
 7    with declared. Protection against price tampering.
 8    """
 9    # Get order items from DB
10    items = await db.get_order_items(order_id)
11
12    # Recalculate amount using DB prices (NOT from client!)
13    calculated_total = sum(
14        item.price * item.quantity  # Price from our DB
15        for item in items
16    )
17
18    # Compare with declared amount
19    # Allow 1 cent tolerance (rounding)
20    return abs(calculated_total - claimed_total) < 0.01

 1# Idempotency — protection against double charging
 2# If webhook arrived twice — ignore second time
 3
 4async def handle_payment_webhook(transaction_id: str, order_id: str, amount: float):
 5    """
 6    Processes payment webhook with duplicate protection.
 7    """
 8    # Check — have we already processed this transaction?
 9    existing = await db.get_transaction(transaction_id)
10    if existing:
11        # Already processed — ignore (don't charge again!)
12        print(f"⚠️ Duplicate webhook ignored: {transaction_id}")
13        return
14
15    # First time — process payment
16    await db.save_transaction(transaction_id, order_id, amount)
17    await db.update_order_status(order_id, "paid")
18    print(f"✅ Payment processed: {transaction_id}")

 1# Protection against duplicate webhook processing
 2# Stripe may send webhook multiple times on retry
 3
 4async def handle_webhook(event_id: str, event_type: str, data: dict):
 5    """
 6    Processes webhook with duplicate protection.
 7    """
 8    # Check — have we seen this event before?
 9    if await db.event_exists(event_id):
10        # Already processed — skip
11        return {"status": "already_processed"}
12
13    # Save event ID to DB (before processing!)
14    await db.save_event(event_id, event_type)
15
16    # Now safe to process
17    if event_type == "checkout.session.completed":
18        await process_payment(data)
19
20    return {"status": "processed"}