Python SDK

l402kit Python SDK の完全リファレンス — Bitcoin Lightning の従量課金APIに対応したFastAPIおよびFlaskデコレーター。

インストール

pip install l402kit

必要条件: Python 3.11以上、FastAPI または Flask（任意）

---

Soberanoモード（100%収益を保持）

独自のLightningプロバイダーを使用 — 支払いは直接あなたのウォレットへ、手数料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%、ノード不要）。以下のプロバイダーセクションを参照してください。

---

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ワーカー使用時）: l402_required は同期FlaskハンドラーからasyncのLightningプロバイダーを呼び出します。Gunicornワーカーがgeventまたはeventletのモンキーパッチ（実行中のイベントループを作成する）を使用している場合、デコレーターはこれを自動検出し、非同期呼び出しを専用スレッドで実行します — 特別な対応は不要です。標準のGunicorn同期ワーカーおよびuvicorn（FastAPI）は影響を受けません。

---

@l402_required — デコレーター

パラメーター

パラメーター	型	デフォルト	説明
price_sats	int	必須	1回の呼び出しあたりの価格（satoshi単位）
lightning	LightningProvider	必須	使用するLightningバックエンド
replay	ReplayAdapter	インメモリ	プラグイン可能なリプレイ保護バックエンド

動作

リクエスト	レスポンス
Authorization ヘッダーなし	インボイス、macaroon、価格を含む 402
有効な L402 <macaroon>:<preimage>	ハンドラーが通常通り実行される
無効または期限切れのトークン	401 Unauthorized
リプレイされたpreimage	401 Token already used

402 レスポンス

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

---

プロバイダー

BlinkProvider

Blink — 少額の場合KYC不要の無料カストディアルLightningウォレット。

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 — クラウドモード（手数料0.3%）

l402kit.comがLightningノードをホストします。各支払いの99.7%を受け取れます — ノードのセットアップは不要です。

from l402kit import ManagedProvider

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

# Optional: register in the public 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",
})

---

リプレイ保護

デフォルト — インメモリ（開発環境）

組み込み済み、設定不要。プロセス再起動時にリセットされます。

Redis（本番環境 — マルチインスタンス）

Gunicorn/uvicornのマルチワーカーデプロイメントでは、Redisを通じてリプレイ状態を共有します:

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 を使用 — アトミックかつレースコンディションなし。

---

スタンドアロンユーティリティ

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

# トークンの検証（True / False）
is_valid = verify_token("eyJoYXNoIjoiYWJjMTIzIiwiZXhwIjoxNzAwMDAwMDAwfQ==:deadbeef...")

# 手動リプレイチェック
is_first_use = check_and_mark_preimage(preimage)
# True = 初回使用、False = 使用済み

---

カスタムプロバイダー

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

# インボイスを支払い後:
curl -H "Authorization: L402 <macaroon>:<preimage>" http://localhost:3000/api/data

---

L402Client — 自動支払い

L402Client は httpx をラップし、402 → 支払い → 再試行のループを自動で処理します。

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()

ウォレット

クラス	インストール	説明
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エージェントフレームワークに最適です。

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クライアント	httpx（同期）	httpx.AsyncClient
get / post	同期	async
最適な用途	スクリプト、Flask	FastAPI、asyncio、LangChain _arun
バジェット / キャッシュ	✅ 同じ	✅ 同じ

---

DevProvider + DevWallet — ローカル開発

設定不要のローカル開発 — Lightningノード不要、実際の支払い不要。本番環境と暗号的に同等: 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"}

# クライアント — 実際のLightningなしで自動的に支払い
client = L402Client(wallet=wallet)
data   = client.get("http://localhost:8000/premium").json()