Handle Requests

Once a user (or agent) purchases a Payment Plan, if this Plan has some AI Agents or Services attached to it, the user can query these AI Agents or Services. To facilitate AI Agents to authorize only the requests of users with a valid Payment Plan, the Payments libraries provide a simple API to do this validation simple and secure.

Authorizing only valid requests to my AI Agents

All the authorization can be done just calling the requests.startProcessingRequest method. This method will receive the access token sent by the user, and will validate:

The user is a subscriber of any of the payment plans giving access to the AI Agent.
The endpoint requested and HTTP method is allowed because was included as part of the AI Agent registration.
The user has enough credits to pay for the request (if the AI Agent is using a credit-based Payment Plan) or the payment plan didn’t expire (if it’s a time-based subscription).

In the example below we are gonna start a simple HTTP server that first thing is gonna do is to validate the request using the startProcessingRequest method. If the request is valid, it will return a 200 OK response, otherwise it will return a 402 Payment Required response.

TypeScript
Python

import http from 'http'

const agentHost = 'https://example.com' // The AI Agent is running in this host

const server = http.createServer(async (req, res) => {
  const authHeader = req.headers['authorization'] as string
  const requestedUrl = `${agentHost}${req.url}`
  const httpVerb = req.method
  console.log('Received request:', { endpoint: requestedUrl, httpVerb, authHeader })
   
  try {
    const isValidReq = await payments.requests.startProcessingRequest(
      agentId,
      authHeader,
      requestedUrl,
      httpVerb!,
    )
    console.log('isValidReq', isValidReq)
    if (isValidReq.balance.isSubscriber) {
      res.writeHead(200, { 'Content-Type': 'application/json' })
      res.end(JSON.stringify({ message: 'Hello from the Agent!' }))
      return
    }
  } catch (error) {
    console.log('Unauthorized access attempt:', authHeader)
    console.log('Error details:', error)
  }

  res.writeHead(402, { 'Content-Type': 'application/json' })
  res.end(JSON.stringify({ error: 'Payment Required' }))
  return
})

server.listen(8889, () => {
  console.log('AI Agent server running on port 8889')
})

from http.server import HTTPServer, BaseHTTPRequestHandler
import json

agent_host = "https://example.com"  # The AI Agent is running in this host

class AgentRequestHandler(BaseHTTPRequestHandler):
    def do_POST(self):
        self._handle_request()

    def do_GET(self):
        self._handle_request()

    def _handle_request(self):
        auth_header = self.headers.get("Authorization")
        requested_url = f"{agent_host}{self.path}"
        http_verb = self.command
        print("Received request:", {"endpoint": requested_url, "httpVerb": http_verb, "authHeader": auth_header})

        try:
            is_valid_req = payments.requests.start_processing_request(
                agent_id,
                auth_header,
                requested_url,
                http_verb
            )
            print("isValidReq", is_valid_req)
            if is_valid_req["balance"]["isSubscriber"]:
                self.send_response(200)
                self.send_header("Content-Type", "application/json")
                self.end_headers()
                self.wfile.write(json.dumps({"message": "Hello from the Agent!"}).encode())
                return
        except Exception as error:
            print("Unauthorized access attempt:", auth_header)
            print("Error details:", error)

        self.send_response(402)
        self.send_header("Content-Type", "application/json")
        self.end_headers()
        self.wfile.write(json.dumps({"error": "Payment Required"}).encode())
        return

# To start the server:
server = HTTPServer(("localhost", 8889), AgentRequestHandler)
print("AI Agent server running on port 8889")
server.serve_forever()

Advanced Request Processing

Complete AI Agent Implementation

Here’s a more comprehensive example that includes AI processing logic:

TypeScript
Python

import express from 'express'
import { Payments } from '@nevermined-io/payments'

const app = express()
app.use(express.json())

const payments = Payments.getInstance({
  nvmApiKey: process.env.NVM_API_KEY!,
  environment: 'testing'
})

