Query Agents - Nevermined Documentation

Once a Payment Plan is purchased, user(s) can query all the AI Agents linked to that plan. For identifying the user as a valid subscriber, they need to send HTTP requests to the AI Agent and include a valid Access Token. This is sent using the payment-signature HTTP header (per the x402 protocol).

Nevermined Proxy instances are standard HTTP Proxies in charge of authorizing users trying to access AI Agents or Services.

Once a user is a subscriber, sending a request is quite simple.

Get the AI Agent or Service Access Token

The access token is minted against a delegation — a one-time authorization that lets the facilitator spend on the subscriber’s behalf, up to a limit. Create the delegation once with createDelegation / create_delegation, then reference it by delegationId whenever you mint a token.

TypeScript
Python

// Create a delegation once — it authorizes spend up to a limit, reusable across tokens.
const { delegationId } = await payments.delegation.createDelegation({
  provider: 'erc4337',       // 'stripe' | 'braintree' | 'visa' for fiat plans
  spendingLimitCents: 10000, // $100 budget
  durationSecs: 604800,      // 7 days
  currency: 'usdc',          // 'usd' for fiat plans
})

const { accessToken } = await payments.x402.getX402AccessToken(planId, agentId, {
  scheme: 'nvm:erc4337',       // 'nvm:card-delegation' for fiat plans
  delegationConfig: { delegationId },
})
// OUTPUT: accessToken is a JWT string containing X402 v2 payment credentials

from payments_py.x402 import X402TokenOptions, DelegationConfig, CreateDelegationPayload

# Create a delegation once — it authorizes spend up to a limit, reusable across tokens.
delegation = payments.delegation.create_delegation(
    CreateDelegationPayload(
        provider="erc4337",          # 'stripe' | 'braintree' | 'visa' for fiat plans
        spending_limit_cents=10000,  # $100 budget
        duration_secs=604800,        # 7 days
        currency="usdc",             # 'usd' for fiat plans
    )
)

token_response = payments.x402.get_x402_access_token(
    plan_id,
    agent_id,
    token_options=X402TokenOptions(
        scheme="nvm:erc4337",  # "nvm:card-delegation" for fiat plans
        delegation_config=DelegationConfig(delegation_id=delegation.delegation_id)
    ),
)
access_token = token_response["accessToken"]
# OUTPUT: access_token is a JWT string containing X402 v2 payment credentials

Is this a fiat plan or a crypto plan? It determines the provider/currency above and whether you must pass a scheme. The direct getX402AccessToken call defaults to crypto (nvm:erc4337) — for card plans pass scheme: 'nvm:card-delegation'. See Order Plans for how to detect a plan’s type before you pay.

Sending a query to the AI Agent

Once you have a valid X402 access token, you can query the AI Agent by making a standard HTTP request with the token in the payment-signature header. This request must be sent directly to the Agent endpoint (the Agent API description is in the Agent Metadata).

Because Nevermined authorizes standard HTTP requests, they can be used to protect any kind of AI Agent or Service exposing an HTTP API.

Using cURL

export AGENT_ACCESS_TOKEN="eJyNj0sKgDAURP9lJQ..."

curl -k -X POST \
  -H "Content-Type: application/json" \
  -H "payment-signature: $AGENT_ACCESS_TOKEN" \
  -d '{"query": "hey there"}' \
  https://my.agent.io/prompt

Using Programming Languages

TypeScript
Python

const agentHTTPOptions = {
  method: 'POST',
  headers: {
    Accept: 'application/json',
    'Content-Type': 'application/json',
    'payment-signature': accessToken,
  },
  body: JSON.stringify({
    query: "What's the weather like today?",
    parameters: {
      location: "New York"
    }
  })
}

const response = await fetch(new URL('https://my.agent.io/prompt'), agentHTTPOptions)
const result = await response.json()
console.log('Agent response:', result)

import requests

# HTTP options for the agent request
headers = {
    "Accept": "application/json",
    "Content-Type": "application/json",
    "payment-signature": access_token,
}

payload = {
    "query": "What's the weather like today?",
    "parameters": {
        "location": "New York"
    }
}

response = requests.post(
    "https://my.agent.io/prompt",
    headers=headers,
    json=payload
)

