Python SDK - l402-kit

Installation

pip install l402kit

Requirements: Python 3.11+, FastAPI or Flask (optional)

Soberano mode (you keep 100%)

Bring your own Lightning provider — payments go directly to your wallet, 0% fees.

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 also supports Managed mode — use ManagedProvider.from_address("you@blink.sv") (0.3% fee, no node needed). See the Providers section below.

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 with gevent/eventlet workers: l402_required calls the async Lightning provider from a sync Flask handler. If your Gunicorn worker uses gevent or eventlet monkey-patching (which creates a running event loop), the decorator detects this automatically and runs the async call in a dedicated thread — no action needed. Standard Gunicorn sync workers and uvicorn (FastAPI) are unaffected.

`@l402_required` — decorator

Parameters

Parameter	Type	Default	Description
`price_sats`	`int`	required	Price per call in satoshis
`lightning`	`LightningProvider`	required	Your Lightning backend
`replay`	`ReplayAdapter`	in-memory	Pluggable replay protection backend

Behavior

Request	Response
No `Authorization` header	`402` with invoice, macaroon, price
Valid `L402 <macaroon>:<preimage>`	Handler executes normally
Invalid or expired token	`401 Unauthorized`
Replayed preimage	`401 Token already used`

402 response

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

Providers

`BlinkProvider`

Blink — free custodial Lightning wallet, no KYC for small amounts.

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 mode (0.3% fee)

l402kit.com hosts the Lightning node. You receive 99.7% of each payment — no node setup required.

from l402kit import ManagedProvider

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

# Optional: register in the public API directory
lightning = ManagedProvider.from_address("you@blink.sv", register_directory={
    "url": "https://api.you.com/v1/weather",
    "name": "Weather API",
    "price_sats": 10,
    "category": "weather",
})

Replay protection

Default — in-memory (development)

Built-in, no configuration needed. Resets on process restart.

Redis (production — multi-instance)

For Gunicorn/uvicorn multi-worker deployments, share replay state via 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 uses SET key 1 NX EX ttl — atomic and race-condition free.

Standalone utilities

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

# Verify a token (True / False)
is_valid = verify_token("eyJoYXNoIjoiYWJjMTIzIiwiZXhwIjoxNzAwMDAwMDAwfQ==:deadbeef...")

# Manual replay check
is_first_use = check_and_mark_preimage(preimage)
# True = first use, False = already used

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)

Testing

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

Running

# FastAPI
uvicorn main:app --port 3000

# Flask
python app.py

# Test — triggers 402
curl http://localhost:3000/api/data

# Pay invoice, then:
curl -H "Authorization: L402 <macaroon>:<preimage>" http://localhost:3000/api/data

L402Client — automatic payment

L402Client wraps httpx and handles the full 402 → pay → retry loop automatically.

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	Install	Description
`BlinkWallet`	`pip install l402kit`	Pay via Blink GraphQL API
`AlbyWallet`	`pip install l402kit`	Pay via Alby REST API

from l402kit.wallets import BlinkWallet, AlbyWallet

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

`build_wallet(env=None)` — auto-detect

Picks the wallet from environment variables. Priority: BLINK_API_KEY + BLINK_WALLET_ID → Blink, else ALBY_TOKEN → Alby. Reads os.environ by default.

from l402kit.wallets import build_wallet

wallet = build_wallet()  # auto-detected
# wallet = build_wallet({"BLINK_API_KEY": ..., "BLINK_WALLET_ID": ...})  # explicit

Raises ValueError if no credentials are configured.

LAW-N adapter — behavioral telemetry (1.10.0+)

Forward L402 payment events to a LAW-N ingest endpoint. Same contract as TS/Rust/Go: POST JSON + HMAC-SHA256 in X-LAW-N-Signature + random X-LAW-N-Request-Id + fire-and-forget. Network errors are swallowed so behavioral writes never block payments.

import os
from l402kit.integrations import create_lawn_adapter

on_event = create_lawn_adapter(
    endpoint="https://law-n.sageworks.ai/ingest/events",
    secret=os.environ["LAWN_SECRET"],
    timeout=5.0,  # seconds, optional
)

# Pass on_event to your L402Client / agent framework so it fires per payment.

AsyncL402Client — async/await

AsyncL402Client uses httpx.AsyncClient internally — ideal for FastAPI, asyncio, and AI agent frameworks that run in an 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())

Difference from `L402Client`

	`L402Client`	`AsyncL402Client`
HTTP client	`httpx` (sync)	`httpx.AsyncClient`
`get` / `post`	sync	`async`
Best for	Scripts, Flask	FastAPI, asyncio, LangChain `_arun`
Budget / cache	✅ same	✅ same

DevProvider + DevWallet — local development

Zero-config local development — no Lightning node, no real payments. Cryptographically identical to 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 — pays automatically without real Lightning
client = L402Client(wallet=wallet)
data   = client.get("http://localhost:8000/premium").json()

​Installation

​Soberano mode (you keep 100%)

​Flask

​@l402_required — decorator

​Parameters

​Behavior

​402 response

​Providers

​BlinkProvider

​LNbitsProvider

​OpenNodeProvider

​ManagedProvider — cloud mode (0.3% fee)

​Replay protection

​Default — in-memory (development)

​Redis (production — multi-instance)

​Standalone utilities

​Custom provider

​Testing

​Running

​L402Client — automatic payment

​Wallets

​build_wallet(env=None) — auto-detect

​LAW-N adapter — behavioral telemetry (1.10.0+)

​AsyncL402Client — async/await

​Difference from L402Client

​DevProvider + DevWallet — local development

Installation

Soberano mode (you keep 100%)

Flask

`@l402_required` — decorator

Parameters

Behavior

402 response

Providers

`BlinkProvider`

`LNbitsProvider`

`OpenNodeProvider`

`ManagedProvider` — cloud mode (0.3% fee)

Replay protection

Default — in-memory (development)

Redis (production — multi-instance)

Standalone utilities

Custom provider

Testing

Running

L402Client — automatic payment

Wallets

`build_wallet(env=None)` — auto-detect

LAW-N adapter — behavioral telemetry (1.10.0+)

AsyncL402Client — async/await

Difference from `L402Client`

DevProvider + DevWallet — local development