AWS Bedrock AgentCore
+ x402-AgentPay

x402-AgentPay sits as a commerce middleware layer between your AWS Bedrock AgentCore agents and on-chain payment rails. It handles escrow, spend enforcement, reputation tracking, and USDC settlement so your agents don't have to manage wallets directly.

Install the SDK

# Install x402-AgentPay Python SDK
pip install x402-agentpay

# For Node.js
npm install @x402/agentpay

Set Environment Variables

# .env
AGENTPAY_API_KEY=your_api_key_here
AGENTPAY_ENDPOINT_ID=your_endpoint_id
AGENTPAY_WALLET_ADDRESS=0xYourBaseL2WalletAddress

# AWS credentials (if not using IAM role)
AWS_ACCESS_KEY_ID=your_aws_key
AWS_SECRET_ACCESS_KEY=your_aws_secret
AWS_DEFAULT_REGION=us-east-1

Initialize the Client

import boto3
from x402_agentpay import AgentPayClient, SpendPolicy

# Initialize x402-AgentPay
agentpay = AgentPayClient(
    api_key="your_api_key",
    endpoint_id="your_endpoint_id",
    wallet_address="0xYourBaseL2Wallet"
)

# Initialize AWS Bedrock AgentCore client
bedrock_agent = boto3.client(
    "bedrock-agent-runtime",
    region_name="us-east-1"
)

Register your AWS Bedrock AgentCore agent as a monetizable endpoint in x402-AgentPay. This enables other agents and users to pay to invoke your agent's capabilities.

import requests

# Register your AgentCore endpoint
response = requests.post(
    "https://x402-agent-pay.com/api/register",
    headers={"Authorization": f"Bearer {AGENTPAY_API_KEY}"},
    json={
        "endpoint_type": "aws_agentcore",
        "name": "My Bedrock Agent",
        "description": "Code analysis and generation agent",
        "aws_agent_id": "YOUR_BEDROCK_AGENT_ID",
        "aws_agent_alias_id": "TSTALIASID",
        "pricing": {
            "per_invocation_usdc": 0.01,   # $0.01 per call
            "per_token_usdc": 0.000005      # optional token-based pricing
        },
        "wallet_address": "0xYourBaseL2Wallet",
        "capabilities": ["code_analysis", "code_generation", "debugging"]
    }
)

endpoint = response.json()
print(f"Endpoint registered: {endpoint['endpoint_id']}")

Spend policies let you set hard limits on how much your agent can spend per session, per day, or per counterparty — preventing runaway costs in autonomous agent loops.

from x402_agentpay import SpendPolicy

# Create a spend policy for your agent session
policy = agentpay.create_policy(SpendPolicy(
    name="bedrock-session-policy",
    daily_limit_usdc=5.00,         # Max $5/day
    per_tx_limit_usdc=0.50,         # Max $0.50 per transaction
    session_limit_usdc=2.00,        # Max $2 per agent session
    allowed_endpoints=[              # Whitelist specific endpoints
        "endpoint_abc123",
        "endpoint_xyz789"
    ],
    require_escrow=True,             # Force escrow for all payments
    auto_approve_under_usdc=0.05    # Auto-approve micro-payments under $0.05
))

print(f"Policy created: {policy.policy_id}")
print(f"Daily remaining: ${policy.daily_remaining}")

Attach Policy to AgentCore Invocation

# Invoke AgentCore with spend policy attached
def invoke_with_spend_policy(prompt: str, policy_id: str):
    # Check policy allows this invocation
    check = agentpay.check_policy(policy_id, estimated_cost_usdc=0.01)
    if not check.allowed:
        raise Exception(f"Spend policy blocked: {check.reason}")

    # Invoke Bedrock AgentCore
    response = bedrock_agent.invoke_agent(
        agentId="YOUR_AGENT_ID",
        agentAliasId="TSTALIASID",
        sessionId="session-123",
        inputText=prompt
    )

    # Record the actual cost
    agentpay.record_usage(
        policy_id=policy_id,
        cost_usdc=0.01,
        metadata={"agent_id": "YOUR_AGENT_ID", "prompt_tokens": 150}
    )

    return response

Your AgentCore agent can pay other x402-registered endpoints directly — enabling true agent-to-agent commerce without human intervention.

import asyncio
from x402_agentpay import Payment

async def agent_pay_for_service(service_endpoint_id: str, amount_usdc: float):
    # Create payment request
    payment = await agentpay.create_payment(Payment(
        from_endpoint=AGENTPAY_ENDPOINT_ID,
        to_endpoint=service_endpoint_id,
        amount_usdc=amount_usdc,
        memo="Payment for code analysis service",
        x402_header=True    # Include x402 payment header for HTTP services
    ))

    # payment.x402_header is ready to attach to your HTTP request
    return payment

