Charge Credits

Copy-paste patterns for deducting credits from subscribers.

Automatic Credit Deduction

When using the standard Bearer token flow, credits are automatically deducted when you validate the request:

TypeScript
Python

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

// Credits are automatically deducted based on plan configuration
// result.balance shows remaining credits AFTER deduction
console.log(`Remaining credits: ${result.balance}`)

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

# Credits are automatically deducted based on plan configuration
# result['balance'] shows remaining credits AFTER deduction
print(f"Remaining credits: {result['balance']}")

Variable Credit Charges

For operations with variable costs, use the x402 flow:

TypeScript
Python

// Calculate cost based on request
function calculateCost(request: any): number {
  const baseCredits = 1

  // Add cost for complexity
  if (request.options?.highQuality) {
    return baseCredits + 5
  }

  if (request.prompt.length > 1000) {
    return baseCredits + 2
  }

  return baseCredits
}

// Use in settlement
const cost = calculateCost(req.body)

await payments.facilitator.settlePermissions({
  planId: PLAN_ID,
  maxAmount: BigInt(cost),
  x402AccessToken: token,
  subscriberAddress: address
})

def calculate_cost(request: dict) -> int:
    base_credits = 1

    # Add cost for complexity
    if request.get('options', {}).get('high_quality'):
        return base_credits + 5

    if len(request.get('prompt', '')) > 1000:
        return base_credits + 2

    return base_credits

# Use in settlement
cost = calculate_cost(request)

await facilitator.settle(payment_payload, PaymentRequirements(
    plan_id=PLAN_ID,
    agent_id=AGENT_ID,
    max_amount=cost,
    extra={'subscriber_address': address}
))

Credit Tracking Middleware

Track credits used per request:

TypeScript
Python

interface CreditUsage {
  requestId: string
  subscriberAddress: string
  creditsUsed: number
  creditsBefore: number
  creditsAfter: number
  timestamp: Date
  endpoint: string
}

async function trackCredits(
  req: Request,
  res: Response,
  next: NextFunction
) {
  const startBalance = req.payment?.balance || 0

  // Store original json method
  const originalJson = res.json.bind(res)

  // Override to track response
  res.json = (body: any) => {
    const creditsUsed = body.credits?.used || 1
    const creditsAfter = body.credits?.remaining || startBalance - creditsUsed

    const usage: CreditUsage = {
      requestId: req.id,
      subscriberAddress: req.payment?.subscriberAddress || 'unknown',
      creditsUsed,
      creditsBefore: startBalance,
      creditsAfter,
      timestamp: new Date(),
      endpoint: req.path
    }

    // Log or store usage
    logCreditUsage(usage)

    return originalJson(body)
  }

  next()
}

function logCreditUsage(usage: CreditUsage) {
  console.log(`Credit usage: ${JSON.stringify(usage)}`)
  // Or send to analytics, database, etc.
}

from dataclasses import dataclass
from datetime import datetime
import uuid

@dataclass
class CreditUsage:
    request_id: str
    subscriber_address: str
    credits_used: int
    credits_before: int
    credits_after: int
    timestamp: datetime
    endpoint: str

async def track_credits(request: Request, call_next):
    # Get initial balance from payment validation
    start_balance = getattr(request.state, 'credits_balance', 0)
    request_id = str(uuid.uuid4())

    response = await call_next(request)

    # Track usage after response
    if hasattr(request.state, 'credits_used'):
        usage = CreditUsage(
            request_id=request_id,
            subscriber_address=getattr(request.state, 'subscriber_address', 'unknown'),
            credits_used=request.state.credits_used,
            credits_before=start_balance,
            credits_after=start_balance - request.state.credits_used,
            timestamp=datetime.now(),
            endpoint=str(request.url.path)
        )

        log_credit_usage(usage)

    return response

def log_credit_usage(usage: CreditUsage):
    print(f"Credit usage: {usage}")
    # Or send to analytics, database, etc.

Tiered Pricing

Charge different amounts based on usage tiers:

TypeScript
Python

interface PricingTier {
  name: string
  minTokens: number
  maxTokens: number
  creditsPerRequest: number
}

const PRICING_TIERS: PricingTier[] = [
  { name: 'small', minTokens: 0, maxTokens: 100, creditsPerRequest: 1 },
  { name: 'medium', minTokens: 101, maxTokens: 500, creditsPerRequest: 3 },
  { name: 'large', minTokens: 501, maxTokens: 2000, creditsPerRequest: 5 },
  { name: 'xlarge', minTokens: 2001, maxTokens: Infinity, creditsPerRequest: 10 }
]

function getCreditsForTokens(tokenCount: number): number {
  const tier = PRICING_TIERS.find(
    t => tokenCount >= t.minTokens && tokenCount <= t.maxTokens
  )
  return tier?.creditsPerRequest || 1
}

// Usage
async function processWithTieredPricing(prompt: string, payment: PaymentInfo) {
  const estimatedTokens = estimateTokens(prompt)
  const requiredCredits = getCreditsForTokens(estimatedTokens)

  if (payment.balance < requiredCredits) {
    throw new Error(`Insufficient credits. Need ${requiredCredits}, have ${payment.balance}`)
  }

  const result = await generateResponse(prompt)
  const actualTokens = countTokens(result)
  const actualCredits = getCreditsForTokens(actualTokens)

  return {
    result,
    credits: {
      estimated: requiredCredits,
      actual: actualCredits,
      remaining: payment.balance - actualCredits
    }
  }
}