if response.status_code == 200:
    result = response.json()
    print('Agent response:', result)
else:
    print(f'Error: {response.status_code} - {response.text}')

Complete Query Flow Example

Here’s a complete example showing the full flow from getting access tokens to querying an agent:

TypeScript
Python

async function queryAIAgent(planId: string, agentId: string, delegationId: string, query: string) {
  try {
    // Step 1: Mint an X402 access token against a delegation you created once
    const { accessToken } = await payments.x402.getX402AccessToken(planId, agentId, {
      scheme: 'nvm:erc4337',       // 'nvm:card-delegation' for fiat plans
      delegationConfig: { delegationId },
    })

    if (!accessToken) {
      throw new Error('No access token received - check if plan is valid and has balance')
    }

    // Step 2: Get agent details
    const agent = await payments.agents.getAgent(agentId)
    const endpoint = agent.endpoints[0] // Use first endpoint

    // Extract URL from endpoint object (endpoints have HTTP method as key)
    const method = Object.keys(endpoint)[0] // e.g., 'POST'
    const url = endpoint[method] // The actual URL

    // Step 3: Make the request
    const response = await fetch(url, {
      method: method,
      headers: {
        'Content-Type': 'application/json',
        'payment-signature': accessToken
      },
      body: JSON.stringify({ query })
    })

    if (response.status === 402) {
      throw new Error('Payment required - insufficient credits or plan expired')
    }

    if (!response.ok) {
      throw new Error(`Agent request failed: ${response.status} ${response.statusText}`)
    }

    const result = await response.json()

    // Step 4: Check remaining balance (optional)
    const balance = await payments.plans.getPlanBalance(planId)
    console.log(`Remaining credits: ${balance.balance}`)

    return result

  } catch (error) {
    console.error('Query failed:', error.message)
    throw error
  }
}

// Usage — create the delegation once, then reuse its id across queries
const { delegationId } = await payments.delegation.createDelegation({
  provider: 'erc4337', spendingLimitCents: 10000, durationSecs: 604800, currency: 'usdc',
})
const result = await queryAIAgent(planId, agentId, delegationId, "Explain quantum computing")
console.log('AI Response:', result)

import requests
import json
from payments_py.x402 import X402TokenOptions, DelegationConfig, CreateDelegationPayload

def query_ai_agent(plan_id, agent_id, delegation_id, query):
    try:
        # Step 1: Mint an X402 access token against a delegation you created once
        token_response = payments.x402.get_x402_access_token(
            plan_id,
            agent_id,
            token_options=X402TokenOptions(
                scheme="nvm:erc4337",  # "nvm:card-delegation" for fiat plans
                delegation_config=DelegationConfig(delegation_id=delegation_id)
            ),
        )
        access_token = token_response.get('accessToken')

        if not access_token:
            raise Exception('No access token received - check if plan is valid and has balance')

        # Step 2: Get agent details
        agent = payments.agents.get_agent(agent_id)
        endpoint = agent['endpoints'][0]  # Use first endpoint

        # Extract URL from endpoint object (endpoints have HTTP method as key)
        method = list(endpoint.keys())[0]  # e.g., 'POST'
        url = endpoint[method]  # The actual URL

        # Step 3: Make the request
        headers = {
            'Content-Type': 'application/json',
            'payment-signature': access_token
        }

        payload = {'query': query}

        response = requests.post(
            url,
            headers=headers,
            json=payload
        )

        if response.status_code == 402:
            raise Exception('Payment required - insufficient credits or plan expired')

        response.raise_for_status()
        result = response.json()

        # Step 4: Check remaining balance (optional)
        balance = payments.plans.get_plan_balance(plan_id)
        print(f"Remaining credits: {balance.get('balance')}")

        return result

    except Exception as error:
        print(f'Query failed: {error}')
        raise error

# Usage — create the delegation once, then reuse its id across queries
delegation = payments.delegation.create_delegation(
    CreateDelegationPayload(
        provider="erc4337", spending_limit_cents=10000, duration_secs=604800, currency="usdc",
    )
)
result = query_ai_agent(plan_id, agent_id, delegation.delegation_id, "Explain quantum computing")
print('AI Response:', result)

