Start here: need to register a service and create a plan first? Follow the
5-minute setup .
Add payment protection to agents running on Amazon Bedrock AgentCore using the x402 protocol . The @requires_payment decorator handles verification and settlement automatically, wrapping verify - work - settle in a single Lambda invocation.
x402 Payment Flow
AgentCore communicates using MCP JSON-RPC 2.0 . The decorator expects the AgentCore Gateway interceptor event format:
{
"mcp" : {
"gatewayRequest" : {
"httpMethod" : "POST" ,
"headers" : {
"payment-signature" : "<x402-access-token>" ,
"Content-Type" : "application/json"
},
"body" : {
"jsonrpc" : "2.0" ,
"id" : "req-1" ,
"method" : "tools/call" ,
"params" : {
"name" : "TargetName___toolName" ,
"arguments" : { "patient_id" : "123" }
}
}
}
}
}
Key points:
The x402 token travels in the payment-signature HTTP header
Tool names follow AgentCore convention: TargetName___toolName
Only tools/call method is billable by default
Installation
See the Python SDK installation guide for full setup details.
No extra dependencies required. The AgentCore decorator uses only the core payments-py package.
Quick Start: Protecting a Lambda Handler
The @requires_payment decorator wraps a Lambda handler with x402 payment verification and settlement.
import json
import os
from payments_py import Payments, PaymentOptions
from payments_py.x402.agentcore import requires_payment
# Initialize Payments
payments = Payments.get_instance(
PaymentOptions(
nvm_api_key = os.environ[ "NVM_API_KEY" ],
environment = os.environ.get( "NVM_ENVIRONMENT" , "sandbox" ),
)
)
PLAN_ID = os.environ[ "NVM_PLAN_ID" ]
AGENT_ID = os.environ.get( "NVM_AGENT_ID" )
@requires_payment (
payments = payments,
plan_id = PLAN_ID ,
agent_id = AGENT_ID ,
credits = 1 ,
endpoint = "https://gateway.bedrock-agentcore.amazonaws.com/mcp" ,
)
def lambda_handler ( event , context = None ):
"""Handler only runs when payment is verified."""
body = event[ "mcp" ][ "gatewayRequest" ][ "body" ]
tool = body[ "params" ][ "name" ]
args = body[ "params" ][ "arguments" ]
# Agent work
result = process_request(tool, args)
return {
"content" : [{ "type" : "text" , "text" : json.dumps(result)}],
}
The decorator automatically:
Returns a 402 PaymentRequired response when no token is provided
Verifies the x402 token via the Nevermined facilitator
Executes the handler on successful verification
Burns credits after successful execution
Adds payment-response header and _meta.x402 transaction data to the response
The handler can return any of these formats:
Bare MCP result (simplest)
return {
"content" : [{ "type" : "text" , "text" : "Hello!" }],
}
With credits override
return {
"content" : [{ "type" : "text" , "text" : "Expensive result" }],
"_meta" : { "creditsToCharge" : 5 }, # Override the decorator's credits
}
Full MCP response body
return {
"jsonrpc" : "2.0" ,
"id" : "req-1" ,
"result" : {
"content" : [{ "type" : "text" , "text" : "Hello!" }],
},
}
Complete InterceptorOutput
return {
"interceptorOutputVersion" : "1.0" ,
"mcp" : {
"transformedGatewayResponse" : {
"statusCode" : 200 ,
"headers" : { "Content-Type" : "application/json" },
"body" : { "jsonrpc" : "2.0" , "id" : "req-1" , "result" : { ... }},
}
},
}
The decorator wraps and enriches all formats with payment-response headers and _meta.x402 transaction data.
Credits Flow
The credits parameter is used for both verify (pre-flight check) and settle (default burn amount). The handler can override the settle amount by returning _meta.creditsToCharge in its response.
Decorator Configuration
Basic
@requires_payment ( payments = payments, plan_id = "plan-123" , credits = 1 )
def handler ( event , context = None ):
...
With Agent ID
@requires_payment (
payments = payments,
plan_id = PLAN_ID ,
credits = 1 ,
agent_id = os.environ.get( "NVM_AGENT_ID" ),
endpoint = "https://my-gateway.amazonaws.com/mcp" ,
)
def handler ( event , context = None ):
...
Dynamic Credits
Calculate credits based on the event:
def calc_credits ( event ):
"""Charge based on request complexity."""
args = (
event.get( "mcp" , {})
.get( "gatewayRequest" , {})
.get( "body" , {})
.get( "params" , {})
.get( "arguments" , {})
)
return 10 if args.get( "detailed" ) else 1
@requires_payment ( payments = payments, plan_id = PLAN_ID , credits = calc_credits)
def handler ( event , context = None ):
...
@requires_payment (
payments = payments,
plan_id = PLAN_ID ,
credits = 1 ,
token_header = "X-Payment" , # Or a list: ["X-Payment", "Authorization"]
)
def handler ( event , context = None ):
...
Lifecycle Hooks
def on_before_verify ( payment_required ):
print ( f "Verifying payment for plan { payment_required.accepts[ 0 ].plan_id } " )
def on_after_verify ( verification ):
print ( f "Verified! Payer: { verification.payer } " )
def on_after_settle ( credits_used , settlement ):
print ( f "Settled { credits_used } credits, tx: { settlement.transaction } " )
def on_payment_error ( error ):
# Return custom error response or None for default 402
return None
@requires_payment (
payments = payments,
plan_id = PLAN_ID ,
credits = 1 ,
on_before_verify = on_before_verify,
on_after_verify = on_after_verify,
on_after_settle = on_after_settle,
on_payment_error = on_payment_error,
)
def handler ( event , context = None ):
...
Client Implementation
import os
from payments_py import Payments, PaymentOptions
# Initialize Payments as subscriber
payments = Payments.get_instance(
PaymentOptions(
nvm_api_key = os.environ[ "NVM_SUBSCRIBER_API_KEY" ],
environment = os.environ.get( "NVM_ENVIRONMENT" , "sandbox" ),
)
)
PLAN_ID = os.environ[ "NVM_PLAN_ID" ]
AGENT_ID = os.environ.get( "NVM_AGENT_ID" )
# Step 1: Check balance
balance = payments.plans.get_plan_balance( plan_id = PLAN_ID )
print ( f "Balance: { balance.balance } credits" )
# Step 2: Get x402 access token
token_result = payments.x402.get_x402_access_token(
plan_id = PLAN_ID ,
agent_id = AGENT_ID ,
)
access_token = token_result[ "accessToken" ]
# Step 3: Call the agent with the token in payment-signature header
import requests
response = requests.post(
"https://your-agent-endpoint.amazonaws.com/mcp" ,
headers = {
"Content-Type" : "application/json" ,
"payment-signature" : access_token,
},
json = {
"jsonrpc" : "2.0" ,
"id" : "1" ,
"method" : "tools/call" ,
"params" : {
"name" : "getPatient" ,
"arguments" : { "patient_id" : "123" },
},
},
)
# Step 4: Check settlement receipt
import base64, json
payment_response = response.headers.get( "payment-response" )
if payment_response:
receipt = json.loads(base64.b64decode(payment_response))
print ( f "Credits redeemed: { receipt[ 'creditsRedeemed' ] } " )
print ( f "Transaction: { receipt[ 'transactionHash' ] } " )
Decorator vs Interceptor
payments-py provides two patterns for AgentCore payment protection:
@requires_payment decoratorAgentCoreInterceptor classInvocations Single (verify + work + settle) Two (REQUEST phase, RESPONSE phase) Use case Agent IS the Lambda Separate interceptor Lambda Credits from response _meta.creditsToCharge override_meta.creditsToCharge extractionComplexity One decorator, one function Separate interceptor + target Lambda Gateway requirement Optional Required (Gateway calls interceptor)
Use the decorator when your agent runs as a Lambda function and you want the simplest integration. Use the interceptor when you need a separate payment layer in front of an existing agent target (MCP server, OpenAPI, Lambda) via the AgentCore Gateway.
Environment Variables
# Nevermined (required)
NVM_API_KEY = sandbox:eyxxxx # sandbox:eyxxxx OR live:eyxxxx
NVM_ENVIRONMENT = sandbox # sandbox OR live
NVM_PLAN_ID = your-plan-id
NVM_AGENT_ID = your-agent-id # Optional
Next Steps
Payment Patterns Advanced credit charging patterns
x402 Protocol Deep dive into x402 payment flows
Strands Integration Decorator for Strands AI agent tools
Payment Models Configure credits, subscriptions, and dynamic pricing