Python SDK

l402kit Python SDK का संपूर्ण संदर्भ — Bitcoin Lightning pay-per-call APIs के लिए FastAPI और Flask decorator।

इंस्टॉलेशन

pip install l402kit

आवश्यकताएं: Python 3.11+, FastAPI या Flask (वैकल्पिक)

---

Soberano मोड (आप 100% रखते हैं)

अपना Lightning provider लाएं — भुगतान सीधे आपके wallet में जाता है, 0% शुल्क।

import os
from fastapi import FastAPI, Request
from l402kit import l402_required
from l402kit.providers.blink import BlinkProvider

app = FastAPI()

lightning = BlinkProvider(
    api_key=os.environ["BLINK_API_KEY"],
    wallet_id=os.environ["BLINK_WALLET_ID"],
)

@app.get("/api/data")
@l402_required(price_sats=10, lightning=lightning)
async def get_data(request: Request):
    return {"data": "premium content"}

Python Managed मोड को भी सपोर्ट करता है — ManagedProvider.from_address("[e-mail]") उपयोग करें (0.3% शुल्क, कोई नोड की जरूरत नहीं)। नीचे Providers अनुभाग देखें।

---

Flask

import os
from flask import Flask, jsonify
from l402kit import l402_required
from l402kit.providers.blink import BlinkProvider

app = Flask(__name__)

lightning = BlinkProvider(
    api_key=os.environ["BLINK_API_KEY"],
    wallet_id=os.environ["BLINK_WALLET_ID"],
)

@app.route("/api/data")
@l402_required(price_sats=10, lightning=lightning)
def get_data():
    return jsonify({"data": "premium content"})

if __name__ == "__main__":
    app.run(port=3000)

Flask + Gunicorn के साथ gevent/eventlet workers: l402_required एक sync Flask handler से async Lightning provider को कॉल करता है। यदि आपका Gunicorn worker gevent या eventlet monkey-patching उपयोग करता है (जो एक running event loop बनाता है), तो decorator इसे स्वचालित रूप से पहचानता है और async कॉल को एक dedicated thread में चलाता है — कोई कार्रवाई आवश्यक नहीं। Standard Gunicorn sync workers और uvicorn (FastAPI) अप्रभावित रहते हैं।

---

@l402_required — decorator

पैरामीटर

पैरामीटर	प्रकार	डिफ़ॉल्ट	विवरण
price_sats	int	आवश्यक	प्रति कॉल satoshis में मूल्य
lightning	LightningProvider	आवश्यक	आपका Lightning backend
replay	ReplayAdapter	in-memory	Pluggable replay protection backend

व्यवहार

अनुरोध	प्रतिक्रिया
कोई Authorization हेडर नहीं	invoice, macaroon, मूल्य के साथ 402
वैध L402 <macaroon>:<preimage>	Handler सामान्य रूप से execute होता है
अमान्य या expired token	401 Unauthorized
Replayed preimage	401 Token already used

402 प्रतिक्रिया

{
  "error": "Payment Required",
  "price_sats": 10,
  "invoice": "lnbc100n1...",
  "macaroon": "eyJoYXNoIjoiYWJjMTIzIiwiZXhwIjoxNzAwMDAwMDAwfQ=="
}

---

Providers

BlinkProvider

Blink — मुफ्त custodial Lightning wallet, छोटी राशियों के लिए कोई KYC नहीं।

from l402kit.providers.blink import BlinkProvider

blink = BlinkProvider(
    api_key=os.environ["BLINK_API_KEY"],    # dashboard.blink.sv → API Keys
    wallet_id=os.environ["BLINK_WALLET_ID"],
)

LNbitsProvider

from l402kit.providers.lnbits import LNbitsProvider

lnbits = LNbitsProvider(
    api_key=os.environ["LNBITS_API_KEY"],
    base_url="https://your-lnbits.com",  # optional
)

OpenNodeProvider

from l402kit.providers.opennode import OpenNodeProvider

opennode = OpenNodeProvider(
    api_key=os.environ["OPENNODE_API_KEY"],
    test_mode=False,  # True for sandbox
)

ManagedProvider — cloud मोड (0.3% शुल्क)

l402kit.com Lightning नोड होस्ट करता है। आप प्रत्येक भुगतान का 99.7% प्राप्त करते हैं — कोई नोड सेटअप आवश्यक नहीं।

from l402kit import ManagedProvider

lightning = ManagedProvider.from_address("[e-mail]")

# वैकल्पिक: सार्वजनिक API directory में पंजीकृत करें
lightning = ManagedProvider.from_address("[e-mail]", register_directory={
    "url": "https://api.you.com/v1/weather",
    "name": "Weather API",
    "price_sats": 10,
    "category": "weather",
})

---

Replay protection

डिफ़ॉल्ट — in-memory (विकास)

Built-in, कोई कॉन्फ़िगरेशन आवश्यक नहीं। Process restart पर रीसेट होता है।

Redis (production — multi-instance)

Gunicorn/uvicorn multi-worker deployments के लिए, Redis के माध्यम से replay state साझा करें:

import os, redis
from l402kit import l402_required, RedisReplayAdapter

