Skip to content

Latest commit

 

History

History

python

README.md

Solana pay-kit — Python

solana-pay-kit

Charge stablecoins (USDC, USDT, PYUSD, ...) for any HTTP endpoint, in Python. One package, one surface, two protocols underneath: x402 and the Machine Payments Protocol. FastAPI, Flask, and Django ride on top of a framework-agnostic core.

You do not need to know anything about Solana to use this. Pick a currency, give it your wallet address, gate a route in two lines. The SDK handles the challenge, the on-chain verification, the broadcast, and the settlement.

Python Coverage Branch coverage


Quick start

Three progressively-realistic snippets. Each one runs as-is, copy, paste, hit the URL. Flask is the framework here; the same surface works in FastAPI and Django (see Examples).

1. Smallest possible app

Gate one route with an inline price. Save the snippet as app.py and boot with python app.py. Zero-config: the package uses a published demo keypair as the recipient and the hosted Surfpool sandbox at https://402.surfnet.dev:8899 as the RPC.

# app.py
from flask import Flask, jsonify

import solana_pay_kit
from solana_pay_kit import usd
from solana_pay_kit.flask import require_payment

solana_pay_kit.configure(network="solana_localnet")

app = Flask(__name__)

@app.get("/report")
@require_payment(usd("0.10"))
def report():
    return jsonify(content="premium content")

app.run(host="127.0.0.1", port=8000)

solana_pay_kit.configure(...) builds the process-wide config once at boot. @require_payment(usd("0.10")) answers a 402 with a payment challenge when no valid proof was sent, or runs the view if one was.

Hit /report with pay curl and the customer walks through a USDC payment.

2. Multiple gates via a registry

When more than one route is paid, lift the prices into a single :class:Pricing registry. Routes reference gates by string handle.

# app.py
from flask import Flask, jsonify

import solana_pay_kit
from solana_pay_kit import Gate, Protocol, Pricing, usd
from solana_pay_kit.flask import require_payment

solana_pay_kit.configure(network="solana_localnet")

class Catalog(Pricing):
    def __init__(self):
        defaults = {
            "default_pay_to": solana_pay_kit.config().effective_recipient(),
            "accept_default": solana_pay_kit.config().accept,
        }
        self.report = Gate.build(name="report", amount=usd("0.10"),
                                 description="Premium report", **defaults)
        self.api_call = Gate.build(name="api_call", amount=usd("0.001"),
                                   accept=(Protocol.X402,), **defaults)

catalog = Catalog()
app = Flask(__name__)

@app.get("/report")
@require_payment("report", pricing=catalog)
def report():
    return jsonify(content="premium content")

@app.get("/api/data")
@require_payment("api_call", pricing=catalog)
def api_data():
    return jsonify(data=[])

app.run(host="127.0.0.1", port=8000)

Gates are validated in Gate.build at boot, wrong currency, missing recipient, fee math that doesn't add up, so configuration errors surface before any traffic. accept= is an allowlist; the api_call gate here refuses to settle over MPP.

3. Production-shape config

Snippet 2's demo recipient and public sandbox are fine for poking around. Production wants explicit keys, a dedicated RPC, and a list of accepted stablecoins. The Flask app is unchanged, only the configure call grows.

# app.py, same routes as snippet 2.
import solana_pay_kit
from solana_pay_kit import Gate, Operator, Pricing, Signer, Stablecoin, usd

PLATFORM = "CXhrFZJLKqjzmP3sjYLcF4dTeXWKCy9e2SXXZ2Yo6MPY"

solana_pay_kit.configure(
    network="solana_mainnet",
    stablecoins=(Stablecoin.USDC, Stablecoin.PYUSD),
    operator=Operator(signer=Signer.file("operator.json")),
    rpc_url="https://mainnet.helius-rpc.com/?api-key=YOUR_HELIUS_KEY",
)

class Catalog(Pricing):
    def __init__(self):
        defaults = {"default_pay_to": solana_pay_kit.config().effective_recipient()}
        self.report = Gate.build(name="report", amount=usd("0.10"),
                                 description="Premium report", **defaults)
        # Platform-fee pattern: customer pays $10.00,
        # operator nets $9.70, PLATFORM nets $0.30. x402 auto-disabled.
        self.marketplace_sale = Gate.build(
            name="marketplace_sale",
            amount=usd("10.00"),
            fee_within={PLATFORM: usd("0.30")},
            **defaults,
        )

