Agentcore Payments Wiring

import os
import json
import base64
import httpx
import boto3

# Payment configuration from environment
PAYMENT_MANAGER_ARN = os.getenv("PAYMENT_MANAGER_ARN")
PAYMENT_INSTRUMENT_ID = os.getenv("PAYMENT_INSTRUMENT_ID")
PAYMENT_SESSION_ID = os.getenv("PAYMENT_SESSION_ID")
PAYMENT_USER_ID = os.environ.get("PAYMENT_USER_ID")  # Required — no insecure default
REGION = os.getenv(

from strands import Agent, tool

@tool
def x402_fetch(url: str, method: str = "GET") -> str:
    """Fetch a URL with automatic x402 payment handling.

    If the endpoint returns 402 Payment Required with an x402 challenge,
    this tool automatically processes the payment and retries with proof.

    Args:
        url: The URL to fetch
        method: HTTP method (GET, POST, etc.)
    """
    return _x402_fetch_impl(url, method)

agent = Agent(
    model="<model_id>",
    tools=[x402_fetch],
    system_prompt=(
        "You are a helpful assistant that can access paid APIs and content. "
        "Use the x402_fetch tool to access URLs that may require payment — "
        "it handles x402 payments automatically."
    ),
)

from langchain_core.tools import tool
from langgraph.prebuilt import create_react_agent
from langchain_aws import ChatBedrock

@tool
def x402_fetch(url: str, method: str = "GET") -> str:
    """Fetch a URL with automatic x402 payment handling.

    If the endpoint returns 402 Payment Required with an x402 challenge,
    this tool automatically processes the payment and retries with proof.

    Args:
        url: The URL to fetch
        method: HTTP method (GET, POST, etc.)
    """
    return _x402_fetch_impl(url, method)

model = ChatBedrock(model_id="<model_id>", region_name=REGION)
graph = create_react_agent(model, tools=[x402_fetch])

# Invoke:
result = graph.invoke({"messages": [("human", "Fetch https://paid-api.example.com/data")]})
print(result["messages"][-1].content)

from agents import Agent, Runner, function_tool

@function_tool
def x402_fetch(url: str, method: str = "GET") -> str:
    """Fetch a URL with automatic x402 payment handling.

    If the endpoint returns 402 Payment Required with an x402 challenge,
    this tool automatically processes the payment and retries with proof.

    Args:
        url: The URL to fetch
        method: HTTP method (GET, POST, etc.)
    """
    return _x402_fetch_impl(url, method)

agent = Agent(
    name="PaymentAgent",
    instructions=(
        "You are a helpful assistant that can access paid APIs and content. "
        "Use the x402_fetch tool to access URLs that may require payment — "
        "it handles x402 payments automatically."
    ),
    tools=[x402_fetch],
)

# Invoke:
import asyncio
result = asyncio.run(Runner.run(agent, "Fetch https://paid-api.example.com/data"))
print(result.final_output)