DEV Community

Cover image for MPP TestKit - PY SDK
MPP TestKit
MPP TestKit

Posted on

MPP TestKit - PY SDK

#web3 #python #api #blockchain

The Python MPP SDK (mpp-test-sdk on PyPI) lets you add pay-per-request billing to any Python API in minutes. No Stripe, no API keys, no billing database — just on-chain Solana payments verified server-side.

Why HTTP 402 in Python?

Python is where most ML models, data APIs, and AI services live. If you're serving embeddings, running inference, or returning processed data, you're either giving it away, hiding it behind a subscription, or building out a billing system you didn't want to build.

HTTP 402 offers a third option: charge per request, on-chain, with no infrastructure. The client pays before you serve. The server verifies on-chain. No shared secrets, no session management, no stripe webhooks.

The Python MPP SDK brings this to Flask, FastAPI, and any Python HTTP framework.

Installation 

pip install mpp-test-sdk
Enter fullscreen mode Exit fullscreen mode

Server (Flask) 

from flask import Flask, jsonify
from mpp_test_sdk import create_test_server

app = Flask(__name__)
server = create_test_server(network="devnet")
print("Recipient:", server.recipient_address)

@app.route("/api/data")
@server.charge(amount="0.001")
def data():
    return jsonify({"result": "here is your data"})

if __name__ == "__main__":
    app.run(port=3001)
Enter fullscreen mode Exit fullscreen mode

The @server.charge(amount="0.001") decorator handles everything:

  • Returns 402 with a Payment-Request header when no receipt is present
  • Verifies the Solana transaction when a Payment-Receipt header is present
  • Calls your handler only after payment is confirmed on-chain

Server (FastAPI) 

from fastapi import FastAPI, Request
from mpp_test_sdk import create_test_server

app = FastAPI()
server = create_test_server(network="devnet")

@app.get("/api/data")
async def data(request: Request):
    result = await server.charge_async(request, amount="0.001")
    if result is not None:
        return result  # 402 or 403 response
    return {"result": "here is your data"}
Enter fullscreen mode Exit fullscreen mode

Client 

import asyncio
from mpp_test_sdk import create_test_client

async def main():
    client = await create_test_client(
        network="devnet",
        on_step=lambda step: print(f"[{step.type}] {step.message}"),
    )
    print("Wallet:", client.address)

    response = await client.fetch("http://localhost:3001/api/data")
    print(response.json())  # {"result": "here is your data"}

asyncio.run(main())
Enter fullscreen mode Exit fullscreen mode

create_test_client generates a Solana keypair, airdrops 2 SOL on devnet automatically, and returns a client whose fetch handles the entire 402 flow.

What happens under the hood 

Client                          Server
  |── GET /api/data ─────────────>|
  |<── 402 ───────────────────────|  Payment-Request: solana; amount="0.001"; recipient="9xK..."
  |                               |
  |── [build + sign Solana tx] ───|  (pure Python ed25519 signing, no wallet lib needed)
  |── [submit + confirm on-chain] |
  |                               |
  |── GET /api/data ─────────────>|  Payment-Receipt: solana; signature="3xK..."; amount="0.001"
  |                               |── [verify tx on Solana RPC]
  |<── 200 OK ────────────────────|
Enter fullscreen mode Exit fullscreen mode

No third-party wallet library. The SDK builds the Solana legacy transaction wire format from scratch using standard Python cryptography.

Lifecycle callbacks 

def on_step(step):
    match step.type:
        case "wallet-created": print("Wallet:", step.data["address"])
        case "funded":         print("Funded 2 SOL via devnet airdrop")
        case "payment":        print(f"Paying {step.data['amount']} SOL")
        case "success":        print("Paid and served:", step.data["status"])

client = await create_test_client(network="devnet", on_step=on_step)
Enter fullscreen mode Exit fullscreen mode

Steps: wallet-created, funded, request, payment, retry, success, error.

Mainnet 

client = await create_test_client(
    network="mainnet",
    secret_key=bytes(my_keypair_bytes),  # 32-byte seed or 64-byte keypair
)
Enter fullscreen mode Exit fullscreen mode

No airdrop on mainnet — pass a pre-funded keypair. Everything else is identical.

Shared client (drop-in requests replacement) 

from mpp_test_sdk import mpp_fetch

response = await mpp_fetch("http://localhost:3001/api/data")
Enter fullscreen mode Exit fullscreen mode

Uses a lazily-created shared devnet wallet. Call reset_mpp_fetch() to force a new wallet.

Error handling 

from mpp_test_sdk import MppFaucetError, MppPaymentError, MppTimeoutError

try:
    response = await client.fetch("http://localhost:3001/api/data")
except MppFaucetError as e:
    print("Airdrop failed for wallet:", e.address)
except MppPaymentError as e:
    print("Payment failed:", e.status)
except MppTimeoutError as e:
    print("Timed out after:", e.timeout_ms, "ms")
Enter fullscreen mode Exit fullscreen mode

Testing your endpoint 

import pytest
from mpp_test_sdk import create_test_client, create_test_server

@pytest.mark.asyncio
async def test_charges_per_request():
    server = create_test_server(network="devnet")
    # ... start your Flask/FastAPI app in a test thread
    client = await create_test_client(network="devnet")
    response = await client.fetch("http://localhost:3001/api/data")
    assert response.status_code == 200
Enter fullscreen mode Exit fullscreen mode

Links

Top comments (0)