r = redis.Redis.from_url(os.environ["REDIS_URL"])
replay = RedisReplayAdapter(r, ttl_seconds=86400)

@app.get("/api/data")
@l402_required(
    price_sats=10,
    lightning=lightning,
    replay=replay,
)
async def get_data(request: Request):
    return {"data": "premium content"}

RedisReplayAdapter SET key 1 NX EX ttl उपयोग करता है — atomic और race-condition मुक्त।

---

Standalone utilities

from l402kit.verify import verify_token
from l402kit.replay import check_and_mark_preimage

# एक token verify करें (True / False)
is_valid = verify_token("eyJoYXNoIjoiYWJjMTIzIiwiZXhwIjoxNzAwMDAwMDAwfQ==:deadbeef...")

# Manual replay जांच
is_first_use = check_and_mark_preimage(preimage)
# True = पहला उपयोग, False = पहले ही उपयोग किया जा चुका है

---

Custom provider

from l402kit.types import LightningProvider, Invoice
import base64, json, time

class MyProvider(LightningProvider):
    async def create_invoice(self, amount_sats: int) -> Invoice:
        result = await my_node.create_invoice(amount_sats)
        exp = int((time.time() + 3600) * 1000)
        macaroon = base64.b64encode(
            json.dumps({"hash": result.hash, "exp": exp}).encode()
        ).decode()
        return Invoice(
            payment_request=result.bolt11,
            payment_hash=result.hash,
            macaroon=macaroon,
            amount_sats=amount_sats,
            expires_at=exp,
        )

    async def check_payment(self, payment_hash: str) -> bool:
        return await my_node.is_paid(payment_hash)

---

परीक्षण

import hashlib, base64, json, time, os
from l402kit.verify import verify_token

def make_test_token() -> str:
    preimage = os.urandom(32).hex()
    payment_hash = hashlib.sha256(bytes.fromhex(preimage)).hexdigest()
    exp = int((time.time() + 3600) * 1000)
    macaroon = base64.b64encode(
        json.dumps({"hash": payment_hash, "exp": exp}).encode()
    ).decode()
    return f"{macaroon}:{preimage}"

assert verify_token(make_test_token()) is True

---

चलाना

# FastAPI
uvicorn main:app --port 3000

# Flask
python app.py

# परीक्षण — 402 ट्रिगर करता है
curl http://localhost:3000/api/data

# invoice का भुगतान करें, फिर:
curl -H "Authorization: L402 <macaroon>:<preimage>" http://localhost:3000/api/data

---

L402Client — स्वचालित भुगतान

L402Client httpx को wrap करता है और पूरे 402 → pay → retry loop को स्वचालित रूप से संभालता है।

from l402kit import L402Client
from l402kit.wallets import BlinkWallet

wallet = BlinkWallet(
    api_key=os.environ["BLINK_API_KEY"],
    wallet_id=os.environ["BLINK_WALLET_ID"],
)

client = L402Client(wallet=wallet)
data = client.get("https://api.example.com/premium").json()

Wallets

Class	इंस्टॉल	विवरण
BlinkWallet	pip install l402kit	Blink GraphQL API के माध्यम से भुगतान
AlbyWallet	pip install l402kit	Alby REST API के माध्यम से भुगतान

from l402kit.wallets import BlinkWallet, AlbyWallet

blink = BlinkWallet(api_key="...", wallet_id="...")
alby  = AlbyWallet(access_token=os.environ["ALBY_TOKEN"])

---

AsyncL402Client — async/await

AsyncL402Client आंतरिक रूप से httpx.AsyncClient उपयोग करता है — FastAPI, asyncio, और AI agent frameworks के लिए आदर्श जो async event loop में चलते हैं।

import asyncio
from l402kit import AsyncL402Client
from l402kit.wallets import BlinkWallet

async def main():
    async with AsyncL402Client(
        wallet=BlinkWallet(os.environ["BLINK_API_KEY"], os.environ["BLINK_WALLET_ID"]),
        budget_sats=500,
    ) as client:
        r = await client.get("https://api.example.com/premium")
        print(r.json())

asyncio.run(main())

L402Client से अंतर

	L402Client	AsyncL402Client
HTTP client	httpx (sync)	httpx.AsyncClient
get / post	sync	async
सबसे उपयुक्त	Scripts, Flask	FastAPI, asyncio, LangChain _arun
Budget / cache	✅ समान	✅ समान

---

DevProvider + DevWallet — स्थानीय विकास

Zero-config स्थानीय विकास — कोई Lightning नोड नहीं, कोई वास्तविक भुगतान नहीं। Cryptographically production के समान: SHA256(preimage) === paymentHash।

from l402kit.dev import DevProvider, DevWallet
from l402kit import L402Client, l402_required
from fastapi import FastAPI, Request

app = FastAPI()
provider = DevProvider()
wallet   = DevWallet(provider)

@app.get("/premium")
@l402_required(price_sats=1, lightning=provider)
async def premium(request: Request):
    return {"data": "premium content"}

# Client — वास्तविक Lightning के बिना स्वचालित रूप से भुगतान करता है
client = L402Client(wallet=wallet)
data   = client.get("http://localhost:8000/premium").json()