const agentId = 'your-agent-id'

// AI processing function (replace with your actual AI logic)
async function processAIRequest(query: string, parameters?: any) {
  // This is where you'd integrate with your AI model
  // For example: OpenAI, local models, or other AI services
  return {
    response: `AI processed query: ${query}`,
    timestamp: new Date().toISOString(),
    parameters
  }
}

app.post('/api/v1/agents/:agentId/tasks', async (req, res) => {
  const authHeader = req.headers['authorization']
  const requestedUrl = `${req.protocol}://${req.get('host')}${req.path}`
  const httpVerb = req.method

  try {
    // Validate the request and user subscription
    const validationResult = await payments.requests.startProcessingRequest(
      agentId,
      authHeader,
      requestedUrl,
      httpVerb
    )

    if (!validationResult.balance.isSubscriber) {
      return res.status(402).json({
        error: 'Payment Required',
        message: 'You need an active subscription to access this agent',
        plans: validationResult.plans || []
      })
    }

    // Extract the query from the request
    const { query, parameters } = req.body

    if (!query) {
      return res.status(400).json({
        error: 'Missing query parameter'
      })
    }

    // Process the AI request
    const aiResponse = await processAIRequest(query, parameters)

    // Optionally, you can redeem additional credits based on processing complexity
    // await payments.requests.redeemCredits(planId, additionalCredits, proof)

    res.json({
      success: true,
      result: aiResponse,
      creditsRemaining: validationResult.balance.balance
    })

  } catch (error) {
    console.error('Request processing error:', error)
    
    if (error.message.includes('insufficient credits')) {
      res.status(402).json({
        error: 'Insufficient Credits',
        message: 'You need more credits to access this service'
      })
    } else {
      res.status(500).json({
        error: 'Internal Server Error',
        message: 'An error occurred while processing your request'
      })
    }
  }
})

// Health check endpoint (usually free/open)
app.get('/api/v1/health', (req, res) => {
  res.json({ status: 'healthy', timestamp: new Date().toISOString() })
})

const PORT = process.env.PORT || 8889
app.listen(PORT, () => {
  console.log(`AI Agent server running on port ${PORT}`)
})

from flask import Flask, request, jsonify
from payments_py import Payments
import os
import json
from datetime import datetime

app = Flask(__name__)

# Initialize payments client
payments = Payments(
    api_key=os.environ.get('NVM_API_KEY'),
    environment='testing'
)

agent_id = 'your-agent-id'

def process_ai_request(query, parameters=None):
    """
    AI processing function (replace with your actual AI logic)
    For example: integrate with OpenAI, local models, or other AI services
    """
    return {
        'response': f'AI processed query: {query}',
        'timestamp': datetime.now().isoformat(),
        'parameters': parameters
    }

@app.route('/api/v1/agents/<agent_id>/tasks', methods=['POST'])
def handle_agent_request(agent_id):
    auth_header = request.headers.get('Authorization')
    requested_url = request.url
    http_verb = request.method

    try:
        # Validate the request and user subscription
        validation_result = payments.requests.start_processing_request(
            agent_id,
            auth_header,
            requested_url,
            http_verb
        )

        if not validation_result.get('balance', {}).get('isSubscriber'):
            return jsonify({
                'error': 'Payment Required',
                'message': 'You need an active subscription to access this agent',
                'plans': validation_result.get('plans', [])
            }), 402

        # Extract the query from the request
        data = request.get_json()
        query = data.get('query')
        parameters = data.get('parameters')

        if not query:
            return jsonify({
                'error': 'Missing query parameter'
            }), 400

        # Process the AI request
        ai_response = process_ai_request(query, parameters)

        return jsonify({
            'success': True,
            'result': ai_response,
            'creditsRemaining': validation_result.get('balance', {}).get('balance')
        })

    except Exception as error:
        print(f'Request processing error: {error}')
        
        if 'insufficient credits' in str(error):
            return jsonify({
                'error': 'Insufficient Credits',
                'message': 'You need more credits to access this service'
            }), 402
        else:
            return jsonify({
                'error': 'Internal Server Error',
                'message': 'An error occurred while processing your request'
            }), 500