configure reads literal values here; in real deployments pull the signer and RPC URL from your environment (Signer.env("OPERATOR_KEY"), os.getenv("RPC_URL")) or drive the whole thing from env vars with solana_pay_kit.configure_from().

Two safety rails fire at boot:

  • network="solana_mainnet" plus the published demo signer raises DemoSignerOnMainnetError, no real funds get routed to a publicly known address by accident.
  • Missing mpp.challenge_binding_secret? Preflight resolves one from the environment, falling back to ./.env, generating and persisting one if neither exists, so the HMAC stays stable across restarts. Override via PAY_KIT_MPP_CHALLENGE_BINDING_SECRET to control it from your secret manager.

Sign with solana-keychain

Pay Kit uses native solana-keychain signers. The Signer factory remains available: Signer.file(...), Signer.env(...), Signer.demo(), and the key-material factories return a MemorySigner. Signer.env(...) returns None when the variable is absent or blank. Demo keys are public; use them only in the sandbox.

Import signer types and backends from solana_keychain, not Pay Kit. SolanaSigner supports message signing; TransactionSigner also supports signing transactions without rewriting or broadcasting them. Use a TransactionSigner for payment clients and operators that settle transactions. Backends that modify or broadcast transactions have separate capability classes and are not interchangeable with TransactionSigner.

import asyncio

from solana_keychain import MemorySigner, SolanaSigner
from solana_pay_kit import Signer


async def prove_identity(signer: SolanaSigner) -> None:
    message = b"pay-kit signing example"
    signature = await signer.sign_message(message)
    assert signature.verify(signer.pubkey, message)
    print(str(signer.pubkey))  # base58 address for JSON, headers, or config


# Read an existing Solana CLI keypair file. Never commit this file.
signer: MemorySigner = Signer.file("payer.json")
asyncio.run(prove_identity(signer))

For an already-built solders.transaction.VersionedTransaction, await signing and continue with the returned transaction:

from solana_keychain import TransactionSigner
from solders.transaction import VersionedTransaction


async def sign_payment(
    signer: TransactionSigner, transaction: VersionedTransaction
) -> VersionedTransaction:
    signed = await signer.sign_transaction(transaction)
    return signed.transaction

Signing does not broadcast. A multi-signer transaction can still need other signatures; check signed.is_complete before submitting it.

Migrate existing signers

Previous API Native API
LocalSigner MemorySigner from solana_keychain
signer.pubkey() returning a string signer.pubkey returning solders.pubkey.Pubkey; serialize with str(...)
signer.sign(message) returning bytes await signer.sign_message(message) returning solders.signature.Signature; use bytes(...) only at a byte-oriented boundary
Extracting signer.keypair to sign a transaction await signer.sign_transaction(transaction), then use .transaction

LocalSigner and its .keypair, .secret_key, .sign, and .is_demo helpers are removed. Pass native signers directly; do not wrap them in a Pay Kit adapter or extract private keys. Existing solders keypairs can be wrapped with MemorySigner(keypair). Message and transaction signing are async so remote backends can implement the same interface without exposing keys.

The x402 upto client helpers build_upto_payload(...) and build_upto_header(...) also require await:

from solana_keychain import TransactionSigner
from solana_pay_kit.protocols.x402.client.upto import build_upto_header
from solana_pay_kit.protocols.x402.upto.types import UptoRequirements


async def payment_header(
    signer: TransactionSigner, requirements: UptoRequirements, expires_at: int
) -> str:
    return await build_upto_header(signer, requirements, expires_at)

Use a remote signing backend

The kms convenience factories return native TransactionSigner backends. AWS and Vault construction is synchronous; GCP construction and all signing are async. Configure the public key that corresponds to your remote Ed25519 key, and keep credentials in your secret manager or environment:

import os

from solana_pay_kit import Operator, kms

operator = Operator(
    signer=kms.vault(
        addr=os.environ["VAULT_ADDR"],       # HTTPS Vault base URL
        path=os.environ["VAULT_KEY_NAME"],   # transit key name
        pubkey=os.environ["OPERATOR_PUBKEY"],
        token=os.environ["VAULT_TOKEN"],
    )
)