Handling Different Response Types

AI agents can return various types of responses. Here’s how to handle them:

TypeScript
Python

async function handleAgentResponse(response: Response) {
  const contentType = response.headers.get('content-type')
  
  if (contentType?.includes('application/json')) {
    // JSON response (most common)
    return await response.json()
  } else if (contentType?.includes('text/')) {
    // Text response
    return await response.text()
  } else if (contentType?.includes('image/')) {
    // Image response
    const blob = await response.blob()
    return URL.createObjectURL(blob)
  } else {
    // Binary or other response
    return await response.arrayBuffer()
  }
}

def handle_agent_response(response):
    content_type = response.headers.get('content-type', '')
    
    if 'application/json' in content_type:
        # JSON response (most common)
        return response.json()
    elif 'text/' in content_type:
        # Text response
        return response.text
    elif 'image/' in content_type:
        # Image response
        return response.content
    else:
        # Binary or other response
        return response.content

Error Handling and Status Codes

200 - Success

Request was successful and credits were deducted from the user’s balance

401 - Unauthorized

Invalid or missing access token. User needs to get a new token

402 - Payment Required

User doesn’t have sufficient credits or their plan has expired

403 - Forbidden

User has access but the specific endpoint is not allowed for their plan

429 - Rate Limited

Too many requests. User should wait before trying again

500 - Server Error

Agent internal error. This doesn’t consume credits

Best Practices

Token Management

Cache access tokens (they’re valid for a limited time)
Refresh tokens when they expire
Never expose tokens in client-side logs

Error Handling

Always check response status codes
Implement retry logic for network failures
Handle 402 responses by prompting for plan renewal

Performance

Reuse access tokens across multiple requests
Implement request timeouts
Use appropriate HTTP methods (GET, POST)

Monitoring

Track credit usage patterns
Monitor response times and error rates
Set up alerts for low credit balances

Advanced Usage Patterns

Batch Requests

For efficiency, you can send multiple queries in a single request:

TypeScript
Python

const batchQuery = {
  requests: [
    { id: '1', query: 'What is AI?' },
    { id: '2', query: 'Explain machine learning' },
    { id: '3', query: 'Define neural networks' }
  ]
}

const response = await fetch(agentEndpoint, {
  method: 'POST',
  headers: {
    'Content-Type': 'application/json',
    'payment-signature': accessToken
  },
  body: JSON.stringify(batchQuery)
})

batch_query = {
    'requests': [
        {'id': '1', 'query': 'What is AI?'},
        {'id': '2', 'query': 'Explain machine learning'},
        {'id': '3', 'query': 'Define neural networks'}
    ]
}

response = requests.post(
    agent_endpoint,
    headers={'payment-signature': access_token},
    json=batch_query
)

Streaming Responses

For long-running AI operations, you might want to handle streaming responses:

const response = await fetch(agentEndpoint, {
  method: 'POST',
  headers: {
    'Content-Type': 'application/json',
    'payment-signature': accessToken,
    'Accept': 'text/event-stream'
  },
  body: JSON.stringify({ query, stream: true })
})

const reader = response.body?.getReader()
if (reader) {
  while (true) {
    const { done, value } = await reader.read()
    if (done) break
    
    const chunk = new TextDecoder().decode(value)
    console.log('Streaming chunk:', chunk)
  }
}

Next Steps

Now that you know how to query AI agents, learn how to:

Build AI Agents

Learn how to create AI agents that accept paid requests

Handle Requests

Learn how to process requests and manage responses

​Get the AI Agent or Service Access Token

​Sending a query to the AI Agent

​Using cURL

​Using Programming Languages

​Complete Query Flow Example

​Handling Different Response Types

​Error Handling and Status Codes

​Best Practices

Token Management

Error Handling

Performance

Monitoring

​Advanced Usage Patterns

​Batch Requests

​Streaming Responses

​Next Steps

Build AI Agents

Handle Requests

Get the AI Agent or Service Access Token

Sending a query to the AI Agent

Using cURL

Using Programming Languages

Complete Query Flow Example

Handling Different Response Types

Error Handling and Status Codes

Best Practices

Advanced Usage Patterns

Batch Requests

Streaming Responses

Next Steps