Validate Requests

Patterns for validating payment tokens in your API, agent, or protected resource

Copy-paste patterns for validating payment requests for APIs, agents/tools, and protected resources.

Looking for framework-specific implementations?

• Express.js Integration
• FastAPI Integration
• Generic HTTP Integration

Basic validation (framework-agnostic)

At a minimum, validation has three steps:

1. Extract x402 token from the payment-signature header
2. Verify permissions using the facilitator
3. Return 402 Payment Required when missing/invalid

TypeScript

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

Python

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

export async function validateWithMinimum(
  token: string,
  body: unknown,
  minCredits: number
) {
  const result = await payments.requests.isValidRequest(token, body)

  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
const validation = await validateWithMinimum(token, body, 10)
if (!validation.valid) {
  // return 402
}

Python

async def validate_with_minimum(token: str, body: dict, min_credits: int) -> dict:
    result = await payments.requests.is_valid_request(token, body)

    if not result['isValid']:
        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
validation = await validate_with_minimum(token, body, 10)
if not validation['valid']:
    # return 402
    pass

Validation response handling

Handle all possible validation outcomes:

TypeScript

async function handleValidation(token: string, body: any) {
  try {
    const result = await payments.requests.isValidRequest(token, body)

    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'
    }
  }
}

Python

async def handle_validation(token: str, body: dict) -> dict:
    try:
        result = await payments.requests.is_valid_request(token, body)

        if not result['isValid']:
            reason = result.get('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

import { LRUCache } from 'lru-cache'

const validationCache = new LRUCache<string, any>({
  max: 1000,
  ttl: 1000 * 30 // 30 seconds
})

async function validateWithCache(token: string, body: any) {
  // Create cache key from token hash
  const cacheKey = hashToken(token)
  const cached = validationCache.get(cacheKey)

  if (cached) {
    return cached
  }

  const result = await payments.requests.isValidRequest(token, body)

  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)
}

Python

from datetime import datetime, timedelta
import hashlib

validation_cache = {}
CACHE_TTL = timedelta(seconds=30)

async def validate_with_cache(token: str, body: dict) -> dict:
    cache_key = hash_token(token)

    # Check cache
    if cache_key in validation_cache:
        cached, timestamp = validation_cache[cache_key]
        if datetime.now() - timestamp < CACHE_TTL:
            return cached

    result = await payments.requests.is_valid_request(token, body)

    if result['isValid']:
        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