For AWS, install solana-pay-kit[aws-kms] and use kms.aws(key_id=..., pubkey=..., region=...) with an ECC_NIST_EDWARDS25519 key. For GCP, install solana-pay-kit[gcp-kms] and use await kms.gcp(key_name=..., pubkey=...) with an EC_SIGN_ED25519 key version. Both accept client= for a preconfigured provider client; Vault accepts http_client=. Other supported backends can be imported directly from solana_keychain; choose one implementing TransactionSigner for transaction settlement.

Create the GCP signer on the event loop that will perform signing, for example in an ASGI application's lifespan handler. Its default gRPC client is bound to that loop: do not create it with a separate asyncio.run(...) at configuration time or share it through synchronous Flask/Django middleware that creates per-request event loops. Those integrations require a loop-neutral provider client or another signing backend.

Run the example

Three runnable examples ship with this package, one per framework. Each one boots zero-config against the Surfpool sandbox.

Boot the server:

git clone https://github.com/solana-foundation/pay-kit
cd pay-kit/python
pip install -e ".[flask]"
python examples/flask/app.py

Consume with pay curl:

# Install the pay CLI:
brew install pay
# or npm install -g @solana/pay

# Fail with 402, payment required
curl -i http://127.0.0.1:8000/report

# Succeed with 200, payment provided
pay curl -i http://127.0.0.1:8000/report

x402

x402 revives HTTP 402 Payment Required as a client-server payment handshake. Your server gates a route; a paying client receives the 402 with payment instructions, signs a Solana transaction off-chain, and replays the same request with a PAYMENT-SIGNATURE header. The server verifies the signature, broadcasts the transaction, and returns the original response with a PAYMENT-RESPONSE header carrying the on-chain settlement signature.

x402 is single-recipient by design: the server's facilitator pays the network fees, the customer's signed transaction settles funds to pay_to. Gates with fee_within or fee_on_top recipients auto-disable x402, because stock x402 facilitators settle to one address.

Scheme Client Server
exact ✅ ✅
upto ✅ ✅
batch-settlement — —

upto charges for actual usage up to a ceiling: the client opens a payment channel depositing the authorized maximum, the handler meters the response and reports it via the Charge dependency, then the gate settles the actual amount and refunds the remainder. It is gated with require_usage / RequireUsage (rather than require_payment) and needs an operator signer.

Client

Use the high-level client to authorize and pay either MPP or x402 challenges. Its safe default permits known stablecoins on mainnet from any HTTP(S) origin, capped at $1 per payment. The fluent policy below narrows origins and adds a higher cap for one exact origin:

import asyncio

from solana_pay_kit import Signer
from solana_pay_kit._paycore.rpc import SolanaRpc
from solana_pay_kit.client import ClientPermissions, OriginPermissionOverride, PayKitClient, usd

async def main():
    signer = Signer.file("payer.json")  # the payer's keypair
    rpc = SolanaRpc("https://api.mainnet-beta.solana.com")
    permissions = (
        ClientPermissions.builder()
        .allow_origin("https://api.example")
        .only_network("mainnet")
        .max_amount_per_payment(usd("1"))
        .override_origin(
            OriginPermissionOverride.builder("https://api.example")
            .max_amount_per_payment(usd("5"))
            .build()
        )
        .build()
    )
    http = (
        PayKitClient.builder()
        .signer(signer)
        .rpc(rpc)
        .network("mainnet")
        .permissions(permissions)
        .build()
    )
    async with http:
        resp = await http.get("https://api.example/report")  # 402 -> pay -> 200
        print(resp.status_code, resp.headers.get("payment-response"))

asyncio.run(main())

Caps can be global or exact-origin overrides. An override changes cap precedence but never grants the origin itself. Unknown assets and non-mainnet networks are denied unless explicitly enabled. Use .permissions(False) only when you intentionally want an unrestricted client. Full precedence and security invariants are in the permission architecture.

The protocol-specific x402 client and low-level building blocks remain exposed: x402_async_client(signer, rpc) returns an auto-paying httpx.AsyncClient; parse_x402_challenge(headers, body, selection) selects an offer, and build_payment_header(signer, rpc, offer) returns the base64 PAYMENT-SIGNATURE value for callers that drive their own HTTP. See examples/x402-client/.