@app.route('/api/v1/health', methods=['GET'])
def health_check():
    """Health check endpoint (usually free/open)"""
    return jsonify({
        'status': 'healthy', 
        'timestamp': datetime.now().isoformat()
    })

if __name__ == '__main__':
    port = int(os.environ.get('PORT', 8889))
    print(f'AI Agent server running on port {port}')
    app.run(host='0.0.0.0', port=port, debug=True)

Integration with Popular Frameworks

FastAPI Integration

Python (FastAPI)

from fastapi import FastAPI, HTTPException, Depends, Header
from pydantic import BaseModel
from payments_py import Payments
import os

app = FastAPI()
payments = Payments(
    api_key=os.environ.get('NVM_API_KEY'),
    environment='testing'
)

class QueryRequest(BaseModel):
    query: str
    parameters: dict = None

class AIResponse(BaseModel):
    success: bool
    result: dict
    creditsRemaining: int

async def validate_subscription(
    authorization: str = Header(None),
    agent_id: str = "your-agent-id"
):
    if not authorization:
        raise HTTPException(status_code=401, detail="Authorization header required")
    
    try:
        validation_result = payments.requests.start_processing_request(
            agent_id,
            authorization,
            "/api/v1/tasks",  # endpoint
            "POST"
        )
        
        if not validation_result.get('balance', {}).get('isSubscriber'):
            raise HTTPException(
                status_code=402, 
                detail="Payment Required - No active subscription"
            )
            
        return validation_result
        
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.post("/api/v1/tasks", response_model=AIResponse)
async def process_task(
    request: QueryRequest,
    validation: dict = Depends(validate_subscription)
):
    # Process AI request here
    ai_result = {
        "response": f"Processed: {request.query}",
        "timestamp": "2023-10-27T10:00:00Z"
    }
    
    return AIResponse(
        success=True,
        result=ai_result,
        creditsRemaining=validation.get('balance', {}).get('balance', 0)
    )

Middleware Pattern

For cleaner code organization, you can create middleware to handle payment validation:

TypeScript (Express Middleware)

import { Request, Response, NextFunction } from 'express'

interface AuthenticatedRequest extends Request {
  validationResult?: any
  userBalance?: number
}

const paymentsMiddleware = (agentId: string) => {
  return async (req: AuthenticatedRequest, res: Response, next: NextFunction) => {
    const authHeader = req.headers['authorization']
    const requestedUrl = `${req.protocol}://${req.get('host')}${req.path}`
    const httpVerb = req.method

    try {
      const validationResult = await payments.requests.startProcessingRequest(
        agentId,
        authHeader,
        requestedUrl,
        httpVerb
      )

      if (!validationResult.balance.isSubscriber) {
        return res.status(402).json({
          error: 'Payment Required',
          plans: validationResult.plans || []
        })
      }

      // Attach validation result to request for use in route handlers
      req.validationResult = validationResult
      req.userBalance = validationResult.balance.balance

      next()
    } catch (error) {
      console.error('Payment validation error:', error)
      res.status(402).json({
        error: 'Payment Required',
        message: error.message
      })
    }
  }
}

// Usage in routes
app.post('/api/v1/tasks', paymentsMiddleware(agentId), async (req: AuthenticatedRequest, res) => {
  // Request is already validated, process AI logic
  const aiResponse = await processAIRequest(req.body.query)
  
  res.json({
    result: aiResponse,
    creditsRemaining: req.userBalance
  })
})

Best Practices for Request Processing

Validation First

Always validate payments before processing expensive AI operations to avoid wasting resources.

Error Handling

