Skip to main content
Start here: need to register a service and create a plan first or get your Nevermined API Key? Follow the 5-minute setup.
Complete Python quick start for adding payments to your AI agent or service.

Installation

pip install payments-py fastapi uvicorn

Environment Setup

# Your Nevermined API key from nevermined.app
export NVM_API_KEY="nvm:your-api-key"

# Your wallet address to receive payments
export BUILDER_ADDRESS="0xYourWalletAddress"

# After registration, set these
export AGENT_ID="did:nv:your-agent-id"
export PLAN_ID="did:nv:your-plan-id"

Complete Example

import os
from fastapi import FastAPI, Request, HTTPException, Depends
from pydantic import BaseModel
from payments_py import Payments, PaymentOptions
from payments_py.plans import get_erc20_price_config, get_fixed_credits_config

app = FastAPI(title="My AI Agent")

# In this example we require a payment a payment of 10 USDC for 100 requests
# For that USDC payment we use USDC on Base Sepolia, so we need its contract address:
USDC_ADDRESS = '0x036CbD53842c5426634e7929541eC2318f3dCF7e'

# Initialize payments
payments = Payments.get_instance(
    PaymentOptions(
        nvm_api_key=os.environ['NVM_API_KEY'],
        environment='sandbox'
    )
)

# Store IDs after registration
agent_id = os.environ.get('AGENT_ID', '')
plan_id = os.environ.get('PLAN_ID', '')


# ============================================
# MODELS
# ============================================
class QueryRequest(BaseModel):
    prompt: str


class QueryResponse(BaseModel):
    result: str
    credits_remaining: int


# ============================================
# SETUP: Run once to register your agent
# ============================================
@app.post("/setup")
async def register_agent():
    global agent_id, plan_id

    result = payments.agents.register_agent_and_plan(
        agent_metadata={
            'name': 'My Python AI Agent',
            'description': 'AI assistant built with Python',
            'tags': ['ai', 'python']
        },
        agent_api={
            'endpoints': [{'POST': 'http://localhost:8000/query'}]
        },
        plan_metadata={
            'name': 'Starter Plan',
            'description': '100 queries'
        },
        price_config=get_erc20_price_config(
            10_000_000,  # 10 USDC
            USDC_ADDRESS,
            os.environ['BUILDER_ADDRESS']
        ),
        credits_config=get_fixed_credits_config(100, 1),
        access_limit='credits'
    )

    agent_id = result['agentId']
    plan_id = result['planId']

    return {
        'agent_id': agent_id,
        'plan_id': plan_id,
        'message': 'Add these to your .env file!'
    }


# ============================================
# MIDDLEWARE: Validate payments
# ============================================
async def validate_payment(request: Request) -> int:
    """Validate payment and return remaining credits."""
    from payments_py.x402.helpers import build_payment_required

    x402_token = request.headers.get("payment-signature")

    if not x402_token:
        raise HTTPException(status_code=402, detail="Payment required")

    # Build payment requirements
    payment_required = build_payment_required(
        plan_id=plan_id,
        endpoint=str(request.url),
        agent_id=agent_id,
        http_verb=request.method
    )

    # Verify permissions
    verification = payments.facilitator.verify_permissions(
        payment_required=payment_required,
        x402_access_token=x402_token,
        max_amount="1"
    )

    if not verification.get("is_valid"):
        raise HTTPException(status_code=402, detail="Invalid payment")

    return verification.get("remaining_credits", 0)


# ============================================
# ROUTES
# ============================================
@app.get("/health")
async def health():
    """Health check (public)."""
    return {
        'status': 'ok',
        'agent_id': agent_id,
        'plan_id': plan_id
    }


@app.post("/query", response_model=QueryResponse)
async def query(
    request: QueryRequest,
    credits: int = Depends(validate_payment)
):
    """Protected AI endpoint."""
    # Your AI logic here
    result = f'You asked: "{request.prompt}". Here\'s my response...'

    return QueryResponse(
        result=result,
        credits_remaining=credits
    )


# ============================================
# MAIN
# ============================================
if __name__ == '__main__':
    import uvicorn

    print(f"Agent ID: {agent_id or 'Not registered'}")
    print(f"Plan ID: {plan_id or 'Not registered'}")

    if not agent_id or not plan_id:
        print("No agent registered yet. POST to /setup to register.")

    uvicorn.run(app, host='0.0.0.0', port=8000)

