Python SDK - l402-kit

安装

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 同样支持托管模式 — 使用 ManagedProvider.from_address("you@blink.sv")（手续费 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 worker：l402_required 会在同步 Flask 处理函数中调用异步 Lightning 提供商。如果你的 Gunicorn worker 使用了 gevent 或 eventlet 的 monkey-patching（会创建一个正在运行的事件循环），装饰器会自动检测到这一情况，并在独立线程中运行异步调用 — 无需任何额外操作。标准 Gunicorn 同步 worker 和 uvicorn（FastAPI）不受影响。

`@l402_required` — 装饰器

参数

参数	类型	默认值	描述
`price_sats`	`int`	必填	每次调用的价格（单位：satoshis）
`lightning`	`LightningProvider`	必填	你的 Lightning 后端
`replay`	`ReplayAdapter`	内存模式	可插拔的重放保护后端

行为说明

请求	响应
无 `Authorization` 请求头	返回 `402`，包含 invoice、macaroon 和价格
有效的 `L402 <macaroon>:<preimage>`	正常执行处理函数
无效或已过期的令牌	`401 Unauthorized`
重放的 preimage	`401 Token already used`

402 响应

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

Providers

`BlinkProvider`

Blink — 免费托管 Lightning 钱包，小额充值无需 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` — 云模式（手续费 0.3%）

l402kit.com 托管 Lightning 节点。你将获得每笔付款的 99.7% — 无需自行搭建节点。

from l402kit import ManagedProvider

lightning = ManagedProvider.from_address("you@blink.sv")

# 可选：在公共 API 目录中注册
lightning = ManagedProvider.from_address("you@blink.sv", register_directory={
    "url": "https://api.you.com/v1/weather",
    "name": "Weather API",
    "price_sats": 10,
    "category": "weather",
})

重放保护

默认 — 内存模式（开发环境）

内置功能，无需任何配置。进程重启后状态重置。

Redis（生产环境 — 多实例）

对于 Gunicorn/uvicorn 多 worker 部署，通过 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 = 已被使用

自定义 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，自动处理完整的 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()

​安装

​Soberano 模式（收益 100% 归你）

​Flask

​@l402_required — 装饰器

​参数

​行为说明

​402 响应

​Providers

​BlinkProvider

​LNbitsProvider

​OpenNodeProvider

​ManagedProvider — 云模式（手续费 0.3%）

​重放保护

​默认 — 内存模式（开发环境）

​Redis（生产环境 — 多实例）

​独立工具函数

​自定义 Provider

​测试

​运行

​L402Client — 自动支付

​钱包

​AsyncL402Client — async/await

​与 L402Client 的区别

​DevProvider + DevWallet — 本地开发

安装

Soberano 模式（收益 100% 归你）

Flask

`@l402_required` — 装饰器

参数

行为说明

402 响应

Providers

`BlinkProvider`

`LNbitsProvider`

`OpenNodeProvider`

`ManagedProvider` — 云模式（手续费 0.3%）

重放保护

默认 — 内存模式（开发环境）

Redis（生产环境 — 多实例）

独立工具函数

自定义 Provider

测试

运行

L402Client — 自动支付

钱包

AsyncL402Client — async/await

与 `L402Client` 的区别

DevProvider + DevWallet — 本地开发