Provide clear error messages and appropriate HTTP status codes for different failure scenarios.

Resource Management

Implement timeouts and resource limits to prevent abuse and ensure fair usage.

Monitoring

Log all requests, validation results, and processing times for analytics and debugging.

Response Patterns

Success Response

{
  "success": true,
  "result": {
    "response": "AI generated response",
    "metadata": {
      "model": "gpt-4",
      "tokens_used": 150,
      "processing_time": "0.8s"
    }
  },
  "creditsRemaining": 85,
  "timestamp": "2023-10-27T10:00:00Z"
}

Payment Required Response

{
  "error": "Payment Required",
  "message": "You need an active subscription to access this agent",
  "plans": [
    {
      "planId": "plan-123",
      "name": "Basic Plan",
      "price": "10 USDC",
      "credits": 100
    }
  ],
  "timestamp": "2023-10-27T10:00:00Z"
}

Error Response

{
  "error": "Insufficient Credits",
  "message": "You have 0 credits remaining. Please purchase more credits to continue.",
  "currentBalance": 0,
  "requiredCredits": 1,
  "timestamp": "2023-10-27T10:00:00Z"
}

Testing Your Implementation

Create test scripts to verify your payment validation:

TypeScript

// Test script for your AI agent
async function testAgent() {
  const testCases = [
    {
      name: 'Valid subscriber',
      token: 'valid-access-token',
      expectedStatus: 200
    },
    {
      name: 'Invalid token',
      token: 'invalid-token',
      expectedStatus: 402
    },
    {
      name: 'No token',
      token: null,
      expectedStatus: 401
    }
  ]

  for (const testCase of testCases) {
    console.log(`Testing: ${testCase.name}`)
    
    const response = await fetch('http://localhost:8889/api/v1/tasks', {
      method: 'POST',
      headers: {
        'Content-Type': 'application/json',
        ...(testCase.token && { 'Authorization': `Bearer ${testCase.token}` })
      },
      body: JSON.stringify({ query: 'Test query' })
    })

    console.log(`Expected: ${testCase.expectedStatus}, Got: ${response.status}`)
    console.log('Response:', await response.text())
    console.log('---')
  }
}

testAgent()

Next Steps

Now that you can process paid requests, you can:

Deploy Your Agent

Learn how to deploy your AI agent to production

Monitor Performance

Set up monitoring and analytics for your AI service

Scale Your Service

Learn best practices for scaling AI agents

Build Advanced Features

Explore advanced integration patterns and examples

Getting Started

Integration Guide

AI Services

Resources

NVM Pay

NVM ID

Authorizing only valid requests to my AI Agents

Advanced Request Processing

Complete AI Agent Implementation

Integration with Popular Frameworks

FastAPI Integration

Middleware Pattern

Best Practices for Request Processing

Validation First

Error Handling

Resource Management

Monitoring

Response Patterns

Success Response

Payment Required Response

Error Response

Testing Your Implementation

Next Steps

Deploy Your Agent

Monitor Performance

Scale Your Service

Build Advanced Features

Getting Started

Integration Guide

AI Services

Resources

NVM Pay

NVM ID

​Authorizing only valid requests to my AI Agents

​Advanced Request Processing

​Complete AI Agent Implementation

​Integration with Popular Frameworks

​FastAPI Integration

​Middleware Pattern

​Best Practices for Request Processing

Validation First

Error Handling

Resource Management

Monitoring

​Response Patterns

​Success Response

​Payment Required Response

​Error Response

​Testing Your Implementation

​Next Steps

Deploy Your Agent

Monitor Performance

Scale Your Service

Build Advanced Features

Authorizing only valid requests to my AI Agents

Advanced Request Processing

Complete AI Agent Implementation

Integration with Popular Frameworks

FastAPI Integration

Middleware Pattern

Best Practices for Request Processing

Response Patterns

Success Response

Payment Required Response

Error Response

Testing Your Implementation

Next Steps