from dataclasses import dataclass
from typing import List

@dataclass
class PricingTier:
    name: str
    min_tokens: int
    max_tokens: int
    credits_per_request: int

PRICING_TIERS: List[PricingTier] = [
    PricingTier('small', 0, 100, 1),
    PricingTier('medium', 101, 500, 3),
    PricingTier('large', 501, 2000, 5),
    PricingTier('xlarge', 2001, float('inf'), 10)
]

def get_credits_for_tokens(token_count: int) -> int:
    for tier in PRICING_TIERS:
        if tier.min_tokens <= token_count <= tier.max_tokens:
            return tier.credits_per_request
    return 1

# Usage
async def process_with_tiered_pricing(prompt: str, payment: dict) -> dict:
    estimated_tokens = estimate_tokens(prompt)
    required_credits = get_credits_for_tokens(estimated_tokens)

    if payment['balance'] < required_credits:
        raise ValueError(
            f"Insufficient credits. Need {required_credits}, have {payment['balance']}"
        )

    result = await generate_response(prompt)
    actual_tokens = count_tokens(result)
    actual_credits = get_credits_for_tokens(actual_tokens)

    return {
        'result': result,
        'credits': {
            'estimated': required_credits,
            'actual': actual_credits,
            'remaining': payment['balance'] - actual_credits
        }
    }

Pre-Authorization Pattern

For long-running operations, pre-authorize credits before starting:

TypeScript
Python

async function processLongRunningTask(
  token: string,
  body: any,
  estimatedCredits: number
) {
  // Step 1: Validate and check balance
  const validation = await payments.requests.isValidRequest(token, body)

  if (!validation.isValid) {
    throw new Error('Invalid token')
  }

  if (validation.balance < estimatedCredits) {
    throw new Error(`Insufficient credits. Need ${estimatedCredits}, have ${validation.balance}`)
  }

  // Step 2: Start the long-running task
  const taskId = await startTask(body)

  try {
    // Step 3: Wait for completion
    const result = await waitForCompletion(taskId)

    // Step 4: Calculate actual cost and charge
    const actualCredits = calculateActualCost(result)

    // Note: For x402 flow, settlement happens here
    // For standard flow, credits were already deducted during validation

    return {
      result,
      credits: {
        estimated: estimatedCredits,
        actual: actualCredits,
        remaining: validation.balance - actualCredits
      }
    }
  } catch (error) {
    // Task failed - don't charge (or refund in x402 flow)
    console.error('Task failed:', error)
    throw error
  }
}

async def process_long_running_task(
    token: str,
    body: dict,
    estimated_credits: int
) -> dict:
    # Step 1: Validate and check balance
    validation = payments.requests.is_valid_request(token, body)

    if not validation['isValid']:
        raise ValueError('Invalid token')

    if validation['balance'] < estimated_credits:
        raise ValueError(
            f"Insufficient credits. Need {estimated_credits}, have {validation['balance']}"
        )

    # Step 2: Start the long-running task
    task_id = await start_task(body)

    try:
        # Step 3: Wait for completion
        result = await wait_for_completion(task_id)

        # Step 4: Calculate actual cost
        actual_credits = calculate_actual_cost(result)

        return {
            'result': result,
            'credits': {
                'estimated': estimated_credits,
                'actual': actual_credits,
                'remaining': validation['balance'] - actual_credits
            }
        }
    except Exception as error:
        # Task failed - don't charge
        print(f'Task failed: {error}')
        raise

Credit Response Headers

Include credit information in response headers:

TypeScript
Python

function addCreditHeaders(res: Response, credits: CreditInfo) {
  res.setHeader('X-Credits-Used', credits.used.toString())
  res.setHeader('X-Credits-Remaining', credits.remaining.toString())
  res.setHeader('X-Credits-Plan', credits.planId)
}

// Usage in middleware
app.use((req, res, next) => {
  const originalJson = res.json.bind(res)

  res.json = (body: any) => {
    if (req.payment && body.credits) {
      addCreditHeaders(res, body.credits)
    }
    return originalJson(body)
  }

  next()
})

from fastapi import Response

def add_credit_headers(response: Response, credits: dict):
    response.headers['X-Credits-Used'] = str(credits['used'])
    response.headers['X-Credits-Remaining'] = str(credits['remaining'])
    response.headers['X-Credits-Plan'] = credits['plan_id']

# Usage in endpoint
@app.post("/query")
async def query(request: dict, response: Response):
    result = await process_query(request)

    add_credit_headers(response, {
        'used': 1,
        'remaining': 99,
        'plan_id': PLAN_ID
    })

    return result

Start here

Use cases

Platform / Guides

Automatic Credit Deduction

Variable Credit Charges

Credit Tracking Middleware

Tiered Pricing

Pre-Authorization Pattern

Credit Response Headers

Next Steps

Subscription Access

Dynamic Pricing

Start here

Use cases

Platform / Guides

​Automatic Credit Deduction

​Variable Credit Charges

​Credit Tracking Middleware

​Tiered Pricing

​Pre-Authorization Pattern

​Credit Response Headers

​Next Steps

Subscription Access

Dynamic Pricing

Automatic Credit Deduction

Variable Credit Charges

Credit Tracking Middleware

Tiered Pricing

Pre-Authorization Pattern

Credit Response Headers

Next Steps