MPP

The Machine Payments Protocol is the broader HTTP Payment Authentication scheme, the same 402 handshake, but the challenge carries a richer intent shape that supports multi-recipient splits, server-side fee accounting, and a separate fee-payer signer.

Use MPP when:

  • Your gate has a platform or gateway fee (the Stripe-Connect "application fee" pattern).
  • You want the server to subsidize the customer's network fee.
  • You want one challenge per gate instead of per-mint-quoted offers.
Intent Client Server
charge/pull ✅ ✅
charge/push — ✅
session ✅ ✅
subscription — —

session ships both sides. Client: ActiveSession voucher signing, the challenge-driven pull/clientVoucher payment-channel openers (fee payer = challenge operator, pending-server-signature placeholder), the metered SessionConsumer, and the SSE streaming helpers (MeteredSseSession, MeteredSseStream, HttpCommitTransport). Server: the session method (new_session) issuing challenges and verifying credentials/vouchers, the server-broadcast open path (openTxSubmitter=server, the operator co-signs and broadcasts the open), the reserve/commit metering side channel (session_routes), the shared channel store, the idle-close watchdog, and on-chain settle-at-close (when a signer and RPC are configured, a closed channel's settledSignature carries the real on-chain signature; without them the close is a state-flip and the signature stays null).

The MPP server owns the full lifecycle: it issues signed challenges with a fresh recentBlockhash, parses and validates the Authorization: Payment credential, pins the echoed charge request, decodes the client-signed transaction and checks recipient, amount, mint, splits, ATA, memos, and compute budget, rejects Surfpool-signed transactions on non-localnet networks, optionally fee-payer co-signs (legacy + v0), broadcasts via sendTransaction, consumes the signature in the replay store, awaits confirmation, and emits payment-receipt with the on-chain signature.


Vocabulary

Term Meaning
gate A protected unit. Has an amount, optional fees, accepted protocols.
amount The base amount a gate charges, before any fee_on_top.
total What the customer pays: amount + sum(fee_on_top). Derived via Gate.total().
price Value object returned by usd(...): number + currency + settlement.
fee_within Fee taken out of the amount. pay_to nets less.
fee_on_top Fee added to the amount. Customer pays more; pay_to nets full.
payment Proof submitted by the client to pass a gate.
protocol Protocol.X402 or Protocol.MPP (top-level dispatch).
scheme x402 sub-form: exact. MPP sub-form: charge.
currency Fiat unit a price is quoted in (USD, EUR, GBP).
accept Ordered preference list (protocols and stablecoins both).
settlement On-chain asset that actually transfers (USDC, USDT).

Three primitives

The framework-agnostic trio, importable from the top level for imperative gating inside a handler, mirrored by the per-framework shims:

Function Purpose
require_payment(request) Returns the verified Payment, raises PaymentRequiredError if unpaid
is_paid(request) Predicate, never raises
get_payment(request) The verified Payment, None until paid

Each framework shim also exposes its own decorator/dependency form: solana_pay_kit.flask.require_payment, solana_pay_kit.fastapi.RequirePayment, and solana_pay_kit.django.require_payment.

Inline pricing

For one-off endpoints that don't warrant a registry entry, skip the gate name and pass a price directly:

@app.get("/oneoff")
@require_payment(usd("0.25"))
def oneoff():
    return jsonify(ok=True)

The bare Price is wrapped into an inline Gate using the configured default recipient and accept list.

Gate DSL

Each gate is a frozen value object built via Gate.build with an amount, an ordered list of accepted protocols, and zero or more named fees.

SELLER = "Ay..."
PLATFORM = "CX..."

# Simple. Customer pays $0.10, pay_to nets $0.10.
Gate.build(name="report", amount=usd("0.10"), description="Premium report")

# x402-only.
Gate.build(name="api_call", amount=usd("0.001"), accept=(Protocol.X402,))

# Stripe-Connect "application fee". Customer pays $10.00,
# SELLER nets $9.70, PLATFORM nets $0.30. x402 auto-disabled.
Gate.build(name="marketplace_sale", amount=usd("10.00"),
           pay_to=SELLER, fee_within={PLATFORM: usd("0.30")})

# Surcharge. Customer pays $10.50, SELLER nets $10.00, PLATFORM nets $0.50.
Gate.build(name="ticket", amount=usd("10.00"),
           pay_to=SELLER, fee_on_top={PLATFORM: usd("0.50")})

# Dynamic per-request pricing.
@gate("tiered")
def tiered(request):
    tier = request.args.get("tier")
    return usd("5.00") if tier == "premium" else usd("0.10")

Boot-time validations (all raise ConfigurationError or a subclass):

  • pay_to is required (gate kwarg or a configured operator recipient).
  • Fee recipient must differ from pay_to. Fold the fee into the amount instead.
  • All fee prices share one denomination with the amount.
  • sum(fee_within) <= amount.
  • accept=(Protocol.X402,) on a fee-bearing gate raises ProtocolIncompatibleError.

Framework-first

solana_pay_kit carries no web-framework dependency in the base install. The framework shims live in optional submodules imported on demand:

  • solana_pay_kit.flask (install solana_pay_kit[flask]), a @require_payment view decorator plus is_paid / payment request accessors.
  • solana_pay_kit.fastapi (install solana_pay_kit[fastapi]), a Django/DRF-style paywall middleware for route metadata and default policies, plus a RequirePayment dependency for Depends(...).
  • solana_pay_kit.django (install solana_pay_kit[django]), a require_payment view decorator and an optional PaymentMiddleware stack form.

Every shim delegates protocol/scheme dispatch and 402-challenge assembly to the host-neutral PayCore; the shim only translates the outcome into its framework's response idioms. A verified Payment is attached to the request (request.state on FastAPI, flask.g on Flask, request.payment on Django) and its settlement headers are merged onto the success response.

# Imperative gating, no decorator, any framework:
from solana_pay_kit import require_payment

def view(request):
    payment = require_payment(request)  # raises PaymentRequiredError if unpaid
    ...

FastAPI apps can avoid duplicating paths in a payment allowlist. Mark paid routes in the route table, then install one paywall:

from fastapi import FastAPI, Request

from solana_pay_kit.fastapi import install_paywall, payment

app = FastAPI()
install_paywall(
    app,
    {
        "enabled": True,
        "network": "solana_localnet",
        "price_usd": "0.01",
        "signer_env": "PAY_OPERATOR_KEY",
    },
    paid_tags=("paid",),
)

@app.post("/v1/chat/completions", tags=["paid"])
async def chat_completions(request: Request):
    verified = payment(request)
    return {"ok": True, "tx": verified.transaction if verified else None}

Use default_policy="paid" to mirror Django's LoginRequiredMiddleware: all matched routes are paid unless marked with @pay_not_required() or a public route tag. Use @pay_required(...) when a route needs a gate other than the app default.

install_paywall() accepts a PayConfig or a plain mapping loaded from TOML/YAML/env:

Field Env suffix Default Purpose
enabled ENABLED False Skip paywall installation unless true.
network NETWORK solana_localnet Solana network for settlement and challenges.
price_usd PRICE_USD 0.01 Default flat price for paid routes.
recipient RECIPIENT None Settlement recipient; falls back to the operator signer.
rpc_url RPC_URL None Override the network default RPC URL.
signer_env SIGNER_ENV EXO_PAY_SIGNER Env var that contains JSON, hex, or base58 key material.
protocols PROTOCOLS x402,mpp Accepted payment protocols.
stablecoins STABLECOINS USDC Accepted settlement assets.
preflight PREFLIGHT True Run boot-time RPC/config validation.

NO_PREFLIGHT=true is the inverse shortcut for preflight=False. NO_PREFLIGHT=false explicitly re-enables preflight.

For lower-level setup, call install_paywall_from_config(app, PaywallConfig(...)) with a concrete gate_ref, pricing, or prebuilt Config. This is useful when you already called solana_pay_kit.configure(...) with custom MppConfig or X402Config and want the paywall to use that exact config.


Examples

Runnable examples ship with this package:

  • examples/simple-server/server.py, the smallest solana_pay_kit server: stdlib http.server with one gated endpoint over the unified solana_pay_kit surface, no web framework.
  • examples/fastapi/app.py, FastAPI server using the route-metadata paywall (install_paywall plus route tags).
  • examples/flask/app.py, Flask server gated with the unified solana_pay_kit surface (@require_payment decorator and the Pricing registry).
  • examples/django/views.py, Django views + URLconf snippet using the @require_payment decorator.

All examples default to solana_localnet, USDC, and the demo recipient. Override the RPC with rpc_url= / PAY_KIT_RPC_URL.

Coverage

cd python
pip install -e ".[dev]"
ruff check src tests
ruff format --check src tests
pyright
pytest --cov=solana_pay_kit --cov-fail-under=90

The solana_pay_kit surface is gated at 90 percent line coverage in CI. The solana_pay_kit.preflight module is omitted from the gate: it wraps live Solana RPC + Surfnet cheatcodes that cannot run inside the offline unit suite, and its two opt-out knobs are covered separately against a stubbed run/RPC.

Harness

The Python server has a direct harness adapter at harness/python-server/server.py, a dual-protocol server that settles both MPP charge and x402-exact. Focused harness commands:

cd harness
MPP_HARNESS_CLIENTS=typescript MPP_HARNESS_SERVERS=python pnpm test
MPP_HARNESS_CLIENTS=rust       MPP_HARNESS_SERVERS=python pnpm test

Spec

This SDK implements the Solana Charge Intent for the HTTP Payment Authentication Scheme, plus the x402 exact scheme on Solana. The wire format, error grammar, and challenge / credential shape are all defined at paymentauth.org.


Repo layout

python/
├── src/solana_pay_kit/                              unified surface over x402 + MPP
│   ├── config.py, operator.py, signer.py, price.py, fee.py, gate.py,
│   │   pricing.py, payment.py, preflight.py, errors.py    # umbrella surface
│   ├── _paycore/                             Currency / Network / Protocol / Stablecoin / Mints / Solana
│   ├── _middleware.py                        host-neutral resolver + require_payment/is_paid/get_payment
│   ├── fastapi.py, flask.py, django.py       framework shims
│   ├── kms.py                                native remote signer factories
│   └── protocols/
│       ├── x402/                             x402-exact adapter (__init__) + verifier/wire shapes (verify.py)
│       └── mpp/                              MPP-charge adapter (__init__) over the consolidated wire layer
│           ├── core/                         canonical JSON, headers, challenge, types, errors, RPC, store
│           ├── intents/charge.py             charge intent
│           ├── server/                       charge handler, middleware, network check, defaults, payment page
│           └── client/                       charge + transport
├── examples/simple-server/                   stdlib http.server, one gated endpoint
├── examples/{fastapi,flask,django}/          solana_pay_kit framework examples
├── tests/                                    pytest suite
└── pyproject.toml

Coding convention

This SDK follows the skills.sh/mindrally/skills/python best-practice skill. Small modules, frozen pydantic value objects, explicit error types with canonical codes, deterministic wire serialization (RFC 8785 canonical JSON), defensive payment verification, and branch tests on security-sensitive paths.

The repo-level pay-sdk-implementation skill remains the protocol source of truth: Rust spec wire format first, Python idioms second.

Migrating MPP session signing to async

Await session operations that sign vouchers or transactions:

  • ActiveSession.prepare_voucher, prepare_increment, sign_voucher, sign_increment, voucher_action, and close_action.
  • build_open_payment_channel_transaction, create_payment_channel_session_opener, and sign_session_authentication.
  • SessionConsumer.commit_directive, MeteredDelivery.ack / commit, and MeteredSseSession.ack / MeteredSseStream.ack.

CommitTransport.commit and HttpCommitTransport.commit remain synchronous. Serialize operations on each ActiveSession or SessionConsumer; they are not safe for concurrent use.

Custom ChannelStore.update_channel implementations must accept synchronous and asynchronous mutators. If the mutator returns an awaitable, await it before persisting its result. Keep exclusive access to that channel throughout the await: operator voucher signing may call a remote signer, and another update must not reuse the same cumulative amount. Pass a detached state snapshot to the mutator and leave committed state unchanged if signing fails or is cancelled. MemoryChannelStore already follows this contract with a per-channel lock.

License

MIT