# Example: Agent pays for a data enrichment API
async def enriched_agent_call(query: str):
    payment = await agent_pay_for_service(
        service_endpoint_id="endpoint_data_enrichment",
        amount_usdc=0.005
    )

    response = requests.post(
        "https://data-service.example.com/enrich",
        headers={
            "X-Payment": payment.x402_header,
            "Content-Type": "application/json"
        },
        json={"query": query}
    )
    return response.json()

Use escrow when your agent delegates work to sub-agents or external services. Funds are locked until task completion is verified, then released on-chain.

from x402_agentpay import EscrowSession

# Create an escrow session for a multi-agent task
escrow = agentpay.create_escrow(EscrowSession(
    payer_endpoint=AGENTPAY_ENDPOINT_ID,
    payee_endpoint="endpoint_subagent_xyz",
    amount_usdc=1.00,
    task_description="Analyze 50 smart contracts for vulnerabilities",
    completion_criteria={
        "type": "webhook",
        "url": "https://your-server.com/escrow-complete",
        "timeout_hours": 24
    }
))

print(f"Escrow created: {escrow.escrow_id}")
print(f"Status: {escrow.status}")  # 'locked'

# Later: release funds after verification
release = agentpay.release_escrow(
    escrow_id=escrow.escrow_id,
    verify_result=True  # Triggers on-chain USDC transfer to payee
)
print(f"Released: ${release.amount_usdc} USDC to {release.payee_wallet}")

For high-frequency agent interactions, use batch settlement to group many micro-payments into a single on-chain transaction. This reduces gas costs by up to 95%.

from x402_agentpay import BatchSession

# Open a batch settlement session
batch = agentpay.open_batch_session(BatchSession(
    endpoint_id=AGENTPAY_ENDPOINT_ID,
    max_batch_size=100,         # Settle after 100 transactions
    max_wait_seconds=300,       # Or every 5 minutes, whichever comes first
    currency="USDC",
    chain="base"
))

# Add micro-payments to the batch (no gas cost per payment)
for i in range(50):
    batch.add(
        to_endpoint=f"endpoint_{i}",
        amount_usdc=0.001,
        memo=f"Micro-task {i} payment"
    )

# Settle the batch on-chain (single transaction, low gas)
settlement = agentpay.settle_batch(batch.batch_id)
print(f"Settled {settlement.tx_count} payments in 1 tx")
print(f"Gas saved: ~{settlement.gas_saved_pct}%")
print(f"Tx hash: {settlement.tx_hash}")

Monitor all your agent spend, escrow positions, and on-chain settlements from the AgentPay Billing Dashboard.

All endpoints are available at https://x402-agent-pay.com/api/v1. Authenticate with your AGENTPAY_API_KEY in the Authorization: Bearer header.

A complete end-to-end example: a user invokes your AgentCore agent (paying $0.01 USDC), your agent sub-delegates to a specialized data agent (paying $0.005 USDC from escrow), and returns the result.

import boto3
import requests
from x402_agentpay import AgentPayClient, SpendPolicy, EscrowSession

# ── Setup ──────────────────────────────────────────────────────
agentpay = AgentPayClient(api_key="YOUR_KEY", endpoint_id="YOUR_ENDPOINT")
bedrock = boto3.client("bedrock-agent-runtime", region_name="us-east-1")

# ── Step 1: Verify incoming payment (user → your agent) ────────
def handle_paid_request(request_headers: dict, user_prompt: str):
    payment_header = request_headers.get("X-Payment")
    verified = agentpay.verify_payment(payment_header, expected_usdc=0.01)

    if not verified.valid:
        return {"error": "Payment required", "x402": agentpay.payment_required()}

    # ── Step 2: Create spend policy for sub-delegation ─────────────
    policy = agentpay.create_policy(SpendPolicy(
        name="session-policy",
        session_limit_usdc=0.05,
        per_tx_limit_usdc=0.01
    ))

    # ── Step 3: Invoke Bedrock AgentCore ──────────────────────────
    bedrock_response = bedrock.invoke_agent(
        agentId="YOUR_BEDROCK_AGENT_ID",
        agentAliasId="TSTALIASID",
        sessionId=verified.session_id,
        inputText=user_prompt
    )

    result = ""
    for event in bedrock_response["completion"]:
        if "chunk" in event:
            result += event["chunk"]["bytes"].decode()

    # ── Step 4: Record usage & respond ────────────────────────────
    agentpay.record_usage(policy_id=policy.policy_id, cost_usdc=0.01)

    return {
        "result": result,
        "payment_confirmed": True,
        "tx_hash": verified.tx_hash
    }