Start here: need to register a service and create a plan first or get your Nevermined API Key? Follow the
5-minute setup.
Installation
pip install payments-py fastapi uvicorn
Environment Setup
# Your Nevermined API key (see: /docs/getting-started/get-your-api-key)
export NVM_API_KEY="sandbox: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 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': [{'verb': 'POST', 'url': '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
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"}'
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
FastAPI Integration
Complete FastAPI patterns and best practices
Python SDK Reference
Full API documentation
Payment Patterns
Advanced patterns for production
x402 Protocol
Implement HTTP 402 payment flows