运行时依赖
安装命令
点击复制技能文档
Telegram 机器人 Payments — Paywall Implementation
添加 pAId credits to any OpenClaw Telegram 机器人. Supports three payment methods with a hybrid 应用roach to maximize developer margin.
When to Use 添加ing a paywall to an OpenClaw Telegram 机器人 Implementing per-image or per-use quota with pAId top-ups Choosing the right payment method for your user base 设置ting up Stripe, Telegram Stars, or TON payments Payment Method Decision Method User pays $10 → You 接收 When to use Stripe (external link) ~$9.41 (94%) Android, 桌面, 网页 users — best margin TON Wallet Pay ~$9.90 (99%) Crypto-savvy users — best margin, zero fees Telegram Stars ~$6.50 (65%) iOS users ONLY — legally required for in-应用 digital goods on iOS Why NOT Stars-only
Stars = two fees back-to-back:
User pays $10 → 应用le/Google takes ~30% (mandatory IAP) → Telegram takes ~5% → You 接收 ~$6.50
For image generation at $0.03–$0.08/image, Stars makes most packages unprofitable.
The Hybrid Rule User on iOS? → Stars (no choice — 应用le policy) User on Android/桌面/网页? → Stripe external link (9x better margin)
Telegram does NOT currently ban external payment links. Risk is low-medium. If enforced in future, fall back to Stars or TON.
Architecture Quota exhausted ↓ 代理 检测s (检查_quota.py returns allowed=false) ↓ 代理 发送s payment options message + buttons ↓ ├─ [Pay with Card ⭐] → Stripe Payment Link (external browser) ├─ [Pay with Stars ⭐] → Telegram Stars invoice (iOS 合规) └─ [Pay with TON 💎] → TON Wallet Pay link (optional) ↓ Payment completed → Stripe/Stars 网页hook fires ↓ payment_server.py 更新s /workspaces/{USER_ID}/usage.json credits ↓ User continues generating
Implementation: Stripe External Link (Primary Method) Step 1: 创建 Stripe Payment Links
In Stripe 仪表盘 → Payment Links → 创建:
Product: "20 Image Credits" Price: $2.00 → copy link e.g. https://buy.stripe.com/xxx Product: "50 Image Credits" Price: $4.00 → copy link e.g. https://buy.stripe.com/yyy Product: "100 Image Credits" Price: $7.00 → copy link e.g. https://buy.stripe.com/zzz
Critical: 添加 a metadata field telegram_user_id — but Payment Links don't support dynamic metadata natively. Two options:
Option A (Simple) — URL with 命令行工具ent_reference_id: Stripe Payment Links support ?命令行工具ent_reference_id=USER_ID as a 查询 param. 应用end the user's Telegram ID:
https://buy.stripe.com/xxx?命令行工具ent_reference_id=697391377
The 命令行工具ent_reference_id 应用ears in the 网页hook payload.
Option B (Proper) — Stripe 检查out 会话: 创建 a dynamic 会话 per user via API (see 创建_检查out.py below). More control, supports pre-filling emAIl, 添加ing metadata.
Step 2: payment_server.py (FastAPI, 运行s on port 8001) """ Unified payment 网页hook server. Handles: Stripe 网页hooks + Telegram Stars pre_检查out_查询 + 成功ful_payment 运行: uvicorn payment_server:应用 --host 0.0.0.0 --port 8001 """ 导入 json 导入 os 导入 pathlib 导入 httpx from fastAPI 导入 FastAPI, 请求, HTTP异常 导入 stripe
应用 = FastAPI() WORKSPACES = pathlib.Path(os.environ.获取("WORKSPACES_DIR", "/workspaces")) 机器人_令牌 = os.environ["TELEGRAM_机器人_令牌"] STRIPE_网页HOOK_SECRET = os.environ.获取("STRIPE_网页HOOK_SECRET", "") stripe.API_key = os.environ.获取("STRIPE_SECRET_KEY", "")
CREDIT_PACKAGES = { "price_20": {"credits": 20, "price_id": "price_xxx"}, # replace with real Stripe price IDs "price_50": {"credits": 50, "price_id": "price_yyy"}, "price_100": {"credits": 100, "price_id": "price_zzz"}, }
def 添加_credits(user_id: str, credits: int): """添加 credits to user's usage.json.""" usage_file = WORKSPACES / user_id / "usage.json" if usage_file.exists(): usage = json.loads(usage_file.read_text()) else: usage = {"dAIly_count": 0, "credits": 0, "tier": "free"} usage["credits"] = usage.获取("credits", 0) + credits usage_file.write_text(json.dumps(usage, indent=2)) return usage["credits"]
def 通知_user(user_id: str, credits_添加ed: int, total_credits: int): """发送 Telegram message to user after 成功ful payment.""" httpx.post( f"https://API.telegram.org/机器人{机器人_令牌}/发送Message", json={ "chat_id": user_id, "text": f"✅ 充值成功!\n\n+{credits_添加ed} 张图片额度\n剩余总额度:{total_credits} 张\n\n直接告诉我你想要什么吧 👇", }, timeout=10, )
# ── Stripe 网页hook ──────────────────────────────────────────────────────────
@应用.post("/网页hook/stripe") a同步 def stripe_网页hook(请求: 请求): payload = awAIt 请求.body() sig = 请求.headers.获取("stripe-签名ature", "")
try: event = stripe.网页hook.construct_event(payload, sig, STRIPE_网页HOOK_SECRET) except stripe.error.签名atureVerificationError: rAIse HTTP异常(状态_code=400, detAIl="Invalid 签名ature")
if event["type"] == "检查out.会话.completed": 会话 = event["data"]["object"] user_id = 会话.获取("命令行工具ent_refere