Looking for framework-specific implementations?
Basic validation (framework-agnostic)
At a minimum, validation has three steps:- Extract x402 token from the
payment-signatureheader - Verify permissions using the facilitator
- Return
402 Payment Requiredwhen missing/invalid
- TypeScript
- Python
import { Payments, buildPaymentRequired } from '@nevermined-io/payments'
const payments = Payments.getInstance({
nvmApiKey: process.env.NVM_API_KEY!,
environment: 'sandbox'
})
export async function validateRequest(
x402Token: string,
planId: string,
agentId: string,
endpoint: string,
httpVerb: string
) {
// Build payment requirements
const paymentRequired = buildPaymentRequired({
planId,
endpoint,
agentId,
httpVerb
})
// Verify permissions
const verification = await payments.facilitator.verifyPermissions({
paymentRequired,
x402AccessToken: x402Token,
maxAmount: BigInt(1)
})
return {
isValid: verification.isValid,
balance: verification.balance,
subscriberAddress: verification.subscriberAddress
}
}
// Pseudo-usage
// - Extract token from payment-signature header
// - Call validateRequest(token, planId, agentId, endpoint, httpVerb)
// - If invalid, return 402
import os
from payments_py import Payments, PaymentOptions
from payments_py.x402.helpers import build_payment_required
payments = Payments.get_instance(
PaymentOptions(nvm_api_key=os.environ['NVM_API_KEY'], environment='sandbox')
)
def validate_request(
x402_token: str,
plan_id: str,
agent_id: str,
endpoint: str,
http_verb: str
) -> dict:
# Build payment requirements
payment_required = build_payment_required(
plan_id=plan_id,
endpoint=endpoint,
agent_id=agent_id,
http_verb=http_verb
)
# Verify permissions
verification = payments.facilitator.verify_permissions(
payment_required=payment_required,
x402_access_token=x402_token,
max_amount="1"
)
return {
'is_valid': verification.is_valid,
'balance': verification.balance,
'subscriber_address': verification.subscriber_address
}
# Pseudo-usage:
# - Extract token from payment-signature header
# - Call validate_request(token, plan_id, agent_id, endpoint, http_verb)
# - If invalid, return 402
Validation with minimum credits
Check that the user has enough credits for the operation:- TypeScript
- Python
import { buildPaymentRequired } from '@nevermined-io/payments'
export async function validateWithMinimum(
x402Token: string,
planId: string,
agentId: string,
endpoint: string,
minCredits: number
) {
const paymentRequired = buildPaymentRequired({
planId, endpoint, agentId, httpVerb: 'POST'
})
const result = await payments.facilitator.verifyPermissions({
paymentRequired,
x402AccessToken: x402Token,
maxAmount: BigInt(minCredits)
})
if (!result.isValid) {
return { valid: false, reason: 'invalid_token' as const }
}
if (result.balance < minCredits) {
return {
valid: false,
reason: 'insufficient_credits' as const,
required: minCredits,
available: result.balance
}
}
return { valid: true, balance: result.balance }
}
// Example: expensive operation requiring 10 credits
const validation = await validateWithMinimum(token, PLAN_ID, AGENT_ID, '/query', 10)
if (!validation.valid) {
// return 402
}
from payments_py.x402.helpers import build_payment_required
async def validate_with_minimum(
x402_token: str,
plan_id: str,
agent_id: str,
endpoint: str,
min_credits: int
) -> dict:
payment_required = build_payment_required(
plan_id=plan_id, endpoint=endpoint,
agent_id=agent_id, http_verb="POST"
)
result = payments.facilitator.verify_permissions(
payment_required=payment_required,
x402_access_token=x402_token,
max_amount=str(min_credits)
)
if not result.is_valid:
return {'valid': False, 'reason': 'invalid_token'}
if result.balance < min_credits:
return {
'valid': False,
'reason': 'insufficient_credits',
'required': min_credits,
'available': result.balance
}
return {'valid': True, 'balance': result.balance}
# Example: expensive operation requiring 10 credits
validation = await validate_with_minimum(token, PLAN_ID, AGENT_ID, "/query", 10)
if not validation['valid']:
# return 402
pass
Validation response handling
Handle all possible validation outcomes:- TypeScript
- Python
async function handleValidation(
x402Token: string,
paymentRequired: string,
maxAmount: bigint
) {
try {
const result = await payments.facilitator.verifyPermissions({
paymentRequired,
x402AccessToken: x402Token,
maxAmount
})
if (!result.isValid) {
// Determine specific error
switch (result.reason) {
case 'TOKEN_EXPIRED':
return {
status: 402,
error: 'Access token has expired',
code: 'TOKEN_EXPIRED',
action: 'refresh_token'
}
case 'INSUFFICIENT_BALANCE':
return {
status: 402,
error: 'Insufficient credits',
code: 'INSUFFICIENT_BALANCE',
action: 'purchase_credits'
}
case 'PLAN_EXPIRED':
return {
status: 402,
error: 'Plan has expired',
code: 'PLAN_EXPIRED',
action: 'renew_plan'
}
default:
return {
status: 402,
error: 'Invalid token',
code: 'INVALID_TOKEN',
action: 'get_new_token'
}
}
}
return { status: 200, balance: result.balance }
} catch (error) {
console.error('Validation error:', error)
return {
status: 500,
error: 'Validation service unavailable',
code: 'SERVICE_ERROR'
}
}
}
async def handle_validation(
x402_token: str,
payment_required: str,
max_amount: str
) -> dict:
try:
result = payments.facilitator.verify_permissions(
payment_required=payment_required,
x402_access_token=x402_token,
max_amount=max_amount
)
if not result.is_valid:
reason = getattr(result, 'reason', 'UNKNOWN')
error_map = {
'TOKEN_EXPIRED': {
'status': 402,
'error': 'Access token has expired',
'code': 'TOKEN_EXPIRED',
'action': 'refresh_token'
},
'INSUFFICIENT_BALANCE': {
'status': 402,
'error': 'Insufficient credits',
'code': 'INSUFFICIENT_BALANCE',
'action': 'purchase_credits'
},
'PLAN_EXPIRED': {
'status': 402,
'error': 'Plan has expired',
'code': 'PLAN_EXPIRED',
'action': 'renew_plan'
}
}
return error_map.get(reason, {
'status': 402,
'error': 'Invalid token',
'code': 'INVALID_TOKEN',
'action': 'get_new_token'
})
return {'status': 200, 'balance': result.balance}
except Exception as e:
print(f'Validation error: {e}')
return {
'status': 500,
'error': 'Validation service unavailable',
'code': 'SERVICE_ERROR'
}
Caching validation results
For high-traffic endpoints, cache validation briefly:- TypeScript
- Python
import { LRUCache } from 'lru-cache'
const validationCache = new LRUCache<string, any>({
max: 1000,
ttl: 1000 * 30 // 30 seconds
})
async function validateWithCache(
x402Token: string,
paymentRequired: string,
maxAmount: bigint
) {
// Create cache key from token hash
const cacheKey = hashToken(x402Token)
const cached = validationCache.get(cacheKey)
if (cached) {
return cached
}
const result = await payments.facilitator.verifyPermissions({
paymentRequired,
x402AccessToken: x402Token,
maxAmount
})
if (result.isValid) {
validationCache.set(cacheKey, result)
}
return result
}
function hashToken(token: string): string {
return crypto.createHash('sha256').update(token).digest('hex').slice(0, 16)
}
from datetime import datetime, timedelta
import hashlib
validation_cache = {}
CACHE_TTL = timedelta(seconds=30)
async def validate_with_cache(
x402_token: str,
payment_required: str,
max_amount: str
) -> dict:
cache_key = hash_token(x402_token)
# Check cache
if cache_key in validation_cache:
cached, timestamp = validation_cache[cache_key]
if datetime.now() - timestamp < CACHE_TTL:
return cached
result = payments.facilitator.verify_permissions(
payment_required=payment_required,
x402_access_token=x402_token,
max_amount=max_amount
)
if result.is_valid:
validation_cache[cache_key] = (result, datetime.now())
return result
def hash_token(token: str) -> str:
return hashlib.sha256(token.encode()).hexdigest()[:16]
Next steps
Charge Credits
Deduct credits for requests
Subscription Access
Time-based access patterns