Run the Example

# Run the server
python main.py

# Or with uvicorn directly
uvicorn main:app --reload --port 8000

Test the Flow

1) Register your agent (first time only)
curl -X POST http://localhost:8000/setup
Save the returned agent_id and plan_id to your environment. 2) Try without payment
curl -X POST http://localhost:8000/query \
  -H "Content-Type: application/json" \
  -d '{"prompt": "Hello"}'
3) As a subscriber, purchase and query
import os
import requests
from payments_py import Payments, PaymentOptions

payments = Payments.get_instance(
    PaymentOptions(
        nvm_api_key=os.environ['SUBSCRIBER_API_KEY'],
        environment='sandbox'
    )
)

def purchase_and_query():
    plan_id = os.environ['PLAN_ID']
    agent_id = os.environ['AGENT_ID']

    # 1. Order the plan
    print('Purchasing plan...')
    order_result = payments.plans.order_plan(plan_id)

    # 2. Get access token
    print('Getting access token...')
    token_res = payments.x402.get_x402_access_token(plan_id, agent_id)
    access_token = token_res['accessToken']

    # 3. Query the agent
    print('Querying agent...')
    response = requests.post(
        'http://localhost:8000/query',
        headers={
            'Content-Type': 'application/json',
            'Payment-Signature': f'{access_token}'
        },
        json={'prompt': 'What is 2+2?'}
    )

    result = response.json()
    print(f'Result: {result}')

if __name__ == '__main__':
    purchase_and_query()

Flask Alternative

import os
from flask import Flask, request, jsonify
from functools import wraps
from payments_py import Payments, PaymentOptions
from payments_py.plans import get_erc20_price_config, get_fixed_credits_config

app = Flask(__name__)

USDC_ADDRESS = '0x036CbD53842c5426634e7929541eC2318f3dCF7e'

payments = Payments.get_instance(
    PaymentOptions(
        nvm_api_key=os.environ['NVM_API_KEY'],
        environment='sandbox'
    )
)

agent_id = os.environ.get('AGENT_ID', '')
plan_id = os.environ.get('PLAN_ID', '')


def require_payment(f):
    """Decorator to validate payment."""
    from payments_py.x402.helpers import build_payment_required

    @wraps(f)
    def decorated(*args, **kwargs):
        x402_token = request.headers.get("payment-signature")

        if not x402_token:
            return {"error": "Payment Required"}, 402

        # Get request body for validation (Flask synchronous method)
        body = request.get_json()

        # Build payment requirements
        payment_required = build_payment_required(
            plan_id=plan_id,
            endpoint=request.url,
            agent_id=agent_id,
            http_verb=request.method
        )

        # Verify permissions
        verification = payments.facilitator.verify_permissions(
            payment_required=payment_required,
            x402_access_token=x402_token,
            max_amount="1"
        )

        if not verification.get("is_valid"):
            return {"error": "Invalid payment"}, 402

        # Get balance
        balance = payments.plans.get_plan_balance(plan_id)

        # Pass credits to the handler
        kwargs['credits'] = balance.get('balance', 0)
        return f(*args, **kwargs)

    return decorated


@app.route('/health')
def health():
    return jsonify({'status': 'ok', 'agent_id': agent_id, 'plan_id': plan_id})


@app.route('/query', methods=['POST'])
@require_payment
def query(credits=0):
    prompt = request.json.get('prompt', '')

    # Your AI logic here
    result = f'You asked: "{prompt}". Here\'s my response...'

    return jsonify({
        'result': result,
        'credits_remaining': credits
    })


if __name__ == '__main__':
    app.run(port=8000, debug=True)

Project Structure

my-ai-agent/
├── main.py              # FastAPI server
├── requirements.txt
├── .env
└── src/
    ├── middleware/
    │   └── payments.py  # Payment validation
    ├── routes/
    │   └── query.py     # AI endpoints
    └── services/
        └── ai.py        # Your AI logic

Requirements File

payments-py>=1.0.0
fastapi>=0.100.0
uvicorn>=0.23.0
python-dotenv>=1.0.0

Next Steps