1
0
mirror of synced 2026-05-22 14:43:35 +00:00

Add pay-for-api-agent use case (#1504)

This commit is contained in:
Chris Wajule
2026-05-21 11:16:06 -04:00
committed by GitHub
parent 768705c208
commit 5a5451555c
41 changed files with 6334 additions and 0 deletions
@@ -0,0 +1,7 @@
# Privy Wallet Hub frontend.
# The notebook clones https://github.com/privy-io/aws-agentcore-sdk
# into `privy-delegation/` on first run (see §4.5e). The folder is a
# vendored upstream tree, not part of this sample, so we never commit
# it back. This rule keeps it out of `git status` for contributors who
# run the notebook locally.
privy-delegation/
@@ -0,0 +1,395 @@
# Pay-For-API
## Overview
**Amazon Bedrock AgentCore Payments** enables AI agents to make autonomous
payments for digital services. Agents never hold private keys or require
human approval for each transaction.
This use case builds two Strands agents that buy metered access to a paid
HTTP API through AgentCore Payments. One agent signs on the Ethereum
Virtual Machine (EVM) (Base Sepolia) and the other on Solana (Solana
Devnet). The seller is a minimal "Fun Facts" service deployed via AWS
CDK: an Amazon API Gateway HTTP API backed by an AWS Lambda function
that charges **$0.01** per call and accepts either network in the x402
response.
When an agent requests a fact, the seller returns HTTP 402 with a
payment requirement. The agent forwards the requirement to AgentCore
Payments' `ProcessPayment` operation and receives a signed proof. It
then retries the request with the proof attached and returns the paid
fact. The agent is designed so it never needs to touch a private key.
Internally, AgentCore Payments manages the wallet, the signing keys,
and the on-chain settlement. Whether the `PaymentManager` is wired to
**Coinbase Developer Platform (CDP)** or **Stripe via Privy**, the
agent code is identical. The service picks the right signer from the
connector tied to the instrument.
This notebook is **self-contained**. It provisions a full AgentCore
Payments stack inline (§5), creates two `EMBEDDED_CRYPTO_WALLET`
instruments under the same connector (ETHEREUM + SOLANA), and deploys
the seller from a CDK stack that lives alongside it (§3). If a
`PaymentManager` and at least one `PaymentInstrument` already exist,
the notebook detects them in §4 and skips the inline setup.
### Use Case Details
| Information | Details |
|:--------------------|:----------------------------------------------------------------------|
| Use case type | Agentic HTTP API consumption with autonomous micropayment |
| AgentCore components| Amazon Bedrock AgentCore Payments |
| Wallet providers | Coinbase CDP ✅ · Stripe via Privy ✅ |
| Payment protocol | x402 (HTTP 402 Payment Required) on the wire |
| Agent type | Single |
| Agentic Framework | Strands Agents |
| LLM model | Anthropic Claude Sonnet 4.5 (Amazon Bedrock, `us.` inference profile) |
| Example complexity | Intermediate |
| SDK used | boto3 |
### Architecture
Three parties participate in every paid request:
1. **Strands agent** — the only tool it calls is `http_request`. The
`AgentCorePaymentsPlugin` intercepts HTTP 402 responses and handles
the payment handshake transparently.
2. **Amazon Bedrock AgentCore Payments** — receives `ProcessPayment`,
returns a signed x402 proof using the wallet tied to the instrument
(Coinbase CDP or Privy).
3. **Seller (CDK stack)** — AWS Lambda function behind Amazon API
Gateway that issues the 402 challenge, verifies the proof, and
serves the content.
Four IAM roles separate concerns operationally, following the
**principle of least privilege**: each role has only the permissions
required for its specific operation, with explicit `Deny` statements
on actions reserved for other roles:
- `AgentCorePaymentsControlPlaneRole` — manages Manager, Connector, Credential Provider
- `AgentCorePaymentsManagementRole` — manages Instrument and Session (explicit `Deny` on `ProcessPayment`)
- `AgentCorePaymentsProcessPaymentRole` — signs payments, reads Instrument and Session
- `AgentCorePaymentsResourceRetrievalRole` — assumed by AgentCore Payments at runtime to retrieve credentials
`test/integration/setup-roles.sh` creates all four with the right
policies. See the public [IAM roles for AgentCore Payments](https://docs.aws.amazon.com/bedrock-agentcore/latest/devguide/payments-iam-roles.html)
reference for the full policy details and an explanation of the
separation-of-duties model.
<div style="text-align:left">
<img src="images/architecture_pay_for_api.png" alt="Pay-for-API architecture diagram: a user prompts a Strands agent on AgentCore Runtime, the agent calls a paid HTTP API on Amazon API Gateway plus AWS Lambda, the seller returns HTTP 402 with a payment requirement, AgentCore Payments signs the payment via Coinbase CDP or Stripe via Privy, the agent retries with the signed proof, the seller settles on chain through the x402 facilitator and returns 200 OK, and the operator audits spend through GetPaymentSession." width="75%"/>
</div>
**Numbered flow (matches the diagram)**
1. **User** sends a query to the **Agent** (AgentCore Runtime + Strands).
2. The agent calls the paid API hosted on **Amazon API Gateway****AWS Lambda**.
3. The seller responds with **HTTP 402 Payment Required** and a payment requirement payload.
4. The agent forwards the requirement to **AgentCore Payments**, which selects the
matching `PaymentInstrument`, checks the session budget, and signs the payment
through the configured wallet provider (Coinbase CDP or Stripe via Privy).
5. The agent retries the request with the signed `X-PAYMENT` header. The seller
verifies, settles on-chain through the x402 facilitator, and returns **200 OK** with the content.
6. The agent answers the user. The operator audits spend through `GetPaymentSession`.
### Use Case Key Features
* Agent is designed not to hold private keys — AgentCore Payments
signs every charge via the configured `PaymentManager` and
`PaymentConnector`
* Wallet-provider-agnostic — the same agent code runs against a Coinbase CDP
instrument or a Stripe-via-Privy instrument
* Human-controlled budget via `maxSpendAmount` on the payment session
* IAM role separation: `ManagementRole` creates sessions, `ProcessPaymentRole` signs
payments (explicit `Deny` in both directions, enforced by IAM rather than
documentation)
* Full audit trail via `GetPaymentSession` — the operator sees exactly what the
agent spent
* Self-contained — the notebook runs from a clean AWS account
---
## Payment Protocol Availability
AgentCore Payments supports multiple wallet providers. The wire format
(x402 for crypto settlement) is an implementation detail. The agent
code in this use case does not change based on provider. The service
picks the right signer from the connector tied to the instrument.
| Wallet Provider | Connector Type | Status | Notes |
|:----------------|:---------------|:-------|:------|
| **Coinbase CDP** | `CoinbaseCDP` | ✅ Available — EVM + Solana | API Key ID, API Key Secret, Wallet Secret. **Enable "Delegated signing"** under Project → Wallet → Embedded Wallets → Policies before use. Inline setup in §5 provisions a Coinbase CDP wallet. |
| **Stripe** (via Privy) | `StripePrivy` | ✅ Available — EVM + Solana | App ID, App Secret, Authorization Key ID, P-256 Authorization Private Key. Privy returns the private key prefixed with `wallet-auth:`**strip the prefix** before storing it. Inline setup in §5 provisions a Privy-backed wallet. No hub redirect is needed for Privy: the authorization key registered on the credential provider is the signing delegation. |
---
## Prerequisites
- **AWS account** with Amazon Bedrock AgentCore Payments available in your chosen region
- **Amazon Bedrock access** enabled for **Anthropic Claude Sonnet 4.5** in your chosen region (cross-region inference profile `us.anthropic.claude-sonnet-4-5-20250929-v1:0`)
- **Python 3.10+** with a Jupyter kernel. If you hit "Running cells requires the ipykernel package", install it once: `python3 -m pip install ipykernel --user`. Any Jupyter frontend works — JupyterLab (4.0+), classic Jupyter Notebook (7.0+), VS Code, or Kiro.
- **AWS Command Line Interface (AWS CLI) v2** configured with credentials (`aws configure`)
- **AWS Cloud Development Kit (CDK) v2** installed globally (`npm install -g aws-cdk`); used by the notebook to deploy the seller
- **Node.js 18+** — required by CDK
- **A wallet provider account** — Coinbase Developer Platform (CDP) (API Key ID, API Key Secret, Wallet Secret) or Stripe via Privy (App ID, App Secret, Authorization Key ID, P-256 Authorization Private Key)
- **Testnet USD Coin (USDC)** from the [Circle testnet faucet](https://faucet.circle.com/) on both **Base Sepolia** and **Solana Devnet**, because §5 creates one wallet per network
---
## Security
The use case relies on AgentCore Identity's **payment credential provider**
to manage wallet provider secrets. Once `CreatePaymentCredentialProvider`
runs in §4, AgentCore Identity stores the Coinbase / Privy API keys, app
secrets, and wallet or authorization secrets in **AWS Secrets Manager**,
encrypts them with **AWS Key Management Service (KMS)** keys, and surfaces
only the secret ARN to your agents (see [Configure credential
provider](https://docs.aws.amazon.com/bedrock-agentcore/latest/devguide/resource-providers.html)).
The agent runtime calls `GetResourcePaymentToken` at signing time to
receive a short-lived vendor-specific token; it never sees the raw API key
or wallet secret.
What AgentCore Payments handles for you:
- **Secret storage** — wallet provider secrets land in AWS Secrets Manager
under AgentCore Identity, encrypted with AWS-owned KMS keys (customer-
managed KMS keys supported)
- **Secret retrieval** — agents call `GetResourcePaymentToken` and receive a
vendor token. The agent runtime never receives the underlying API key,
app secret, or wallet secret
- **Audit trail** — every `ProcessPayment` call writes to AWS CloudTrail
and to the AgentCore Payments managed log group. Use `GetPaymentSession`
for operator-visible spend totals
- **Budget enforcement** — the operator sets `maxSpendAmount` on the
payment session. AgentCore Payments rejects any `ProcessPayment` that
would exceed it
- **IAM least privilege** — the four roles in §2 each receive only the
actions and resources required for one operation. Cross-role permissions
are explicitly denied (`ManagementRole` cannot call `ProcessPayment`,
`ProcessPaymentRole` cannot manage sessions or instruments)
What you handle locally:
- **Initial credential paste** — Coinbase / Privy secrets are pasted into
`.env` once, before §4 runs. The notebook reads them only to call
`CreatePaymentCredentialProvider`. After that call returns, the secrets
are inside the AgentCore Identity-managed vault (Secrets Manager) and
the local `.env` copies are no longer needed by the agent. They remain
in `.env` so re-running §4 is idempotent
- **Encryption in transit** — all calls to AgentCore Payments, Amazon
Bedrock, and the seller HTTP API run over TLS (`https://`). The
Dockerfile health check is the only HTTP URL and is loopback-only
### Production hardening
This is an L100 tutorial. Before deploying anything resembling this
sample to production:
- **Drop `.env` after first run.** Once §4 has called
`CreatePaymentCredentialProvider`, blank the secret values from `.env`.
Subsequent notebook runs read the credential provider ARN from `.env`
(which is non-sensitive) and the actual secrets stay in Secrets Manager
- **Use customer-managed KMS keys.** AgentCore Identity defaults to
AWS-owned KMS keys; switch to customer-managed keys for additional
audit and rotation control
- **Tighten IAM role wildcards.** Once Manager IDs are stable, replace
`payment-manager/*` with the specific Manager ARN, or scope by tag
- **Switch the AgentCore Runtime to VPC mode** with private subnets and
VPC endpoints for AWS APIs (the tutorial uses `networkMode=PUBLIC`)
- **Restrict the seller's Amazon API Gateway CORS** to the specific agent
runtime domains that need to call it
- **Pin the `bedrock-agentcore` Python SDK and `@x402/*` Node packages**
to specific versions in production builds
---
## Running the Use Case
Before opening the notebook, create a Python virtual environment so
dependency installs and notebook state stay isolated from the global
Python.
**Option 1 — Terminal (cross-platform)**
```bash
python3 -m venv .venv
source .venv/bin/activate # On Windows: .venv\Scripts\activate
python3 -m pip install --upgrade pip ipykernel
python3 -m ipykernel install --user --name pay-for-api-venv --display-name "Python (pay-for-api-venv)"
```
**Option 2 — VS Code / Kiro**
1. Open `pay-for-api.ipynb`.
2. Choose the kernel selector in the top-right of the notebook (or the
Python version indicator in the bottom status bar).
3. Choose **Python: Create Environment...**.
4. Choose **Venv**.
5. Pick a Python 3.10+ interpreter. The IDE creates `.venv/` and selects
it automatically.
6. When prompted to install kernel dependencies (`ipykernel`), accept.
After the venv is active, open `pay-for-api.ipynb` and run cells in
order. The CLI equivalent of opening the notebook is:
```bash
jupyter notebook pay-for-api.ipynb
```
The notebook handles dependency install, IAM role creation, credential
prompts, seller deploy, payment provisioning, agent runs, and teardown:
- §1 installs the Python dependencies from `requirements.txt`
- §2 creates the four IAM roles and interactively prompts for wallet provider credentials (Coinbase CDP or Stripe via Privy)
- §3 deploys the Fun Facts seller stack via CDK and captures the URL
- §4 decides whether to run inline setup or reuse existing AgentCore Payments infrastructure
- §5 provisions a Credential Provider + Manager + Connector for the chosen provider, then creates two Payment Instruments (ETHEREUM + SOLANA) under the same connector
- §6 creates two budget-limited payment sessions, one per network
- §7 builds the Strands agent factory: one pattern that wraps the `AgentCorePaymentsPlugin` around whichever (instrument, session, network) is passed in
- §8 runs the agent once on EVM and once on Solana against the same seller
- §9 optionally deploys the agent to AgentCore Runtime via `agent/cdk/` and invokes it remotely
- §10 inspects the data plane for both networks: GetPaymentSession, balance, ListPaymentInstruments, ListPaymentSessions
- §11 tears everything down: sessions, seller stack, agent runtime (if §9 was run), and AgentCore Payments resources (optional)
---
## Key Notes
- The seller stack deploys to the same region as AgentCore Payments —
set by `AWS_REGION` in `.env`.
- USDC amounts use 6 decimal places: `"$0.01"``10000` atomic units
on the wire. The `@x402/hono` library handles the conversion.
- The seller emits multi-network `accepts` — one entry for EVM
(Base Sepolia) and one for Solana (Devnet) when both payout wallets
are configured. The agent picks the entry matching the instrument's
network.
- Responses use the `{ x402_content, x402_meta }` shape so the seller
is discoverable through the AgentCore Registry / Bazaar Model
Context Protocol (MCP).
- The `ProcessPaymentRole` has an explicit IAM `Deny` on all session
and instrument management; the `ManagementRole` has an explicit
`Deny` on `ProcessPayment`. The trust boundary is enforced by IAM,
not by documentation.
- The seller verifies payment proofs against the public x402
facilitator (`https://x402.org/facilitator`). Point it at a private
facilitator by editing `seller/lambda/index.js` and redeploying.
- When a `StripePrivy` instrument is used, the agent and the seller do
not change. AgentCore Payments routes the signing request to Privy's
key-management service transparently. Privy-backed instruments
settle on both EVM (Base / Base Sepolia) and Solana (Solana / Solana
Devnet).
- The agent never calls the plugin's read-only management tools
(`get_payment_instrument`, `list_payment_instruments`,
`get_payment_session`). Those are reserved for operator debug flows.
The system prompt in §7 tells the model to use only `http_request`.
---
## Cleanup
> ⚠️ **Cost notice:** The resources deployed in this use case incur
> AWS charges while running. AWS Lambda, Amazon API Gateway, AgentCore
> Runtime, AgentCore Memory, and AgentCore Payments all bill on
> per-request and per-resource models. Run §11 of the notebook to tear
> them down when you are done.
§11 of the notebook handles teardown end-to-end:
| Step | What it does | What it removes |
|------|--------------|-----------------|
| Revoke session | `DeletePaymentSession` on each session created in §6 | Active session budgets (no undelete) |
| Tear down the seller stack | `cdk destroy` on the seller CDK app | Amazon API Gateway HTTP API, AWS Lambda function, IAM execution role |
| Tear down the agent runtime | `cdk destroy` on the agent CDK app (only if §9 was run) | AgentCore Runtime, AgentCore Memory, Amazon ECR repository, AWS CodeBuild project, IAM execution role |
| Tear down AgentCore Payments resources | Calls `DeletePaymentInstrument`, `DeletePaymentConnector`, `DeletePaymentManager`, `DeletePaymentCredentialProvider` in dependency order | All Manager / Connector / Instrument / Credential Provider resources created by §5 |
| Remove local build artifacts | Deletes `.venv/`, `cdk.out/`, `__pycache__/`, `outputs.json`, `privy-delegation/`, `seller/lambda/node_modules/` | Local working-copy files only — no cloud resources |
The IAM roles created by `setup-roles.sh` in §2 have no standing cost
and are retained for re-runs. To delete them by hand:
```bash
aws iam delete-role --role-name AgentCorePaymentsControlPlaneRole
aws iam delete-role --role-name AgentCorePaymentsManagementRole
aws iam delete-role --role-name AgentCorePaymentsProcessPaymentRole
aws iam delete-role --role-name AgentCorePaymentsResourceRetrievalRole
```
CloudWatch log groups under `/aws/bedrock-agentcore/` and `/bedrock-agentcore/payments/`
are retained after teardown so you can review historical traces. Delete
them from the CloudWatch console if you want to clear historical data.
### Manual cleanup (without the notebook)
If the notebook is unavailable, run the same teardown from a shell:
```bash
# 1. Destroy the seller stack
bash test/integration/destroy-seller.sh
# 2. Destroy the agent runtime stack (only if §9 was run)
bash test/integration/destroy-agent.sh
# 3. AgentCore Payments resources require boto3 calls — see §11 of
# the notebook for the exact API sequence.
```
### Verify cleanup succeeded
Confirm no CloudFormation stacks remain:
```bash
aws cloudformation list-stacks \
--stack-status-filter CREATE_COMPLETE UPDATE_COMPLETE \
--query "StackSummaries[?starts_with(StackName, 'AgentCorePayments')].StackName"
```
The output should be empty.
---
## Conclusion
This use case demonstrates how Amazon Bedrock AgentCore Payments
enables an AI agent to make autonomous micropayments for paid HTTP APIs
without holding private keys or requiring per-transaction human
approval. The same agent code paid for the same content through two
different wallet providers (Coinbase CDP and Stripe via Privy) and on
two different networks (EVM and Solana), demonstrating the
provider-agnostic and network-agnostic design.
Key takeaways:
- **Separation of concerns** — IAM roles isolate session creation,
payment signing, and credential retrieval. The trust boundary is
enforced by IAM, not by code.
- **Budget control** — operators set a maximum spend per session.
AgentCore Payments enforces it, and `GetPaymentSession` provides a
full audit trail.
- **Wire format** — x402 (HTTP 402 Payment Required) is the open spec
on the wire. The `@x402/hono` library on the seller side and the
`AgentCorePaymentsPlugin` on the agent side handle the protocol so
that the application code remains a normal HTTP request.
Use the [Learn more](#learn-more) links to go deeper, and adapt the
patterns in this notebook to your own paid-API integrations.
---
## Learn more
Public AgentCore Payments documentation:
- [Overview](https://docs.aws.amazon.com/bedrock-agentcore/latest/devguide/payments.html)
- [How it works](https://docs.aws.amazon.com/bedrock-agentcore/latest/devguide/payments-how-it-works.html)
- [Core concepts](https://docs.aws.amazon.com/bedrock-agentcore/latest/devguide/payments-concepts.html)
- [Prerequisites](https://docs.aws.amazon.com/bedrock-agentcore/latest/devguide/payments-prerequisites.html)
- [IAM roles](https://docs.aws.amazon.com/bedrock-agentcore/latest/devguide/payments-iam-roles.html)
- [Set up a credential provider](https://docs.aws.amazon.com/bedrock-agentcore/latest/devguide/payments-setup-credential-provider.html)
- [Create a payment manager](https://docs.aws.amazon.com/bedrock-agentcore/latest/devguide/payments-create-manager.html)
- [Create a payment instrument](https://docs.aws.amazon.com/bedrock-agentcore/latest/devguide/payments-create-instrument.html)
- [Create a payment session](https://docs.aws.amazon.com/bedrock-agentcore/latest/devguide/payments-create-session.html)
- [Process a payment](https://docs.aws.amazon.com/bedrock-agentcore/latest/devguide/payments-process-payment.html) — plugin reference, interrupt contract, network preferences, `auto_payment=False` for human-in-the-loop flows
- [Connect to Bazaar](https://docs.aws.amazon.com/bedrock-agentcore/latest/devguide/payments-connect-bazaar.html) — make a seller discoverable through the AgentCore Registry
Announcement:
[Agents that transact — Introducing Amazon Bedrock AgentCore Payments, built with Coinbase and Stripe](https://aws.amazon.com/blogs/machine-learning/agents-that-transact-introducing-amazon-bedrock-agentcore-payments-built-with-coinbase-and-stripe/)
@@ -0,0 +1,146 @@
# Pay-For-API — Buyer Agent
A minimal Strands Agent, wired for Amazon Bedrock Claude Sonnet 4.5,
that buys a fact from the seller API by delegating the x402 payment to
**Amazon Bedrock AgentCore Payments** through the
`AgentCorePaymentsPlugin`.
Two ways to run the same agent:
| Mode | Where | When |
|------|-------|------|
| **Local** | Notebook cell in `pay-for-api.ipynb` (§8) | Teaching / fast iteration |
| **Runtime** | AgentCore Runtime container deployed via CDK (§9) | Production-shaped deploy |
The agent code is identical in both modes. The container folder wraps
the same `Agent()` construction in a FastAPI `/invocations` endpoint
so it fits the AgentCore Runtime contract.
## Prerequisites
Before deploying the agent runtime, complete the parent use-case
prerequisites in [`../README.md`](../README.md). Specifically:
- AWS account with Amazon Bedrock AgentCore Payments enabled in the
target region
- Amazon Bedrock model access for `us.anthropic.claude-sonnet-4-5-20250929-v1:0`
- AWS CDK v2 (`npm install -g aws-cdk`) and Node.js 18+
- Python 3.10+ with the use-case venv active
- Completed §1-§6 of the parent notebook (so a `PaymentManager`,
`PaymentInstrument`, and at least one `PaymentSession` exist for the
runtime to invoke against)
## Folder layout
```
agent/
├── cdk/
│ ├── app.py CDK app entry point
│ ├── agent_stack.py ECR + IAM + Runtime
│ ├── cdk.json
│ └── requirements.txt
├── container/
│ ├── Dockerfile
│ ├── agent.py FastAPI server + Strands Agent
│ └── requirements.txt
└── README.md
```
## How the payment flow works
1. The agent tries `http_request.GET <seller-url>/facts?topic=<x>`.
2. The seller returns **HTTP 402** with an x402 `accepts` array.
3. `AgentCorePaymentsPlugin` intercepts the 402, calls
**`ProcessPayment`** against the configured Payment Manager,
Session, and Instrument, receives the signed `CRYPTO_X402` proof,
base64-encodes it into the `X-PAYMENT` header (per the x402 protocol
spec), and retries the request transparently.
4. The seller verifies the proof with the x402 facilitator, settles
on-chain, and returns the paid fact as **HTTP 200**.
The agent never sees a private key, never assembles the `X-PAYMENT`
header, and never touches a boto3 client. The only tool it calls is
`http_request`. The plugin does also register three read-only
management tools (`get_payment_instrument`,
`list_payment_instruments`, `get_payment_session`) but the system
prompt in §7 of the notebook tells the model not to use them — they
are reserved for operator debug flows.
## Identity model
- Every payment operation runs under the **vendor-level user ID** from
`paymentInstrument.userId` — the value the service returns on
`CreatePaymentInstrument`. The notebook captures that ID and passes
it to the agent as `paymentUserId` on invocation.
- For Privy-backed instruments, this is the Privy DID.
- For Coinbase-backed instruments, this is the CDP end-user UUID (hub
flow).
- There is **no tenant/Cognito sub on the wire** — identity is
vendor-rooted end to end.
## Deploy
> ⚠️ **Cost notice:** This deploys an AgentCore Runtime, an Amazon ECR
> repository, an AWS CodeBuild project, an AgentCore Memory resource,
> and the supporting CloudWatch log groups. CodeBuild (per-build-minute)
> and the Runtime (per-invocation) are the highest-cost items. Run the
> [Clean up](#clean-up) steps when you are done.
```bash
cd agent/cdk
python3 -m venv .venv && source .venv/bin/activate
pip install -r requirements.txt
cdk bootstrap # only once per account/region
cdk deploy
```
Outputs: `AgentRuntimeArn`, `AgentRuntimeEndpoint`,
`AgentExecutionRoleArn`, `AgentEcrRepoUri`,
`AgentBuildProjectName`, `AgentMemoryId`.
The notebook's §9 calls into the CDK for you. See
`pay-for-api.ipynb`.
## Clean up
Tear the runtime down when you no longer need it. The notebook's §11
runs the same teardown plus the AgentCore Payments resource cleanup.
```bash
bash test/integration/destroy-agent.sh
```
Or directly through CDK:
```bash
cd agent/cdk
source .venv/bin/activate
cdk destroy
```
This removes the AgentCore Runtime, the AgentCore Memory resource, the
ECR repository (with its images), and the CodeBuild project. Verify by
listing CloudFormation stacks:
```bash
aws cloudformation list-stacks \
--stack-status-filter CREATE_COMPLETE UPDATE_COMPLETE \
--query "StackSummaries[?starts_with(StackName, 'AgentCorePaymentsBuyerAgent')].StackName"
```
The output should be empty.
## Conclusion
This folder packages the buyer-side half of the Pay-For-API use case
into a deployable AgentCore Runtime. The same Strands Agent pattern
runs locally in §7 of the parent notebook and in production-shaped
fashion through the CDK stack here, demonstrating how to graduate a
local agent prototype to a managed runtime without code changes. The
`AgentCorePaymentsPlugin` makes the x402 payment flow transparent to
the agent, so the same `http_request` tool call pays for content
through whichever wallet provider the operator configured.
For a deeper walkthrough, run `pay-for-api.ipynb` end to end. For the
service-side reference, see the
[AgentCore Payments documentation](https://docs.aws.amazon.com/bedrock-agentcore/latest/devguide/payments.html).
@@ -0,0 +1,573 @@
"""Pay for API — buyer agent CDK stack.
Provisions the full AgentCore Runtime stack for the buyer agent **without
requiring Docker on the machine running `cdk deploy`**:
1. **Amazon S3 asset** — zips and uploads ``agent/container/`` to the CDK bootstrap
assets bucket.
2. **Amazon ECR repository** — destination for the built image.
3. **AWS CodeBuild project** — ARM64 Linux environment that pulls the S3
asset, runs ``docker build``, and pushes to ECR. Runs in AWS, so the
caller needs only ``cdk deploy`` and AWS credentials.
4. **Build trigger AWS Lambda function** — custom resource that starts
the CodeBuild run and polls until the image is in ECR before the
Runtime resource is created.
5. **IAM execution role** with the minimum perms the runtime needs at
invoke time (Amazon Bedrock, AgentCore Payments data plane, Amazon
CloudWatch Logs, AWS X-Ray, Amazon CloudWatch Application Signals,
vended log delivery).
6. **AgentCore Runtime** pointing at the freshly-built image.
Outputs the Runtime ARN, invoke URL, and execution role ARN so the
notebook can invoke the deployed agent by name.
"""
from __future__ import annotations
from pathlib import Path
from aws_cdk import (
CfnOutput,
CustomResource,
Duration,
RemovalPolicy,
Stack,
aws_bedrockagentcore as bedrockagentcore,
aws_codebuild as codebuild,
aws_ecr as ecr,
aws_iam as iam,
aws_lambda as aws_lambda,
aws_s3_assets as s3_assets,
)
from constructs import Construct
# The container source lives in a sibling folder to cdk/ — resolve the
# absolute path once so S3 asset + docker build share the same context.
CONTAINER_DIR = str(Path(__file__).resolve().parent.parent / "container")
class AgentCorePaymentsBuyerAgentStack(Stack):
"""AgentCore Runtime + IAM for the Pay for API buyer agent."""
def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
super().__init__(scope, construct_id, **kwargs)
# ── ECR repository ──
agent_repo = ecr.Repository(
self,
"AgentEcrRepo",
repository_name="pay-for-api-agent",
removal_policy=RemovalPolicy.DESTROY,
empty_on_delete=True,
lifecycle_rules=[
ecr.LifecycleRule(
max_image_count=5,
description="Keep the 5 most recent images",
)
],
)
# ── S3 asset: zip of agent/container/ ──
# CDK uploads this to the bootstrap assets bucket on every
# `cdk deploy`. CodeBuild pulls it from S3 — no GitHub, no
# CodeCommit, no Docker-on-laptop.
agent_source = s3_assets.Asset(
self,
"AgentSourceAsset",
path=CONTAINER_DIR,
)
# ── CodeBuild project ──
build_project = codebuild.Project(
self,
"AgentBuildProject",
project_name="pay-for-api-agent-build",
environment=codebuild.BuildEnvironment(
# ARM64 matches AgentCore Runtime's Graviton hosts.
build_image=codebuild.LinuxArmBuildImage.AMAZON_LINUX_2_STANDARD_3_0,
compute_type=codebuild.ComputeType.SMALL,
privileged=True, # docker-in-docker for image build
),
source=codebuild.Source.s3(
bucket=agent_source.bucket,
path=agent_source.s3_object_key,
),
environment_variables={
"AWS_ACCOUNT_ID": codebuild.BuildEnvironmentVariable(value=self.account),
"AWS_DEFAULT_REGION": codebuild.BuildEnvironmentVariable(value=self.region),
"ECR_REPO_URI": codebuild.BuildEnvironmentVariable(value=agent_repo.repository_uri),
"IMAGE_TAG": codebuild.BuildEnvironmentVariable(value=agent_source.asset_hash),
},
build_spec=codebuild.BuildSpec.from_object(
{
"version": "0.2",
"phases": {
"pre_build": {
"commands": [
"echo Logging in to ECR...",
"aws ecr get-login-password --region $AWS_DEFAULT_REGION | "
"docker login --username AWS --password-stdin "
"$AWS_ACCOUNT_ID.dkr.ecr.$AWS_DEFAULT_REGION.amazonaws.com",
],
},
"build": {
"commands": [
"echo Building agent image...",
"docker build -t $ECR_REPO_URI:$IMAGE_TAG .",
],
},
"post_build": {
"commands": [
"echo Pushing to ECR...",
"docker push $ECR_REPO_URI:$IMAGE_TAG",
"docker tag $ECR_REPO_URI:$IMAGE_TAG $ECR_REPO_URI:latest",
"docker push $ECR_REPO_URI:latest",
],
},
},
}
),
)
agent_repo.grant_pull_push(build_project)
# ── Custom resource: kick off the build and wait for it to finish ──
# The Runtime resource below references the image URI — we need the
# image in ECR before CloudFormation moves past this step.
build_trigger_role = iam.Role(
self,
"BuildTriggerRole",
assumed_by=iam.ServicePrincipal("lambda.amazonaws.com"),
managed_policies=[
iam.ManagedPolicy.from_aws_managed_policy_name("service-role/AWSLambdaBasicExecutionRole"),
],
)
build_trigger_role.add_to_policy(
iam.PolicyStatement(
actions=["codebuild:StartBuild", "codebuild:BatchGetBuilds"],
resources=[build_project.project_arn],
)
)
build_trigger_fn = aws_lambda.Function(
self,
"BuildTriggerFn",
function_name="pay-for-api-agent-build-trigger",
runtime=aws_lambda.Runtime.PYTHON_3_12,
handler="index.handler",
role=build_trigger_role,
timeout=Duration.minutes(15),
memory_size=128,
code=aws_lambda.Code.from_inline(
r"""
import json
import time
import urllib.request
import boto3
def handler(event, context):
props = event.get("ResourceProperties", {})
project_name = props.get("ProjectName", "")
# No rebuild on stack delete — ECR contents are torn down by the
# repository's lifecycle.
if event["RequestType"] == "Delete":
return _respond(event, context, "SUCCESS", {"ImageBuilt": "skipped"})
cb = boto3.client("codebuild")
try:
build = cb.start_build(projectName=project_name)
build_id = build["build"]["id"]
print(f"Started CodeBuild: {build_id}")
# Poll every 30 seconds for up to ~14 minutes.
for _ in range(28):
time.sleep(30)
result = cb.batch_get_builds(ids=[build_id])
status = result["builds"][0]["buildStatus"]
print(f"Build status: {status}")
if status == "SUCCEEDED":
return _respond(event, context, "SUCCESS", {"BuildId": build_id})
if status in ("FAILED", "FAULT", "STOPPED", "TIMED_OUT"):
return _respond(
event, context, "FAILED",
{"Error": f"CodeBuild {status}"},
)
return _respond(event, context, "FAILED", {"Error": "Build timed out"})
except Exception as exc: # noqa: BLE001
print(f"Error: {exc}")
return _respond(event, context, "FAILED", {"Error": str(exc)})
def _respond(event, context, status, data):
body = json.dumps({
"Status": status,
"Reason": json.dumps(data),
"PhysicalResourceId": context.log_stream_name,
"StackId": event["StackId"],
"RequestId": event["RequestId"],
"LogicalResourceId": event["LogicalResourceId"],
"Data": data,
})
req = urllib.request.Request(
event["ResponseURL"],
data=body.encode(),
method="PUT",
headers={"Content-Type": ""},
)
urllib.request.urlopen(req)
"""
),
)
trigger_build = CustomResource(
self,
"TriggerImageBuild",
service_token=build_trigger_fn.function_arn,
properties={
"ProjectName": build_project.project_name,
# Tie the CR hash to the asset hash — any change in
# agent/container/ triggers a rebuild automatically.
"SourceHash": agent_source.asset_hash,
},
)
# ── IAM: runtime execution role ──
execution_role = iam.Role(
self,
"AgentExecutionRole",
assumed_by=iam.ServicePrincipal("bedrock-agentcore.amazonaws.com"),
description=(
"Pay for API buyer agent runtime execution role. "
"Grants Bedrock model invoke + AgentCore Payments DP ops the "
"AgentCorePaymentsPlugin needs at runtime."
),
)
# Bedrock model invoke — Claude Sonnet 4.5 via the cross-region US
# inference profile. Both the foundation model ARN and the
# inference-profile ARN are granted because Bedrock resolves
# through the profile.
execution_role.add_to_policy(
iam.PolicyStatement(
actions=[
"bedrock:InvokeModel",
"bedrock:InvokeModelWithResponseStream",
],
resources=[
# Inference profile (cross-region routing)
f"arn:aws:bedrock:{self.region}:{self.account}:inference-profile/us.anthropic.claude-sonnet-4-5-20250929-v1:0",
# Underlying foundation model in each US region the profile
# can route to.
"arn:aws:bedrock:us-east-1::foundation-model/anthropic.claude-sonnet-4-5-20250929-v1:0",
"arn:aws:bedrock:us-east-2::foundation-model/anthropic.claude-sonnet-4-5-20250929-v1:0",
"arn:aws:bedrock:us-west-2::foundation-model/anthropic.claude-sonnet-4-5-20250929-v1:0",
],
)
)
# AgentCore Payments data-plane operations the plugin calls at
# runtime. The Manager / Instrument / Session IDs are not known
# at role creation time (the notebook creates them in §4), so
# the resource list is wildcarded to all PaymentManagers in the
# caller's account. Production hardening: scope to the specific
# Manager ARN once it is stable, or add a tag-based condition
# on `aws:ResourceTag/Project`.
execution_role.add_to_policy(
iam.PolicyStatement(
actions=[
"bedrock-agentcore:ProcessPayment",
"bedrock-agentcore:GetPaymentSession",
"bedrock-agentcore:GetPaymentInstrument",
"bedrock-agentcore:GetPaymentInstrumentBalance",
"bedrock-agentcore:GetResourcePaymentToken",
],
resources=[
f"arn:aws:bedrock-agentcore:{self.region}:{self.account}:payment-manager/*",
f"arn:aws:bedrock-agentcore:{self.region}:{self.account}:payment-manager/*/instrument/*",
f"arn:aws:bedrock-agentcore:{self.region}:{self.account}:payment-manager/*/session/*",
],
)
)
# CloudWatch Logs — Runtime expects the role to be able to write its
# own log stream.
execution_role.add_to_policy(
iam.PolicyStatement(
actions=[
"logs:CreateLogGroup",
"logs:CreateLogStream",
"logs:PutLogEvents",
"logs:DescribeLogGroups",
"logs:DescribeLogStreams",
],
resources=[
f"arn:aws:logs:{self.region}:{self.account}:log-group:/aws/bedrock-agentcore/*",
],
)
)
# ── Observability ──
# The agent container runs AWS Distro for OpenTelemetry and also
# wires up CloudWatch Logs vended delivery for the PaymentManager
# on first invocation (see `_ensure_vended_log_delivery` in
# agent.py). Both paths need the permissions below.
# Logs vended-delivery pipeline: Payments → CloudWatch Logs.
# The delivery source/destination/delivery objects are not
# resource-scoped (CloudWatch Logs creates them per-region per-
# account), so the resource list stays wildcarded. The log
# group writes themselves are scoped to the agentcore-payments
# log group prefix.
execution_role.add_to_policy(
iam.PolicyStatement(
sid="CloudWatchLogsVendedDelivery",
actions=[
"logs:CreateDelivery",
"logs:CreateLogGroup",
"logs:CreateLogStream",
"logs:DeleteDelivery",
"logs:DeleteDeliveryDestination",
"logs:DeleteDeliverySource",
"logs:DeleteLogGroup",
"logs:DeleteResourcePolicy",
"logs:DescribeLogGroups",
"logs:DescribeResourcePolicies",
"logs:GetDelivery",
"logs:GetDeliveryDestination",
"logs:GetDeliverySource",
"logs:PutDeliveryDestination",
"logs:PutDeliverySource",
"logs:PutLogEvents",
"logs:PutResourcePolicy",
"logs:PutRetentionPolicy",
],
# CloudWatch Logs does not permit resource-level scoping
# on Describe* and Put*Delivery* APIs. The log group
# actions are implicitly scoped by the delivery target,
# which we restrict via DeliveryDestination. Production
# hardening: scope to specific log group prefixes once
# stable.
resources=["*"],
)
)
# X-Ray + CloudWatch Application Signals — ADOT emit targets.
# X-Ray and Application Signals do not accept resource-level
# ARNs on these actions; the documented IAM policy for ADOT
# observability uses Resource: "*". The agent's traces are
# implicitly scoped to its own session via OpenTelemetry
# context, not via IAM.
execution_role.add_to_policy(
iam.PolicyStatement(
sid="XRayApplicationSignalsCloudTrail",
actions=[
"xray:GetTraceSegmentDestination",
"xray:ListResourcePolicies",
"xray:PutResourcePolicy",
"xray:PutTelemetryRecords",
"xray:PutTraceSegments",
"xray:UpdateTraceSegmentDestination",
"application-signals:StartDiscovery",
"cloudtrail:CreateServiceLinkedChannel",
],
resources=["*"],
)
)
# Service-linked role for Application Signals — created once per
# account, condition-scoped to that specific SLR.
execution_role.add_to_policy(
iam.PolicyStatement(
sid="CreateServiceLinkedRoleForAppSignals",
actions=["iam:CreateServiceLinkedRole"],
resources=[
"arn:*:iam::*:role/aws-service-role/"
"application-signals.cloudwatch.amazonaws.com/"
"AWSServiceRoleForCloudWatchApplicationSignals",
],
)
)
# PaymentsAllowVendedLogDeliveryForResource +
# AllowVendedLogDeliveryForResource on the PaymentManager —
# what lets Payments emit logs through the vended pipeline
# above. CloudWatch checks both actions implicitly when
# `logs.put_delivery_source` runs against a Payment Manager
# ARN: the Payments-prefixed one as the product-level gate, the
# unprefixed one as the AgentCore-wide gate. Scoped to
# PaymentManager resources only.
execution_role.add_to_policy(
iam.PolicyStatement(
sid="BedrockAgentCorePaymentsVendedLogDelivery",
actions=[
"bedrock-agentcore:PaymentsAllowVendedLogDeliveryForResource",
"bedrock-agentcore:AllowVendedLogDeliveryForResource",
],
resources=[
f"arn:aws:bedrock-agentcore:{self.region}:{self.account}:payment-manager/*",
],
)
)
# ECR pull — the runtime pulls the image we built above.
agent_repo.grant_pull(execution_role)
# Allow this role to be passed to bedrock-agentcore.amazonaws.com.
execution_role.add_to_policy(
iam.PolicyStatement(
actions=["iam:PassRole"],
resources=[execution_role.role_arn],
conditions={"StringEquals": {"iam:PassedToService": "bedrock-agentcore.amazonaws.com"}},
)
)
# ── AgentCore Memory ──
# Persistent conversation memory for the buyer agent. Short
# event expiry because the demo is stateless between notebook
# runs; bump to 30+ days for real workloads.
agent_memory = bedrockagentcore.CfnMemory(
self,
"AgentMemory",
name="pay_for_api_agent_memory",
description=(
"Conversation memory for the Pay for API buyer agent. "
"Each invocation gets its own session under the caller's "
"paymentUserId actor."
),
event_expiry_duration=7,
)
# Grant runtime role the memory CRUD actions it needs at invoke
# time. Scoped to the Memory resource we just created.
execution_role.add_to_policy(
iam.PolicyStatement(
sid="AgentCoreMemoryCRUD",
actions=[
"bedrock-agentcore:CreateMemory",
"bedrock-agentcore:GetMemory",
"bedrock-agentcore:UpdateMemory",
"bedrock-agentcore:DeleteMemory",
"bedrock-agentcore:CreateMemoryRecord",
"bedrock-agentcore:GetMemoryRecord",
"bedrock-agentcore:UpdateMemoryRecord",
"bedrock-agentcore:ListMemoryRecords",
"bedrock-agentcore:SearchMemoryRecords",
"bedrock-agentcore:DeleteMemoryRecord",
"bedrock-agentcore:CreateEvent",
"bedrock-agentcore:ListEvents",
"bedrock-agentcore:GetEvent",
"bedrock-agentcore:DeleteEvent",
"bedrock-agentcore:ListActors",
"bedrock-agentcore:ListSessions",
],
resources=[
f"arn:aws:bedrock-agentcore:{self.region}:{self.account}:memory/*",
],
)
)
# ── AgentCore Runtime ──
# containerUri points at the image we built in CodeBuild. The
# asset_hash is used as the tag so a change in agent/container/
# cycles a new image + triggers Runtime update.
#
# networkMode=PUBLIC: the runtime container has outbound
# internet access, which the agent uses to call the seller's
# HTTP API. For production deployments that integrate with
# private services, switch to VPC mode and route the runtime
# through a NAT Gateway with VPC endpoints for AWS APIs.
runtime = bedrockagentcore.CfnRuntime(
self,
"AgentRuntime",
agent_runtime_name="pay_for_api_agent_runtime",
description=(
"Pay for API buyer agent — Strands Agent with Claude Sonnet "
"4.5 and AgentCorePaymentsPlugin for autonomous x402 payment."
),
role_arn=execution_role.role_arn,
network_configuration={"networkMode": "PUBLIC"},
protocol_configuration="HTTP",
agent_runtime_artifact={
"containerConfiguration": {
"containerUri": f"{agent_repo.repository_uri}:{agent_source.asset_hash}",
},
},
environment_variables={
"AWS_REGION": self.region,
"MODEL_ID": "us.anthropic.claude-sonnet-4-5-20250929-v1:0",
"ENABLE_PAYMENTS_PLUGIN": "1",
# Turn on the vended log delivery wiring in agent.py on
# first invocation. Set to "0" for debugging.
"ENABLE_VENDED_LOG_DELIVERY": "1",
# AgentCore Memory resource the agent attaches to via
# AgentCoreMemorySessionManager in agent.py.
"BEDROCK_AGENTCORE_MEMORY_ID": agent_memory.attr_memory_id,
# ADOT auto-instrumentation (matches the defaults in
# agent.py so opentelemetry-instrument picks them up too).
"AGENT_OBSERVABILITY_ENABLED": "true",
"OTEL_PYTHON_DISTRO": "aws_distro",
"OTEL_PYTHON_CONFIGURATOR": "aws_configurator",
"OTEL_EXPORTER_OTLP_PROTOCOL": "http/protobuf",
"OTEL_TRACES_EXPORTER": "otlp",
"OTEL_LOGS_EXPORTER": "otlp",
"OTEL_METRICS_EXPORTER": "none",
},
)
# Runtime must wait on the CodeBuild-built image being ready.
runtime.node.add_dependency(trigger_build)
# And on the memory resource being created so the env var is resolvable.
runtime.node.add_dependency(agent_memory)
# ── Outputs ──
CfnOutput(
self,
"AgentRuntimeArn",
value=runtime.attr_agent_runtime_arn,
description="ARN of the deployed AgentCore Runtime",
)
CfnOutput(
self,
"AgentRuntimeId",
value=runtime.attr_agent_runtime_id,
description="ID of the deployed AgentCore Runtime",
)
CfnOutput(
self,
"AgentRuntimeEndpoint",
# Resolved at deploy time: the {region} and {runtime_id}
# placeholders are substituted into the AgentCore endpoint
# template by the CDK f-string before CloudFormation sees
# the value.
value=(
f"https://bedrock-agentcore.{self.region}.amazonaws.com/"
f"runtimes/{runtime.attr_agent_runtime_id}/invocations"
),
description="Invoke URL for the deployed Runtime",
)
CfnOutput(
self,
"AgentExecutionRoleArn",
value=execution_role.role_arn,
description="IAM role the Runtime assumes at invoke time",
)
CfnOutput(
self,
"AgentEcrRepoUri",
value=agent_repo.repository_uri,
description="ECR repository URI the Runtime pulls from",
)
CfnOutput(
self,
"AgentBuildProjectName",
value=build_project.project_name,
description="CodeBuild project that builds the agent image",
)
CfnOutput(
self,
"AgentMemoryId",
value=agent_memory.attr_memory_id,
description="AgentCore Memory resource the runtime uses for sessions",
)
@@ -0,0 +1,30 @@
#!/usr/bin/env python3
"""CDK app entry point for the Pay for API buyer agent runtime."""
import os
import aws_cdk as cdk
from agent_stack import AgentCorePaymentsBuyerAgentStack
app = cdk.App()
env = cdk.Environment(
account=os.environ.get("CDK_DEFAULT_ACCOUNT"),
region=os.environ.get(
"CDK_DEFAULT_REGION",
os.environ.get("AWS_REGION", "us-west-2"),
),
)
AgentCorePaymentsBuyerAgentStack(
app,
"AgentCorePaymentsBuyerAgentStack",
env=env,
description=(
"AgentCore Payments sample — Pay for API buyer agent (Strands Agent + "
"AgentCorePaymentsPlugin, deployed to AgentCore Runtime)"
),
)
app.synth()
@@ -0,0 +1,18 @@
{
"app": "python3 app.py",
"watch": {
"include": ["**"],
"exclude": [
"README.md",
"cdk*.json",
"requirements*.txt",
"source.bat",
"**/__init__.py",
"**/__pycache__",
"tests"
]
},
"context": {
"@aws-cdk/aws-lambda:recognizeVersionProps": true
}
}
@@ -0,0 +1,2 @@
aws-cdk-lib>=2.140.0
constructs>=10.3.0,<11.0.0
@@ -0,0 +1,34 @@
FROM public.ecr.aws/docker/library/python:3.12-slim
WORKDIR /app
COPY requirements.txt .
# curl ships in slim only as a tiny binary — needed for HEALTHCHECK.
# Keep apt cache wipe inline so the layer stays small.
RUN apt-get update \
&& apt-get install -y --no-install-recommends curl \
&& rm -rf /var/lib/apt/lists/* \
&& pip install --no-cache-dir -r requirements.txt
# Agent source
COPY agent.py .
# Non-root user
RUN useradd -m -r agent && chown -R agent:agent /app
USER agent
EXPOSE 8080
# AgentCore Runtime requires GET /ping returning 200; we hit the same
# endpoint here so docker / orchestrators can detect a stuck process.
# The URL is loopback-only (the request never leaves the container)
# so HTTP without TLS is appropriate and intentional — there's no
# network path for an attacker to intercept it.
HEALTHCHECK --interval=30s --timeout=5s --start-period=20s --retries=3 \
CMD curl -fsS http://127.0.0.1:8080/ping || exit 1
# Wrap with opentelemetry-instrument so ADOT's auto-instrumentation
# attaches to uvicorn + boto3 at process start. The in-process
# `_load_instrumentors()` fallback in agent.py covers the case where
# this wrapper is not used (e.g. local dev).
CMD ["opentelemetry-instrument", "python", "agent.py"]
@@ -0,0 +1,473 @@
"""Pay for API — AgentCore Runtime buyer agent.
A minimal Strands Agent, wrapped in a FastAPI ``/invocations`` endpoint so it
conforms to the AgentCore Runtime contract. The agent has exactly one tool —
``http_request`` from ``strands-agents-tools`` — and relies on the
``AgentCorePaymentsPlugin`` (from ``bedrock-agentcore``) to transparently
handle HTTP 402 → ``ProcessPayment`` → retry.
No private keys. No manual x402 assembly. The caller supplies the payment
context (manager ARN, instrument ID, session ID, vendor-level user ID) on
every invocation, mirroring the pattern in ``agentcore-payments/payment-agent``.
Runtime invocation contract:
POST /invocations
{
"prompt": "Tell me one fact about space",
"sellerUrl": "https://example.com/",
"managerArn": "arn:aws:bedrock-agentcore:…:payment-manager/…",
"instrumentId": "payment-instrument-…",
"sessionId": "payment-session-…",
"paymentUserId": "<CDP UUID | Privy DID>",
"region": "us-west-2" # optional, defaults to AWS_REGION
}
Health endpoint:
GET /ping → {"status": "ok"}
"""
from __future__ import annotations
# ── ADOT auto-instrumentation (must run before any other imports) ──
# These env vars tell AWS Distro for OpenTelemetry how to export traces
# and logs to CloudWatch via the ADOT collector that AgentCore Runtime
# injects into the container. Setting them at the top of the module is
# required because some OTEL libraries read env at import time.
import os
os.environ.setdefault("AGENT_OBSERVABILITY_ENABLED", "true")
os.environ.setdefault("OTEL_PYTHON_DISTRO", "aws_distro")
os.environ.setdefault("OTEL_PYTHON_CONFIGURATOR", "aws_configurator")
os.environ.setdefault("OTEL_EXPORTER_OTLP_PROTOCOL", "http/protobuf")
os.environ.setdefault("OTEL_TRACES_EXPORTER", "otlp")
os.environ.setdefault("OTEL_LOGS_EXPORTER", "otlp")
# Metrics disabled — traces + logs cover the observability surface we care
# about (payment calls, tool use, HTTP requests). Enable if you need them.
os.environ.setdefault("OTEL_METRICS_EXPORTER", "none")
try:
from opentelemetry.instrumentation.auto_instrumentation._load import (
_load_configurators,
_load_distro,
_load_instrumentors,
)
_distro = _load_distro()
_distro.configure()
_load_configurators()
_load_instrumentors(_distro)
except Exception as _otel_err: # noqa: BLE001 — ADOT optional for local dev
import sys
print(f"[WARN] ADOT auto-instrumentation skipped: {_otel_err}", file=sys.stderr)
# ── Standard imports ──
import logging
import boto3
import botocore.exceptions
import fastapi
import uvicorn
from fastapi import FastAPI
from fastapi.responses import JSONResponse
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s %(name)s %(levelname)s %(message)s",
)
logger = logging.getLogger("pay-for-api-agent")
AWS_REGION = os.environ.get("AWS_REGION", "us-west-2")
# Claude Sonnet 4.5 cross-region inference profile (US).
MODEL_ID = os.environ.get(
"MODEL_ID",
"us.anthropic.claude-sonnet-4-5-20250929-v1:0",
)
# AgentCore Memory — if set, every invocation threads through an
# AgentCoreMemorySessionManager keyed on (memory_id, actor_id=paymentUserId,
# session_id=per-invocation). The CDK stack sets this in the container's
# environment; if the variable is missing the agent runs without memory.
MEMORY_ID = os.environ.get("BEDROCK_AGENTCORE_MEMORY_ID", "")
# AgentCorePaymentsPlugin gate. Defaults to enabled in the container — the
# runtime is isolated and the notebook is driving the invocation. Flip to
# "0" / "false" to fall back to a no-payments agent for debugging.
ENABLE_PAYMENTS_PLUGIN = os.environ.get("ENABLE_PAYMENTS_PLUGIN", "1").lower() in (
"1",
"true",
"yes",
)
# AgentCore Payments vended-log delivery gate. When enabled and a
# `managerArn` is supplied on the first invocation, the agent configures
# CloudWatch Logs vended delivery for that Manager. Idempotent — re-runs
# are no-ops. Defaults to enabled.
ENABLE_VENDED_LOG_DELIVERY = os.environ.get("ENABLE_VENDED_LOG_DELIVERY", "1").lower() in ("1", "true", "yes")
# Track Manager ARNs we have already configured vended delivery for, so
# the agent doesn't re-call the control-plane on every invocation.
_VENDED_LOG_DELIVERY_CONFIGURED: set[str] = set()
SYSTEM_PROMPT = (
"You are a research agent powered by Amazon Bedrock AgentCore Payments.\n"
"\n"
"Your only tool is `http_request`. Use it to fetch paid facts from the\n"
"Fun Facts API. Each `GET` returns exactly one fact and costs $0.01 in\n"
"USDC. The AgentCore Payments plugin pays on your behalf — you never\n"
"handle private keys, assemble payment headers, or retry failed calls.\n"
"\n"
"SELLER CONTRACT\n"
" Endpoint: GET <seller>/facts?topic=<topic>\n"
" Supported topics: space, oceans, ai, payments\n"
" (any other value falls back to a random general fact)\n"
' Success body: {"x402_content": {"data": "<JSON string>", ...},\n'
' "x402_meta": {"seller": ..., "generated_at": ...}}\n'
" `x402_content.data` is a JSON string — parse it to\n"
' read `{"topic": ..., "fact": ...}`.\n'
" Price per call: $0.01 USDC.\n"
"\n"
"RULES\n"
" 1. One `http_request` GET per topic the user asks about.\n"
" If the user asks for two topics, make two calls.\n"
" 2. If the user's topic is not in the supported list, pick the closest\n"
" supported topic rather than letting the seller fall back silently —\n"
" e.g. 'volcanoes''space', 'whales''oceans'.\n"
" 3. Parse `x402_content.data` to get the `fact` and answer concisely,\n"
" citing each fact verbatim.\n"
" 4. End every response with the total amount spent in USD — $0.01 per\n"
" successful call.\n"
)
def _ensure_vended_log_delivery(manager_arn: str, region: str) -> None:
"""Idempotently wire CloudWatch Logs vended delivery for a PaymentManager.
Three control-plane ops, each a no-op on re-run:
1. ``CreateLogGroup`` — destination Log Group, if missing.
2. ``PutDeliverySource`` — Payments → logs pipe.
3. ``PutDeliveryDestination`` — target the Log Group.
4. ``CreateDelivery`` — bind source to destination.
Authorization for the Manager to vend logs is granted by the IAM
permissions
``bedrock-agentcore:PaymentsAllowVendedLogDeliveryForResource`` and
``bedrock-agentcore:AllowVendedLogDeliveryForResource`` on the
calling principal (already attached to the agent runtime's
execution role in the CDK stack — CloudWatch checks both as a
product-level + service-level gate). There is no SDK call to "arm"
vended delivery; CloudWatch authorizes both implicitly when
``put_delivery_source`` runs against a Payment Manager ARN.
See ``docs.aws.amazon.com/AmazonCloudWatch/latest/logs/AWS-logs-infrastructure-V2-service-specific.html``.
Any ``ConflictException`` / already-exists shape is swallowed so this
can run on every Manager the agent sees without side effects.
"""
if not ENABLE_VENDED_LOG_DELIVERY or not manager_arn:
return
if manager_arn in _VENDED_LOG_DELIVERY_CONFIGURED:
return
# Derive a stable, Manager-scoped log group name from the Manager ID so
# re-runs of the same Manager hit the same log group instead of creating
# duplicates. The Manager ID is the last path segment of the ARN.
manager_id = manager_arn.rsplit("/", 1)[-1]
log_group_name = f"/bedrock-agentcore/payments/{manager_id}"
source_name = f"pay-for-api-payments-src-{manager_id}"
destination_name = f"pay-for-api-payments-dest-{manager_id}"
logs_client = boto3.client("logs", region_name=region)
# STS account lookup so we can construct the destination ARN below.
account_id = boto3.client("sts", region_name=region).get_caller_identity()["Account"]
destination_arn = f"arn:aws:logs:{region}:{account_id}:delivery-destination:{destination_name}"
log_group_arn = f"arn:aws:logs:{region}:{account_id}:log-group:{log_group_name}"
def _swallow(code_set: set[str], fn, **kwargs):
"""Call fn(**kwargs); swallow the given error codes."""
try:
return fn(**kwargs)
except botocore.exceptions.ClientError as exc:
err_code = exc.response["Error"].get("Code", "")
if err_code in code_set:
return None
raise
# 1. Ensure the log group exists before we point a delivery at it.
_swallow(
{"ResourceAlreadyExistsException"},
logs_client.create_log_group,
logGroupName=log_group_name,
)
# 2. Delivery source — Payments resource emits APPLICATION_LOGS.
# CloudWatch validates the caller's
# bedrock-agentcore:PaymentsAllowVendedLogDeliveryForResource and
# bedrock-agentcore:AllowVendedLogDeliveryForResource permissions
# against the resourceArn at this point. Without either, this call
# returns AccessDeniedException.
_swallow(
{"ConflictException", "ResourceAlreadyExistsException"},
logs_client.put_delivery_source,
name=source_name,
resourceArn=manager_arn,
logType="APPLICATION_LOGS",
)
# 3. Delivery destination — Log Group we just ensured.
_swallow(
{"ConflictException", "ResourceAlreadyExistsException"},
logs_client.put_delivery_destination,
name=destination_name,
deliveryDestinationConfiguration={
"destinationResourceArn": log_group_arn,
},
)
# 4. Bind source to destination. CreateDelivery is idempotent on the
# (source, destination) pair — returns ConflictException on re-runs.
_swallow(
{"ConflictException", "ResourceAlreadyExistsException"},
logs_client.create_delivery,
deliverySourceName=source_name,
deliveryDestinationArn=destination_arn,
)
_VENDED_LOG_DELIVERY_CONFIGURED.add(manager_arn)
logger.info(
"Vended log delivery ensured for Manager %s%s",
manager_id,
log_group_name,
)
def _build_agent(payment_config: dict | None):
"""Construct a Strands Agent with one http_request tool and — if payment
context is provided — the AgentCorePaymentsPlugin for automatic x402
handling.
``payment_config`` keys:
- manager_arn, instrument_id, session_id, payment_user_id, region
"""
from strands import Agent
from strands.models.bedrock import BedrockModel
from strands_tools import http_request
model = BedrockModel(
model_id=MODEL_ID,
region_name=AWS_REGION,
temperature=0.7,
)
# ── AgentCoreMemorySessionManager ──
# Memory is keyed on (memory_id, actor_id, session_id). We use the
# vendor-assigned paymentUserId as actor so all of a user's
# invocations roll up under one actor regardless of which notebook
# kernel or process is driving the runtime. If memory is unavailable
# (bad SDK version, missing resource, etc.) we log and continue
# without — the plugin still works.
session_manager = None
actor_id = (payment_config or {}).get("payment_user_id") or ""
if MEMORY_ID and actor_id:
try:
import uuid as _uuid
from bedrock_agentcore.memory.integrations.strands.config import (
AgentCoreMemoryConfig,
)
from bedrock_agentcore.memory.integrations.strands.session_manager import (
AgentCoreMemorySessionManager,
)
session_id = f"{actor_id}-{_uuid.uuid4().hex[:8]}"
memory_config = AgentCoreMemoryConfig(
memory_id=MEMORY_ID,
session_id=session_id,
actor_id=actor_id,
)
session_manager = AgentCoreMemorySessionManager(
agentcore_memory_config=memory_config,
region_name=AWS_REGION,
)
logger.info(
"AgentCoreMemorySessionManager attached memory=%s actor=%s session=%s",
MEMORY_ID,
actor_id,
session_id,
)
except Exception as exc: # noqa: BLE001
logger.warning(
"Memory session manager unavailable, continuing without: %s",
exc,
)
plugins: list = []
if ENABLE_PAYMENTS_PLUGIN and payment_config:
missing = [
k for k in ("manager_arn", "instrument_id", "session_id", "payment_user_id") if not payment_config.get(k)
]
if missing:
logger.info(
"AgentCorePaymentsPlugin skipped — missing fields on this invocation: %s",
missing,
)
else:
try:
from bedrock_agentcore.payments.integrations.config import (
AgentCorePaymentsPluginConfig,
)
from bedrock_agentcore.payments.integrations.strands.plugin import (
AgentCorePaymentsPlugin,
)
plugin_cfg = AgentCorePaymentsPluginConfig(
payment_manager_arn=payment_config["manager_arn"],
user_id=payment_config["payment_user_id"],
payment_instrument_id=payment_config["instrument_id"],
payment_session_id=payment_config["session_id"],
region=payment_config.get("region") or AWS_REGION,
agent_name="pay-for-api-agent",
network_preferences_config=payment_config.get("network_preferences"),
)
plugins.append(AgentCorePaymentsPlugin(config=plugin_cfg))
logger.info(
"AgentCorePaymentsPlugin attached — manager=%s instrument=%s session=%s user=%s",
payment_config["manager_arn"],
payment_config["instrument_id"],
payment_config["session_id"],
payment_config["payment_user_id"],
)
except Exception as exc: # noqa: BLE001 — plugin optional at edit time
logger.warning(
"AgentCorePaymentsPlugin init failed, continuing without: %s",
exc,
)
kwargs: dict = {
"model": model,
"tools": [http_request],
"system_prompt": SYSTEM_PROMPT,
}
if plugins:
kwargs["plugins"] = plugins
if session_manager is not None:
kwargs["session_manager"] = session_manager
# Wrap agent construction in a try/retry so a corrupt memory session
# doesn't break the invocation. On failure we drop memory and retry
# with a fresh agent — the plugin still pays.
try:
return Agent(**kwargs)
except Exception as exc: # noqa: BLE001
if session_manager is None:
raise
logger.warning(
"Agent init with memory failed (%s) — retrying without memory",
exc,
)
kwargs.pop("session_manager", None)
return Agent(**kwargs)
# ── FastAPI app ──
app = FastAPI(title="Pay for API — Buyer Agent", version="1.0.0")
@app.get("/ping")
async def ping():
return JSONResponse(content={"status": "ok"}, status_code=200)
# Maximum prompt length the /invocations endpoint accepts. Bounds the
# Bedrock token bill on each invoke and prevents a flood of multi-MB
# prompts from filling the runtime memory. Tune up if your use case
# legitimately needs longer prompts; this is a defensive cap, not a
# product constraint.
MAX_PROMPT_LEN = 5000
@app.post("/invocations")
async def invocations(request: fastapi.Request):
try:
data = await request.json()
except Exception as exc: # noqa: BLE001
logger.warning("Invalid JSON body on /invocations: %s", exc)
return JSONResponse(
content={"error": "Invalid JSON body"},
status_code=400,
)
prompt = data.get("prompt") or data.get("text") or data.get("message") or ""
seller_url = data.get("sellerUrl") or data.get("seller_url")
if not prompt:
return JSONResponse(content={"error": "prompt is required"}, status_code=400)
if not seller_url:
return JSONResponse(content={"error": "sellerUrl is required"}, status_code=400)
# ── Defensive input validation ──
# The prompt is forwarded to Bedrock and the seller URL is fetched
# by the http_request tool. Bounding both keeps the runtime
# behaving on hostile input even though AgentCore Runtime fronts
# this endpoint with its own auth + payload validation.
if len(prompt) > MAX_PROMPT_LEN:
return JSONResponse(
content={"error": f"prompt exceeds {MAX_PROMPT_LEN} characters"},
status_code=400,
)
if not (seller_url.startswith("https://") or seller_url.startswith("http://")):
return JSONResponse(
content={"error": "sellerUrl must be an http(s) URL"},
status_code=400,
)
payment_config = {
"manager_arn": data.get("managerArn") or data.get("manager_arn", ""),
"instrument_id": data.get("instrumentId") or data.get("instrument_id", ""),
"session_id": data.get("sessionId") or data.get("session_id", ""),
"payment_user_id": data.get("paymentUserId") or data.get("payment_user_id", ""),
"region": data.get("region", AWS_REGION),
"network_preferences": (data.get("networkPreferences") or data.get("network_preferences")),
}
# Wire up vended log delivery the first time we see a Manager — no-op
# thereafter for the same Manager in the same process. Any errors are
# logged but do not fail the invocation, since observability is a
# best-effort add-on.
try:
_ensure_vended_log_delivery(
manager_arn=payment_config["manager_arn"],
region=payment_config["region"],
)
except Exception as exc: # noqa: BLE001
logger.warning("Vended log delivery setup failed, continuing: %s", exc)
# We prepend the seller URL to the user prompt so the agent knows the
# exact URL to GET. Keeping it out of the system prompt lets the
# notebook point the same agent at different sellers without rebuild.
enriched_prompt = f"Seller URL: {seller_url.rstrip('/')}/facts\n\n{prompt}"
try:
agent = _build_agent(payment_config=payment_config)
result = agent(enriched_prompt)
return JSONResponse(content={"response": str(result)})
except Exception as exc: # noqa: BLE001
logger.error("Invocation error: %s", exc, exc_info=True)
return JSONResponse(
content={"error": "Agent invocation failed. See runtime logs for details."},
status_code=500,
)
if __name__ == "__main__":
port = int(os.environ.get("PORT", "8080"))
# AgentCore Runtime routes traffic into the container on all
# interfaces, so bind to 0.0.0.0 inside the container by default.
# Override with HOST=127.0.0.1 when running the container directly on
# a developer machine.
host = os.environ.get("HOST", "0.0.0.0") # nosec B104 — required by AgentCore Runtime
logger.info("Starting pay-for-api agent on %s:%s", host, port)
uvicorn.run(app, host=host, port=port, log_level="info")
@@ -0,0 +1,27 @@
# Pay for API — buyer agent container dependencies.
# Core web framework
fastapi>=0.115.0
uvicorn[standard]>=0.30.0
# AWS SDK
# Note: AgentCore Payments observability uses the standard CloudWatch
# vended-logs pattern. There is no dedicated SDK call to "arm" a
# Manager — CloudWatch authorizes the
# bedrock-agentcore:AllowVendedLogDeliveryForResource IAM permission
# implicitly when the agent calls logs.put_delivery_source against the
# Manager ARN. Any boto3 release with the standard logs client works.
boto3>=1.42.0
botocore>=1.42.0
# AgentCore SDK — provides AgentCorePaymentsPlugin
bedrock-agentcore>=1.9.0
# Strands Agents framework + http_request tool
strands-agents>=1.39.0
strands-agents-tools>=0.5.0
# AWS Distro for OpenTelemetry — auto-instruments FastAPI + boto3 so
# traces and logs flow to CloudWatch via the ADOT collector that
# AgentCore Runtime injects into the container.
aws-opentelemetry-distro>=0.10.1
@@ -0,0 +1,147 @@
# ══════════════════════════════════════════════════════════════════════
# AgentCore Payments — Pay for API (Fun Facts)
#
# Copy this file to `.env` and fill in values. The notebook's §2 seeds
# `.env` from this template on first run and lists the keys you still
# need to fill in.
#
# Or copy manually and edit by hand:
# cp env-sample.txt .env
#
# Every env var the notebook, seller deploy script, and agent container
# read is here. Empty values are filled in as you go:
# * `setup-roles.sh` writes the four IAM role ARNs.
# * `deploy-seller.sh` writes `SELLER_API_URL`.
# * §4 of the notebook writes Manager, Connector, Credential Provider,
# and Payment Instrument IDs.
# * §5 writes the two Session IDs.
#
# ── Wallet provider secret handoff ──
# Wallet provider secrets (Coinbase / Privy keys, Privy authorization
# private key) are pasted into this file once. The notebook's §4 reads
# them to call `CreatePaymentCredentialProvider`, which hands them to
# AgentCore Identity. The service stores them in AWS Secrets Manager
# under AWS KMS encryption and surfaces only the secret ARN to your
# agent. The agent runtime calls `GetResourcePaymentToken` at signing
# time and never receives the raw secret.
#
# After §4 runs, the secrets in this file are no longer used at runtime.
# `.env` is `.gitignore`-d so it is never committed; you can blank the
# secret values once setup is complete and re-running §4 from a fresh
# .env will recreate the credential provider with whatever you paste.
# See "Configure credential provider" in the AgentCore Payments docs.
# ══════════════════════════════════════════════════════════════════════
# ── AWS ──────────────────────────────────────────────────────────────
# AgentCore Payments is available in preview in these regions:
# us-east-1 — US East (N. Virginia)
# us-west-2 — US West (Oregon)
# eu-central-1 — Europe (Frankfurt)
# ap-southeast-2 — Asia Pacific (Sydney)
# boto3 derives the control-plane and data-plane endpoints from the
# region, so no endpoint URLs are needed here.
AWS_REGION=us-west-2
# AWS_PROFILE=default # Uncomment to use a named AWS CLI profile
# ── IAM roles — created by test/integration/setup-roles.sh ───────────
# setup-roles.sh creates four roles idempotently. The notebook assumes
# into each one for the operation that role is allowed to perform.
# Run it once before the notebook — it writes the four ARNs below
# back into this .env automatically:
# bash test/integration/setup-roles.sh
CONTROL_PLANE_ROLE_ARN=arn:aws:iam::<ACCOUNT_ID>:role/AgentCorePaymentsControlPlaneRole
MANAGEMENT_ROLE_ARN=arn:aws:iam::<ACCOUNT_ID>:role/AgentCorePaymentsManagementRole
PROCESS_PAYMENT_ROLE_ARN=arn:aws:iam::<ACCOUNT_ID>:role/AgentCorePaymentsProcessPaymentRole
RESOURCE_RETRIEVAL_ROLE_ARN=arn:aws:iam::<ACCOUNT_ID>:role/AgentCorePaymentsResourceRetrievalRole
# ── Wallet provider credentials — both providers ─────────────────────
# The notebook wires up BOTH providers side-by-side under a single
# Manager. Fill in both sets. §4.2 creates the Credential Providers,
# §4.4 creates the Connectors, §4.5 creates two Instruments per
# provider (EVM + Solana = four wallets total).
# Coinbase CDP — coinbase.com/developer-platform → Project → API Keys
# and Project → Wallet → Wallet Secret. Enable "Delegated signing"
# under Project → Wallet → Embedded Wallets → Policies before first use.
COINBASE_API_KEY_ID=
COINBASE_API_KEY_SECRET=
COINBASE_WALLET_SECRET=
# Stripe-via-Privy — Privy dashboard → App → API Keys and
# App → Authorization Keys. The private key is returned prefixed with
# `wallet-auth:` — strip the prefix before pasting below.
PRIVY_APP_ID=
PRIVY_APP_SECRET=
PRIVY_AUTHORIZATION_ID=
PRIVY_AUTHORIZATION_PRIVATE_KEY=
# ── AgentCore Payments state — populated by the notebook ────────────
# Leave empty on first run. §4 creates one Manager, two Credential
# Providers (CDP + Privy), two Connectors, and four Instruments.
# §5 creates two Sessions (one per provider).
MANAGER_ID=
MANAGER_ARN=
CRED_PROVIDER_NAME_CDP=
CREDENTIAL_PROVIDER_ARN_CDP=
CRED_PROVIDER_NAME_PRIVY=
CREDENTIAL_PROVIDER_ARN_PRIVY=
CONNECTOR_ID_CDP=
CONNECTOR_ID_PRIVY=
PAYMENT_INSTRUMENT_ID_CDP_EVM=
WALLET_ADDRESS_CDP_EVM=
PAYMENT_INSTRUMENT_ID_CDP_SOL=
WALLET_ADDRESS_CDP_SOL=
PAYMENT_INSTRUMENT_ID_PRIVY_EVM=
WALLET_ADDRESS_PRIVY_EVM=
PAYMENT_INSTRUMENT_ID_PRIVY_SOL=
WALLET_ADDRESS_PRIVY_SOL=
SESSION_ID_CDP=
SESSION_ID_PRIVY=
# ── Payment session tuning ───────────────────────────────────────────
# Budget the operator authorises per session. USD amount.
SESSION_MAX_SPEND=1.00
# 15480 minutes; enforced by the service.
SESSION_EXPIRY_MINUTES=120
# Informational — used as the `userId` header on CreatePaymentSession.
# Note: for hub-provisioned Coinbase instruments the service assigns its
# own CDP end-user UUID and ignores what we send; §6 reads the
# authoritative id back from `paymentInstrument.userId`.
USER_ID=
# Email the embedded wallets are linked to. CreatePaymentInstrument
# requires `linkedAccounts[{email:{emailAddress}}]` — the service uses
# it to resolve or create the vendor-side end-user that owns the
# wallet. Prefer a real inbox you control so Coinbase Wallet Hub and
# Privy can send verification mail for signing-delegation grants.
# Format: a real email address you can read. Example: alex@example.com
INSTRUMENT_EMAIL=
# ── Seller (Fun Facts API) ───────────────────────────────────────────
# Populated by deploy-seller.sh after `cdk deploy` prints SellerApiUrl.
SELLER_API_URL=
# Seller payout wallets. Used by deploy-seller.sh at seller deploy time —
# not by the notebook agent. Set at least the EVM wallet; Solana is
# optional but required for the §7 SOL run.
# EVM format: 0x-prefixed 40-character hex.
# Example: 0x1234567890abcdef1234567890abcdef12345678
# Solana format: base58, 32-44 characters.
# Example: 5Yb8Zr9T3kQv2Xm4Np1Wq6Rs7Ht8Ju9Kv0Lp1Mq2Nr3
SELLER_WALLET_ADDRESS=
SELLER_SOLANA_WALLET_ADDRESS=
# Seller tuning (optional). Leave commented for the defaults.
# X402_FACILITATOR_URL=https://x402.org/facilitator
# X402_PRICE=$0.01
# ── Agent container (optional overrides) ─────────────────────────────
# Only consumed inside the AgentCore Runtime container in §8. Ignored
# elsewhere. The CDK stack already defaults these — override here only
# if you want a different model or want to disable the plugin at runtime.
# MODEL_ID=us.anthropic.claude-sonnet-4-5-20250929-v1:0
# ENABLE_PAYMENTS_PLUGIN=1
Binary file not shown.

After

Width:  |  Height:  |  Size: 489 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 304 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 57 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 246 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 81 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 36 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 66 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 284 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 222 KiB

@@ -0,0 +1,7 @@
strands-agents>=1.39.0
strands-agents-tools>=0.5.0
bedrock-agentcore>=1.9.0
boto3>=1.42.0
botocore>=1.42.0
requests>=2.32.0
python-dotenv>=1.0.0
@@ -0,0 +1,27 @@
#!/usr/bin/env python3
"""CDK app entry point for the Pay for API — Fun Facts seller stack."""
import os
import aws_cdk as cdk
from seller_stack import AgentCorePaymentsFunFactsSellerStack
app = cdk.App()
# Region comes from the usual CDK resolution order:
# CDK_DEFAULT_REGION → AWS_REGION → AWS CLI profile region.
# We default to us-west-2 to match the default AgentCore Payments region.
env = cdk.Environment(
account=os.environ.get("CDK_DEFAULT_ACCOUNT"),
region=os.environ.get("CDK_DEFAULT_REGION", os.environ.get("AWS_REGION", "us-west-2")),
)
AgentCorePaymentsFunFactsSellerStack(
app,
"AgentCorePaymentsFunFactsSellerStack",
env=env,
description="AgentCore Payments sample — Fun Facts x402 seller (pay per API call)",
)
app.synth()
@@ -0,0 +1,20 @@
{
"app": "python3 app.py",
"watch": {
"include": ["**"],
"exclude": [
"README.md",
"cdk*.json",
"requirements*.txt",
"source.bat",
"**/__init__.py",
"**/__pycache__",
"tests"
]
},
"context": {
"@aws-cdk/aws-apigateway:usagePlanKeyOrderInsensitiveId": true,
"@aws-cdk/aws-lambda:recognizeVersionProps": true,
"@aws-cdk/aws-lambda:recognizeLayerVersion": true
}
}
@@ -0,0 +1,2 @@
aws-cdk-lib>=2.140.0
constructs>=10.3.0,<11.0.0
@@ -0,0 +1,153 @@
"""Fun Facts seller CDK stack.
Mirrors the agentcore-payments house-standard seller pattern
(backend/lambdas/sellers/crypto-price):
- Node.js 20 ARM64 AWS Lambda function with pre-installed
``node_modules`` packaged into the asset (the deploy script runs
``npm install`` before ``cdk deploy``).
- Two env vars for payout — ``SELLER_WALLET_ADDRESS`` (EVM / Base
Sepolia) and ``SELLER_SOLANA_WALLET_ADDRESS`` (Solana / Devnet). Both
are forwarded by the x402 seller library into the ``accepts`` array on
each 402 response. Set one, both, or neither; the Lambda emits one
``accepts`` entry per configured network.
- ``X402_FACILITATOR_URL`` — override to point at a private facilitator.
Defaults to the public x402.org facilitator.
- One route: ``GET /facts`` behind the x402 payment middleware, plus
public ``GET /`` and ``GET /health`` for sanity checks.
"""
from __future__ import annotations
import os
from pathlib import Path
from aws_cdk import (
CfnOutput,
Duration,
Stack,
)
from aws_cdk import aws_apigatewayv2 as apigwv2
from aws_cdk import aws_apigatewayv2_integrations as apigwv2_integrations
from aws_cdk import aws_lambda as _lambda
from aws_cdk import aws_logs as logs
from constructs import Construct
LAMBDA_CODE_DIR = str(Path(__file__).resolve().parent.parent / "lambda")
class AgentCorePaymentsFunFactsSellerStack(Stack):
"""A minimal x402 seller: HTTP API → Node.js Lambda."""
def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
super().__init__(scope, construct_id, **kwargs)
# ── Seller config ────────────────────────────────────────────
# Override via CDK context (`cdk deploy -c seller_wallet=0x…`) or
# environment variables at deploy time. Both networks are optional;
# if neither is set the Lambda still runs but the facilitator will
# reject payment proofs. Set at least one.
#
# Defaults to "WALLET_NOT_CONFIGURED" (mirrors seller/lambda/index.js)
# so an unset wallet shows up as a clearly invalid placeholder
# rather than an empty string.
evm_wallet = (
self.node.try_get_context("seller_wallet")
or os.environ.get("SELLER_WALLET_ADDRESS")
or "WALLET_NOT_CONFIGURED"
)
solana_wallet = (
self.node.try_get_context("seller_solana_wallet")
or os.environ.get("SELLER_SOLANA_WALLET_ADDRESS")
or "WALLET_NOT_CONFIGURED"
)
facilitator_url = os.environ.get("X402_FACILITATOR_URL") or "https://x402.org/facilitator"
price = os.environ.get("X402_PRICE") or "$0.01"
# ── Lambda function ──────────────────────────────────────────
seller_fn = _lambda.Function(
self,
"SellerFunction",
runtime=_lambda.Runtime.NODEJS_20_X,
architecture=_lambda.Architecture.ARM_64,
handler="index.handler",
# The deploy script runs `npm install` in the lambda/ folder
# before `cdk deploy` so the asset ships node_modules inline.
# Matches the pattern used by agentcore-payments sellers.
code=_lambda.Code.from_asset(LAMBDA_CODE_DIR),
timeout=Duration.seconds(30),
memory_size=256,
environment={
"SELLER_WALLET_ADDRESS": evm_wallet,
"SELLER_SOLANA_WALLET_ADDRESS": solana_wallet,
"X402_FACILITATOR_URL": facilitator_url,
"X402_PRICE": price,
},
log_retention=logs.RetentionDays.ONE_WEEK,
description="Fun Facts x402 seller — AgentCore Payments use case",
)
# ── HTTP API ─────────────────────────────────────────────────
# CORS is wide-open for the demo so the seller is reachable from
# any caller (the AgentCore Runtime container, a browser-based
# debugger, a curl session). For production, restrict origins to
# the specific agent runtime endpoints that need to call this
# seller, and limit methods to GET + OPTIONS.
http_api = apigwv2.HttpApi(
self,
"SellerHttpApi",
api_name="pay-for-api-fun-facts",
description="Fun Facts x402 seller — pay-per-fact via x402",
cors_preflight=apigwv2.CorsPreflightOptions(
# Demo configuration — restrict to specific origins in
# production (for example, your agent runtime domains).
allow_origins=["*"],
allow_methods=[apigwv2.CorsHttpMethod.ANY],
allow_headers=["*"],
),
)
integration = apigwv2_integrations.HttpLambdaIntegration(
"SellerLambdaIntegration",
handler=seller_fn,
)
# Single proxy route catches GET /, GET /facts, GET /health.
http_api.add_routes(
path="/{proxy+}",
methods=[apigwv2.HttpMethod.ANY],
integration=integration,
)
http_api.add_routes(
path="/",
methods=[apigwv2.HttpMethod.ANY],
integration=integration,
)
# ── Outputs ──────────────────────────────────────────────────
CfnOutput(
self,
"SellerApiUrl",
value=http_api.api_endpoint,
description="Invoke URL for the Fun Facts x402 seller API",
)
CfnOutput(
self,
"SellerEvmWallet",
value=evm_wallet or "(unset)",
description=(
"EVM (Base Sepolia) wallet that receives USDC for paid "
"requests. Set via `cdk deploy -c seller_wallet=0x…` or "
"the SELLER_WALLET_ADDRESS env var."
),
)
CfnOutput(
self,
"SellerSolanaWallet",
value=solana_wallet or "(unset)",
description=(
"Solana (Devnet) wallet that receives USDC for paid "
"requests. Set via `cdk deploy -c seller_solana_wallet=…` "
"or the SELLER_SOLANA_WALLET_ADDRESS env var."
),
)
@@ -0,0 +1,2 @@
node_modules/
package-lock.json
@@ -0,0 +1,273 @@
/**
* Fun Facts x402 seller — Node.js AWS Lambda function.
*
* Mirrors the house-standard pattern used by agentcore-payments sellers
* (see backend/lambdas/sellers/crypto-price):
*
* - `@x402/hono` `paymentMiddlewareFromHTTPServer` does the full 402 +
* facilitator verify/settle handshake for us — no manual base64 header
* assembly, no manual /verify /settle HTTP calls.
* - Chain Agnostic Improvement Proposal 2 (CAIP-2) network identifiers
* (`eip155:84532` for Base Sepolia, `solana:…` for Devnet) — this is
* what the AgentCore Payments plugin emits on the wire when it signs
* an x402 payload, not the short `base-sepolia` / `solana-devnet`
* strings.
* - Price expressed as human-readable USD (`"$0.01"`) — the x402
* middleware converts to on-chain atomic amounts.
* - Response shape: `{ x402_content, x402_meta }` — the bazaar-friendly
* schema the AgentCore Registry can index.
* - `declareDiscoveryExtension` so this seller is discoverable through
* the Bazaar Model Context Protocol (MCP).
*
* Multi-network: when both `SELLER_WALLET_ADDRESS` (EVM) and
* `SELLER_SOLANA_WALLET_ADDRESS` are set, both `accepts` entries are
* emitted and the agent picks whichever network its instrument is on.
*/
import { Hono } from "hono";
import { handle } from "hono/aws-lambda";
import {
paymentMiddlewareFromHTTPServer,
x402HTTPResourceServer,
x402ResourceServer,
} from "@x402/hono";
import { HTTPFacilitatorClient } from "@x402/core/server";
import { registerExactEvmScheme } from "@x402/evm/exact/server";
// SVM = Solana Virtual Machine — the on-chain runtime Solana programs
// execute under. The x402 SVM scheme builds + verifies SPL-token
// transfer transactions on Solana.
import { registerExactSvmScheme } from "@x402/svm/exact/server";
import {
bazaarResourceServerExtension,
declareDiscoveryExtension,
} from "@x402/extensions/bazaar";
// ── Config (from Lambda env vars) ───────────────────────────────────────
// Wallet addresses default to "WALLET_NOT_CONFIGURED" so an unconfigured
// seller emits clearly invalid placeholders in the 402 response. The
// facilitator rejects them at settlement and the agent surfaces a
// helpful error pointing the operator at SELLER_WALLET_ADDRESS /
// SELLER_SOLANA_WALLET_ADDRESS in `.env`.
const X402_CONFIG = {
facilitatorUrl:
process.env.X402_FACILITATOR_URL || "https://x402.org/facilitator",
// CAIP-2 network identifiers
evmNetwork: "eip155:84532", // Base Sepolia
evmPayTo: process.env.SELLER_WALLET_ADDRESS || "WALLET_NOT_CONFIGURED",
solanaNetwork: "solana:EtWTRABZaYq6iMfeYKouRu166VU2xqa1", // Devnet
solanaPayTo:
process.env.SELLER_SOLANA_WALLET_ADDRESS || "WALLET_NOT_CONFIGURED",
};
const PRICE = process.env.X402_PRICE || "$0.01";
// ── Fun Facts data ──────────────────────────────────────────────────────
const FACTS = {
space: [
"A day on Venus is longer than its year — it takes 243 Earth days to rotate but only 225 days to orbit the sun.",
"Neutron stars are so dense that a sugar-cube-sized sample would weigh about 1 billion tons on Earth.",
"The largest known volcano in the solar system, Olympus Mons on Mars, is nearly three times taller than Mount Everest.",
"There is a planet made largely of diamond — 55 Cancri e, about 40 light-years away.",
"Saturn's density is so low that, hypothetically, it would float in a bathtub of water large enough to hold it.",
],
oceans: [
"More than 80 percent of the ocean has never been mapped, explored, or even seen by humans.",
"The Mariana Trench reaches nearly 11,000 meters deep — taller than Mount Everest turned upside down.",
"Hydrothermal vents on the ocean floor support ecosystems that never see sunlight.",
"Blue whales' hearts are so large that a human could swim through their arteries.",
"Plankton in the ocean produce more than half of the oxygen we breathe.",
],
ai: [
"The term 'artificial intelligence' was coined at the Dartmouth Workshop in 1956.",
"Transformer architectures, introduced in 2017, underpin nearly every modern large language model.",
"Reinforcement learning from human feedback (RLHF) is what made instruction-following LLMs practical.",
"Chess AI definitively surpassed human world champions in 1997 with IBM's Deep Blue.",
"Modern LLMs are trained on tokens measured in the trillions.",
],
payments: [
"The x402 protocol revives an HTTP status code — 402 Payment Required — that was reserved in RFC 7231 but never standardized.",
"Stablecoins like USDC settle on-chain in seconds, versus days for traditional wire transfers.",
"Micropayments were first proposed by Ted Nelson in the 1960s as part of his Project Xanadu vision.",
"Account abstraction on Ethereum makes gasless agent payments possible via meta-transactions.",
"The first cryptocurrency micropayment channel was demonstrated in 2013 by Meni Rosenfeld and Peter Todd.",
],
default: [
"Honey found in Egyptian tombs is still edible — honey does not spoil.",
"Octopuses have three hearts and blue blood.",
"Bananas are berries, but strawberries are not.",
"The Eiffel Tower can grow more than 15 cm taller in summer due to thermal expansion.",
"Wombat droppings are cube-shaped.",
],
};
const SUPPORTED_TOPICS = Object.keys(FACTS).filter((k) => k !== "default");
function pickFact(rawTopic) {
const key = String(rawTopic || "").trim().toLowerCase();
const resolved = FACTS[key] ? key : "default";
const pool = FACTS[resolved];
return { topic: resolved, fact: pool[Math.floor(Math.random() * pool.length)] };
}
function buildAccepts(price) {
const NOT_CONFIGURED = "WALLET_NOT_CONFIGURED";
const accepts = [];
// Treat the placeholder the same as an unset env var: still emit the
// accepts entry so the 402 response has the right shape, but the
// facilitator will reject any payment proof at settlement and the
// agent surfaces a clear error message.
if (X402_CONFIG.evmPayTo && X402_CONFIG.evmPayTo !== NOT_CONFIGURED) {
accepts.push({
scheme: "exact",
price,
network: X402_CONFIG.evmNetwork,
payTo: X402_CONFIG.evmPayTo,
});
}
if (X402_CONFIG.solanaPayTo && X402_CONFIG.solanaPayTo !== NOT_CONFIGURED) {
accepts.push({
scheme: "exact",
price,
network: X402_CONFIG.solanaNetwork,
payTo: X402_CONFIG.solanaPayTo,
});
}
if (!accepts.length) {
// No wallet configured — emit an EVM entry anyway so the 402 response
// has the right shape; the facilitator will reject the proof at
// settlement. Keeps the error message useful during first-run setup.
accepts.push({
scheme: "exact",
price,
network: X402_CONFIG.evmNetwork,
payTo: NOT_CONFIGURED,
});
}
return accepts;
}
// ── Hono app + x402 middleware ──────────────────────────────────────────
const app = new Hono();
// Request logging — same shape as the reference seller so CloudWatch
// queries are portable.
app.use("*", async (c, next) => {
const start = Date.now();
const sig = c.req.header("payment-signature");
console.log(
JSON.stringify({
event: "request_in",
method: c.req.method,
path: c.req.path,
hasPaymentSignature: !!sig,
paymentSignatureLength: sig?.length || 0,
})
);
await next();
console.log(
JSON.stringify({
event: "response_out",
method: c.req.method,
path: c.req.path,
status: c.res.status,
durationMs: Date.now() - start,
hasPaymentSignature: !!sig,
})
);
});
// x402 server — EVM + SVM schemes, Bazaar discovery extension.
const facilitatorClient = new HTTPFacilitatorClient({
url: X402_CONFIG.facilitatorUrl,
});
const server = new x402ResourceServer(facilitatorClient);
registerExactEvmScheme(server);
registerExactSvmScheme(server);
server.registerExtension(bazaarResourceServerExtension);
// Declare one paid route: GET /facts. The Bazaar discovery extension
// exposes the topic query-parameter schema + an example output so the
// AgentCore Registry can list this seller.
const routes = {
"GET /facts": {
accepts: buildAccepts(PRICE),
extensions: {
...declareDiscoveryExtension({
input: { topic: "space" },
inputSchema: {
properties: {
topic: {
type: "string",
description: `One of ${SUPPORTED_TOPICS.join(", ")} (or any other string for a random general fact).`,
},
},
required: [],
},
bodyType: "query",
output: {
example: {
x402_content: {
type: "text",
data: '{"topic":"space","fact":"A day on Venus is longer than its year …"}',
title: "Fun fact: space",
mime_type: "application/json",
},
x402_meta: {
seller: "pay-for-api-fun-facts",
version: "1.0",
},
},
},
}),
},
},
};
const httpServer = new x402HTTPResourceServer(server, routes);
await httpServer.initialize();
app.use(
paymentMiddlewareFromHTTPServer(httpServer, undefined, undefined, false)
);
// ── Routes ──────────────────────────────────────────────────────────────
// Paid route
app.get("/facts", (c) => {
const topic = c.req.query("topic") || "default";
const { topic: resolvedTopic, fact } = pickFact(topic);
return c.json({
x402_content: {
type: "text",
data: JSON.stringify({ topic: resolvedTopic, fact }),
title: `Fun fact: ${resolvedTopic}`,
mime_type: "application/json",
},
x402_meta: {
seller: "pay-for-api-fun-facts",
version: "1.0",
generated_at: new Date().toISOString(),
supported_topics: SUPPORTED_TOPICS,
},
});
});
// Public health check — no payment required.
app.get("/health", (c) =>
c.json({
status: "ok",
service: "pay-for-api-fun-facts",
price: PRICE,
networks: buildAccepts(PRICE).map((a) => a.network),
supported_topics: SUPPORTED_TOPICS,
})
);
// Discovery root.
app.get("/", (c) =>
c.json({
service: "pay-for-api-fun-facts",
paidEndpoints: ["GET /facts?topic=<topic>"],
price: PRICE,
})
);
export const handler = handle(app);
@@ -0,0 +1,14 @@
{
"name": "x402-seller-pay-for-api",
"version": "1.0.0",
"type": "module",
"description": "x402 Fun Facts seller — pay-for-API sample",
"dependencies": {
"@x402/hono": "^2.7.0",
"@x402/core": "^2.7.0",
"@x402/evm": "^2.7.0",
"@x402/svm": "^2.7.0",
"@x402/extensions": "^2.7.0",
"hono": "^4.0.0"
}
}
@@ -0,0 +1,32 @@
# test/integration/
Operational scripts for the **Pay-For-API** use case. Run them from the
use-case root (`02-use-cases/01-pay-for-api/`); each script resolves its
paths relative to this folder, so it does not matter which directory you
invoke them from as long as the repo layout is intact.
Mirrors the pattern used by
[`agentcore-payments/test/integration/`](../../../../agentcore-payments/test/integration).
| Script | What it does |
|--------|--------------|
| `setup-roles.sh` | Creates the four IAM roles the notebook assumes into (`ControlPlane`, `Management`, `ProcessPayment`, `ResourceRetrieval`) with the separation-of-duties policy model described in the main [README](../../README.md). Idempotent — safe to re-run. Writes the role ARNs back into `.env`. |
| `setup-env.sh` | Interactive env setup. Copies `env-sample.txt``.env` on first run, then walks through the empty values (role ARNs, Coinbase CDP credentials, seller payout wallet) and prompts only for the ones that are still blank. Re-run with `--force-reprompt` to replace already-set values. |
| `deploy-seller.sh` | `npm install` the seller Lambda's `node_modules`, then `cdk bootstrap` (first run only) and `cdk deploy` the seller stack. Writes `seller/cdk/outputs.json` and prints `SellerApiUrl`. |
| `destroy-seller.sh` | `cdk destroy --force` the seller stack. |
## Typical order
```bash
# From 02-use-cases/01-pay-for-api/
bash test/integration/setup-roles.sh # create IAM roles (once per account)
bash test/integration/setup-env.sh # prompt for Coinbase creds + other secrets
bash test/integration/deploy-seller.sh # deploy the paid API
# paste SellerApiUrl into .env as SELLER_API_URL
jupyter notebook pay-for-api.ipynb
# …work through the notebook…
bash test/integration/destroy-seller.sh # when done
```
The notebook's §3 also invokes `deploy-seller.sh` for you, so running the
script manually is optional — whichever is more comfortable.
@@ -0,0 +1,91 @@
#!/usr/bin/env bash
# Deploy the Pay for API buyer agent to AgentCore Runtime via AWS CDK.
#
# The agent container image is built in AWS CodeBuild (not on this
# machine) so no Docker install is required. `cdk deploy` uploads
# agent/container/ as an S3 asset, CodeBuild pulls it, builds + pushes
# to ECR, and the Runtime resource pulls from there on invoke.
#
# Prerequisites:
# - AWS CLI v2 configured (aws configure)
# - AWS CDK v2 installed (npm install -g aws-cdk)
# - Python 3.10+ with pip (for the CDK Python dependencies)
#
# Usage (from anywhere):
# bash test/integration/deploy-agent.sh
#
# Writes outputs to agent/cdk/outputs.json. The notebook's §8 reads that
# file to pick up the Runtime ARN.
set -euo pipefail
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
USE_CASE_ROOT="$(cd "${SCRIPT_DIR}/../.." && pwd)"
CDK_DIR="${USE_CASE_ROOT}/agent/cdk"
CONTAINER_DIR="${USE_CASE_ROOT}/agent/container"
# Pull region from .env so it matches whatever the notebook provisioned.
if [ -f "${USE_CASE_ROOT}/.env" ]; then
# Guard against unreplaced placeholders from env-sample.txt.
if grep -q "<ACCOUNT_ID>" "${USE_CASE_ROOT}/.env"; then
echo "${USE_CASE_ROOT}/.env still contains <ACCOUNT_ID> placeholders." >&2
echo " Run: bash test/integration/setup-roles.sh" >&2
echo " before deploying the agent." >&2
exit 1
fi
set -a
# shellcheck disable=SC1091
source "${USE_CASE_ROOT}/.env"
set +a
fi
REGION="${AWS_REGION:-us-west-2}"
echo "── Pay for API — Agent Deploy ─────────────────────────────"
echo "Region: ${REGION}"
echo "CDK: ${CDK_DIR}"
echo "Container: ${CONTAINER_DIR}"
echo ""
echo "The container image is built in AWS CodeBuild (no Docker needed on"
echo "this machine). First run can take 46 minutes for the build; subsequent"
echo "deploys only rebuild if agent/container/ changed."
echo ""
# ── 1. CDK Python venv ──
if [ ! -d "${CDK_DIR}/.venv" ]; then
echo "Creating Python venv for CDK..."
python3 -m venv "${CDK_DIR}/.venv"
fi
# shellcheck disable=SC1091
source "${CDK_DIR}/.venv/bin/activate"
echo "Installing CDK Python dependencies..."
pip install --quiet --upgrade pip
pip install --quiet -r "${CDK_DIR}/requirements.txt"
# ── 2. Bootstrap (idempotent) ──
ACCOUNT_ID="$(aws sts get-caller-identity --query Account --output text)"
if ! aws cloudformation describe-stacks --stack-name CDKToolkit --region "${REGION}" >/dev/null 2>&1; then
echo ""
echo "Bootstrapping CDK for ${ACCOUNT_ID}/${REGION}..."
(cd "${CDK_DIR}" && cdk bootstrap "aws://${ACCOUNT_ID}/${REGION}")
else
echo "CDK already bootstrapped for ${ACCOUNT_ID}/${REGION}."
fi
# ── 3. Deploy ──
echo ""
echo "Deploying AgentCorePaymentsBuyerAgentStack..."
echo "(CDK synth + asset upload + CodeBuild run — typically 58 min on the"
echo " first deploy, ~2 min on subsequent runs if nothing changed.)"
(cd "${CDK_DIR}" && cdk deploy --require-approval never --outputs-file ./outputs.json)
RUNTIME_ARN="$(python3 -c 'import json; print(json.load(open("'"${CDK_DIR}"'/outputs.json"))["AgentCorePaymentsBuyerAgentStack"]["AgentRuntimeArn"])')"
RUNTIME_ID="$(python3 -c 'import json; print(json.load(open("'"${CDK_DIR}"'/outputs.json"))["AgentCorePaymentsBuyerAgentStack"]["AgentRuntimeId"])')"
echo ""
echo "── Deploy Complete ─────────────────────────────────────────"
echo "✅ AgentRuntimeArn: ${RUNTIME_ARN}"
echo " AgentRuntimeId: ${RUNTIME_ID}"
echo ""
echo "The notebook §8 reads agent/cdk/outputs.json to pick up these values."
@@ -0,0 +1,153 @@
#!/usr/bin/env bash
# Deploy the Fun Facts x402 seller stack via AWS CDK.
#
# The Lambda is Node.js with pre-installed node_modules (same pattern as
# agentcore-payments sellers) so this script runs `npm install` inside
# seller/lambda/ before `cdk deploy` packages the asset.
#
# Prerequisites:
# - AWS CLI v2 configured (aws configure)
# - AWS CDK v2 installed (npm install -g aws-cdk)
# - Node.js 20+ and npm
# - Python 3.10+ with pip (for the CDK Python dependencies)
#
# Optional:
# - SELLER_WALLET_ADDRESS=0x… # EVM (Base Sepolia) payout wallet
# - SELLER_SOLANA_WALLET_ADDRESS=… # Solana (Devnet) payout wallet
# - X402_FACILITATOR_URL=… # Override facilitator (defaults to x402.org)
#
# Usage (from anywhere):
# bash test/integration/deploy-seller.sh
#
# After deploy, copy the printed SellerApiUrl into .env as SELLER_API_URL.
set -euo pipefail
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
# Scripts live at <use-case>/test/integration/ — ../../ resolves the
# use-case root, the anchor for seller/ and .env.
USE_CASE_ROOT="$(cd "${SCRIPT_DIR}/../.." && pwd)"
LAMBDA_DIR="${USE_CASE_ROOT}/seller/lambda"
CDK_DIR="${USE_CASE_ROOT}/seller/cdk"
# Pull the payout wallets + region from .env so the values the notebook
# prompted for in §2 flow through to the CDK deploy. Shell-env vars
# already set on the current session take precedence.
if [ -f "${USE_CASE_ROOT}/.env" ]; then
# Guard against unreplaced placeholders like "<ACCOUNT_ID>" — bash
# would try to interpret `<ACCOUNT_ID>` as a redirection and error
# out with "No such file or directory" when sourcing. Tell the user
# cleanly what went wrong instead.
if grep -q "<ACCOUNT_ID>" "${USE_CASE_ROOT}/.env"; then
echo "${USE_CASE_ROOT}/.env still contains <ACCOUNT_ID> placeholders." >&2
echo " Run: bash test/integration/setup-roles.sh" >&2
echo " (or re-run §2 in the notebook) before deploying." >&2
exit 1
fi
set -a
# shellcheck disable=SC1091
source "${USE_CASE_ROOT}/.env"
set +a
fi
REGION="${AWS_REGION:-us-west-2}"
echo "── Pay for API — Seller Deploy ────────────────────────────"
echo "Region: ${REGION}"
echo "Lambda: ${LAMBDA_DIR}"
echo "CDK: ${CDK_DIR}"
echo ""
# ── 0. Wallet sanity check ──
warn=()
if [ -z "${SELLER_WALLET_ADDRESS:-}" ]; then
warn+=(" • SELLER_WALLET_ADDRESS (EVM) — required for Base Sepolia payments")
fi
if [ -z "${SELLER_SOLANA_WALLET_ADDRESS:-}" ]; then
warn+=(" • SELLER_SOLANA_WALLET_ADDRESS (Solana) — required for Solana Devnet payments")
fi
if [ ${#warn[@]} -gt 0 ]; then
echo "⚠️ One or more payout wallets are not set:"
for line in "${warn[@]}"; do
echo "${line}"
done
echo ""
echo " Without a payout wallet for a given network the seller emits an"
echo " invalid 402 for that network and the agent cannot pay on it."
echo " At minimum you need SELLER_WALLET_ADDRESS for the §8 EVM run."
echo ""
echo " Set the missing ones in .env and re-run this script, e.g.:"
echo " export SELLER_WALLET_ADDRESS=0xYourBaseSepoliaAddress"
echo " export SELLER_SOLANA_WALLET_ADDRESS=YourSolanaDevnetAddress"
echo ""
read -r -p " Continue anyway? [y/N] " ok
case "${ok}" in
y|Y|yes|YES) ;;
*) echo " Aborted."; exit 1 ;;
esac
echo ""
fi
# ── 1. Install Lambda node_modules ──
echo "Installing Lambda node_modules..."
(cd "${LAMBDA_DIR}" && npm install --silent --omit=dev)
# ── 2. CDK Python venv ──
if [ ! -d "${CDK_DIR}/.venv" ]; then
echo "Creating Python venv for CDK..."
python3 -m venv "${CDK_DIR}/.venv"
fi
# shellcheck disable=SC1091
source "${CDK_DIR}/.venv/bin/activate"
echo "Installing CDK Python dependencies..."
pip install --quiet --upgrade pip
pip install --quiet -r "${CDK_DIR}/requirements.txt"
# ── 3. Bootstrap (idempotent) ──
ACCOUNT_ID="$(aws sts get-caller-identity --query Account --output text)"
if ! aws cloudformation describe-stacks --stack-name CDKToolkit --region "${REGION}" >/dev/null 2>&1; then
echo ""
echo "Bootstrapping CDK for ${ACCOUNT_ID}/${REGION}..."
(cd "${CDK_DIR}" && cdk bootstrap "aws://${ACCOUNT_ID}/${REGION}")
fi
# ── 4. Deploy ──
echo ""
echo "Deploying AgentCorePaymentsFunFactsSellerStack..."
(cd "${CDK_DIR}" && cdk deploy --require-approval never --outputs-file ./outputs.json)
API_URL="$(python3 -c 'import json; print(json.load(open("'"${CDK_DIR}"'/outputs.json"))["AgentCorePaymentsFunFactsSellerStack"]["SellerApiUrl"])')"
EVM_WALLET="$(python3 -c 'import json; print(json.load(open("'"${CDK_DIR}"'/outputs.json"))["AgentCorePaymentsFunFactsSellerStack"]["SellerEvmWallet"])')"
SVM_WALLET="$(python3 -c 'import json; print(json.load(open("'"${CDK_DIR}"'/outputs.json"))["AgentCorePaymentsFunFactsSellerStack"]["SellerSolanaWallet"])')"
echo ""
echo "── Deploy Complete ─────────────────────────────────────────"
echo "✅ SellerApiUrl: ${API_URL}"
echo " EVM payout wallet: ${EVM_WALLET}"
echo " Solana payout wallet: ${SVM_WALLET}"
echo ""
# Upsert SELLER_API_URL into .env so §3/§5/§7 in the notebook pick it
# up automatically on the next load_dotenv() without the user editing
# by hand. Preserves comments and other lines.
ENV_FILE="${USE_CASE_ROOT}/.env"
if [ ! -f "${ENV_FILE}" ]; then
cp "${USE_CASE_ROOT}/env-sample.txt" "${ENV_FILE}"
fi
python3 - <<PY
import pathlib
path = pathlib.Path("${ENV_FILE}")
lines = path.read_text().splitlines() if path.exists() else []
out, replaced = [], False
for line in lines:
if line.startswith("SELLER_API_URL="):
out.append(f"SELLER_API_URL=${API_URL}")
replaced = True
else:
out.append(line)
if not replaced:
out.append(f"SELLER_API_URL=${API_URL}")
path.write_text("\n".join(out) + "\n")
PY
echo "💾 .env updated: SELLER_API_URL=${API_URL}"
@@ -0,0 +1,37 @@
#!/usr/bin/env bash
# Tear down the Pay for API buyer agent runtime + its CloudFormation stack.
#
# Uses the CDK venv deploy-agent.sh created (or creates it on demand so the
# script works standalone). Idempotent — safe to re-run; if the stack is
# already gone, CDK reports "No stacks match the name pattern" and exits
# cleanly.
#
# Usage (from anywhere):
# bash test/integration/destroy-agent.sh
set -euo pipefail
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
USE_CASE_ROOT="$(cd "${SCRIPT_DIR}/../.." && pwd)"
CDK_DIR="${USE_CASE_ROOT}/agent/cdk"
# Activate the CDK venv created by deploy-agent.sh. If it's missing (e.g.
# user cleaned up artifacts), rebuild it so `cdk destroy` can synth the
# Python app.
if [ ! -d "${CDK_DIR}/.venv" ]; then
echo "Creating Python venv for CDK..."
python3 -m venv "${CDK_DIR}/.venv"
# shellcheck disable=SC1091
source "${CDK_DIR}/.venv/bin/activate"
pip install --quiet --upgrade pip
pip install --quiet -r "${CDK_DIR}/requirements.txt"
else
# shellcheck disable=SC1091
source "${CDK_DIR}/.venv/bin/activate"
fi
echo "Destroying AgentCorePaymentsBuyerAgentStack..."
(cd "${CDK_DIR}" && cdk destroy --force)
echo ""
echo "✅ Agent runtime stack destroyed."
@@ -0,0 +1,24 @@
#!/usr/bin/env bash
# Tear down the Fun Facts seller stack.
#
# Usage (from anywhere):
# bash test/integration/destroy-seller.sh
set -euo pipefail
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
# Scripts live at <use-case>/test/integration/ — ../../ resolves the
# use-case root.
USE_CASE_ROOT="$(cd "${SCRIPT_DIR}/../.." && pwd)"
CDK_DIR="${USE_CASE_ROOT}/seller/cdk"
if [ -d "${CDK_DIR}/.venv" ]; then
# shellcheck disable=SC1091
source "${CDK_DIR}/.venv/bin/activate"
fi
echo "Destroying AgentCorePaymentsFunFactsSellerStack..."
(cd "${CDK_DIR}" && cdk destroy --force)
echo ""
echo "✅ Seller stack destroyed."
@@ -0,0 +1,10 @@
#!/usr/bin/env bash
# Seed .env from env-sample.txt and generate a fresh USER_ID. Idempotent:
# re-runs leave existing values alone.
#
# Usage:
# bash test/integration/setup-env.sh
set -euo pipefail
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
exec python3 "${SCRIPT_DIR}/setup_env.py"
@@ -0,0 +1,363 @@
#!/usr/bin/env bash
# setup-roles.sh — create the four IAM roles the notebook assumes into.
#
# Creates, idempotently:
# AgentCorePaymentsControlPlaneRole — manages Manager/Connector/CredentialProvider
# AgentCorePaymentsManagementRole — manages Instrument/Session (explicit Deny on ProcessPayment)
# AgentCorePaymentsProcessPaymentRole — signs payments, reads Instrument/Session
# AgentCorePaymentsResourceRetrievalRole — service-assumed, retrieves credentials at runtime
#
# Policies are based on the four-role separation-of-duties model
# recommended for AgentCore Payments (ControlPlane / Management /
# ProcessPayment / ResourceRetrieval — see the main README for the
# full policy text).
# After creating the roles, writes their ARNs into the use-case .env so the
# notebook picks them up without further editing.
#
# Re-running is safe: existing roles are left alone, their policies are
# updated in place, and .env values are only written if empty.
#
# Usage:
# bash test/integration/setup-roles.sh
set -euo pipefail
# ── Path plumbing ─────────────────────────────────────────────────────
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
USE_CASE_ROOT="$(cd "${SCRIPT_DIR}/../.." && pwd)"
ENV_FILE="${USE_CASE_ROOT}/.env"
TEMPLATE="${USE_CASE_ROOT}/env-sample.txt"
# ── Prerequisites ─────────────────────────────────────────────────────
command -v aws >/dev/null 2>&1 || {
echo "❌ aws CLI not found — install AWS CLI v2 first." >&2
exit 1
}
ACCOUNT_ID="$(aws sts get-caller-identity --query Account --output text)"
if [ -z "${ACCOUNT_ID}" ] || [ "${ACCOUNT_ID}" = "None" ]; then
echo "❌ Could not resolve AWS account. Run 'aws configure' first." >&2
exit 1
fi
echo "✅ Account: ${ACCOUNT_ID}"
echo
# ── Role definitions ──────────────────────────────────────────────────
CP_ROLE="AgentCorePaymentsControlPlaneRole"
MGMT_ROLE="AgentCorePaymentsManagementRole"
PP_ROLE="AgentCorePaymentsProcessPaymentRole"
RR_ROLE="AgentCorePaymentsResourceRetrievalRole"
# Standard account trust policy — lets any IAM principal in this account
# assume the role. Good enough for a tutorial; tighten for production.
ACCOUNT_TRUST_POLICY=$(cat <<JSON
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Principal": {"AWS": "arn:aws:iam::${ACCOUNT_ID}:root"},
"Action": "sts:AssumeRole"
}
]
}
JSON
)
# Service trust policy for the ResourceRetrievalRole. The service assumes
# it on behalf of whichever Payment Manager it is acting for; the condition
# keys scope access to this account.
SERVICE_TRUST_POLICY=$(cat <<JSON
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Principal": {"Service": "bedrock-agentcore.amazonaws.com"},
"Action": "sts:AssumeRole",
"Condition": {
"StringEquals": {"aws:SourceAccount": "${ACCOUNT_ID}"}
}
}
]
}
JSON
)
# ── ControlPlaneRole policy ───────────────────────────────────────────
RR_ROLE_ARN="arn:aws:iam::${ACCOUNT_ID}:role/${RR_ROLE}"
CP_POLICY=$(cat <<JSON
{
"Version": "2012-10-17",
"Statement": [
{
"Sid": "AllowPaymentManagerOperations",
"Effect": "Allow",
"Action": [
"bedrock-agentcore:CreatePaymentManager",
"bedrock-agentcore:GetPaymentManager",
"bedrock-agentcore:ListPaymentManagers",
"bedrock-agentcore:DeletePaymentManager",
"bedrock-agentcore:UpdatePaymentManager"
],
"Resource": ["arn:aws:bedrock-agentcore:*:${ACCOUNT_ID}:payment-manager/*"]
},
{
"Sid": "AllowPaymentConnectorOperations",
"Effect": "Allow",
"Action": [
"bedrock-agentcore:CreatePaymentConnector",
"bedrock-agentcore:GetPaymentConnector",
"bedrock-agentcore:ListPaymentConnectors",
"bedrock-agentcore:DeletePaymentConnector",
"bedrock-agentcore:UpdatePaymentConnector"
],
"Resource": ["arn:aws:bedrock-agentcore:*:${ACCOUNT_ID}:payment-manager/*/connector/*"]
},
{
"Sid": "AllowCredentialProviderOperations",
"Effect": "Allow",
"Action": [
"bedrock-agentcore:CreatePaymentCredentialProvider",
"bedrock-agentcore:GetPaymentCredentialProvider",
"bedrock-agentcore:ListPaymentCredentialProviders",
"bedrock-agentcore:DeletePaymentCredentialProvider",
"bedrock-agentcore:UpdatePaymentCredentialProvider"
],
"Resource": ["arn:aws:bedrock-agentcore:*:${ACCOUNT_ID}:token-vault/*/paymentcredentialprovider/*"]
},
{
"Sid": "AllowVendedLogDelivery",
"Effect": "Allow",
"Action": ["bedrock-agentcore:AllowVendedLogDeliveryForResource"],
"Resource": ["arn:aws:bedrock-agentcore:*:${ACCOUNT_ID}:payment-manager/*"]
},
{
"Sid": "AllowPassResourceRetrievalRole",
"Effect": "Allow",
"Action": "iam:PassRole",
"Resource": "${RR_ROLE_ARN}",
"Condition": {
"StringEquals": {"iam:PassedToService": "bedrock-agentcore.amazonaws.com"}
}
}
]
}
JSON
)
# ── ManagementRole policy ─────────────────────────────────────────────
# This role manages every PaymentManager / Instrument / Session in the
# account. Wildcards on the Resource line scope to the account but not
# to a specific Manager because the Manager IDs do not exist at role
# creation time (the notebook creates them in §4).
#
# Production hardening: once Manager IDs are stable, replace the `*`
# segments with concrete IDs (for example,
# `payment-manager/${MANAGER_ID}`) or add a tag-based condition such as
# `"Condition": {"StringLike": {"aws:ResourceTag/Project": "pay-for-api"}}`
# to confine the role to tagged resources.
MGMT_POLICY=$(cat <<JSON
{
"Version": "2012-10-17",
"Statement": [
{
"Sid": "AllowPaymentManagement",
"Effect": "Allow",
"Action": [
"bedrock-agentcore:CreatePaymentInstrument",
"bedrock-agentcore:GetPaymentInstrument",
"bedrock-agentcore:GetPaymentInstrumentBalance",
"bedrock-agentcore:ListPaymentInstruments",
"bedrock-agentcore:DeletePaymentInstrument",
"bedrock-agentcore:CreatePaymentSession",
"bedrock-agentcore:GetPaymentSession",
"bedrock-agentcore:ListPaymentSessions",
"bedrock-agentcore:UpdatePaymentSession",
"bedrock-agentcore:DeletePaymentSession"
],
"Resource": [
"arn:aws:bedrock-agentcore:*:${ACCOUNT_ID}:payment-manager/*",
"arn:aws:bedrock-agentcore:*:${ACCOUNT_ID}:payment-manager/*/instrument/*",
"arn:aws:bedrock-agentcore:*:${ACCOUNT_ID}:payment-manager/*/session/*"
]
},
{
"Sid": "DenyProcessPayment",
"Effect": "Deny",
"Action": "bedrock-agentcore:ProcessPayment",
"Resource": "*"
}
]
}
JSON
)
# ── ProcessPaymentRole policy ─────────────────────────────────────────
PP_POLICY=$(cat <<JSON
{
"Version": "2012-10-17",
"Statement": [
{
"Sid": "AllowProcessPayment",
"Effect": "Allow",
"Action": "bedrock-agentcore:ProcessPayment",
"Resource": ["arn:aws:bedrock-agentcore:*:${ACCOUNT_ID}:payment-manager/*/session/*"]
},
{
"Sid": "AllowPaymentReadOperations",
"Effect": "Allow",
"Action": [
"bedrock-agentcore:GetPaymentInstrument",
"bedrock-agentcore:GetPaymentInstrumentBalance",
"bedrock-agentcore:GetPaymentSession"
],
"Resource": [
"arn:aws:bedrock-agentcore:*:${ACCOUNT_ID}:payment-manager/*",
"arn:aws:bedrock-agentcore:*:${ACCOUNT_ID}:payment-manager/*/instrument/*",
"arn:aws:bedrock-agentcore:*:${ACCOUNT_ID}:payment-manager/*/session/*"
]
}
]
}
JSON
)
# ── ResourceRetrievalRole policy ──────────────────────────────────────
# Base permissions only. Per-connector permissions are appended by the
# service itself when a connector is added to the Manager.
RR_POLICY=$(cat <<JSON
{
"Version": "2012-10-17",
"Statement": [
{
"Sid": "WorkloadIdentityCreation",
"Effect": "Allow",
"Action": ["bedrock-agentcore:CreateWorkloadIdentity"],
"Resource": [
"arn:aws:bedrock-agentcore:*:${ACCOUNT_ID}:workload-identity-directory/default",
"arn:aws:bedrock-agentcore:*:${ACCOUNT_ID}:workload-identity-directory/default/workload-identity/*"
]
},
{
"Sid": "WorkloadIdentityAccess",
"Effect": "Allow",
"Action": ["bedrock-agentcore:GetWorkloadAccessToken"],
"Resource": [
"arn:aws:bedrock-agentcore:*:${ACCOUNT_ID}:workload-identity-directory/default",
"arn:aws:bedrock-agentcore:*:${ACCOUNT_ID}:workload-identity-directory/default/workload-identity/*"
]
},
{
"Sid": "PaymentTokenBaseAccess",
"Effect": "Allow",
"Action": ["bedrock-agentcore:GetResourcePaymentToken"],
"Resource": [
"arn:aws:bedrock-agentcore:*:${ACCOUNT_ID}:token-vault/default",
"arn:aws:bedrock-agentcore:*:${ACCOUNT_ID}:workload-identity-directory/default",
"arn:aws:bedrock-agentcore:*:${ACCOUNT_ID}:workload-identity-directory/default/workload-identity/*"
]
}
]
}
JSON
)
# ── Helpers ───────────────────────────────────────────────────────────
role_exists() {
aws iam get-role --role-name "$1" >/dev/null 2>&1
}
create_or_update_role() {
local name="$1"
local trust="$2"
local policy_name="$3"
local policy_doc="$4"
if role_exists "${name}"; then
echo "${name} already exists — updating trust + policy"
aws iam update-assume-role-policy \
--role-name "${name}" \
--policy-document "${trust}" >/dev/null
else
echo " + Creating ${name}"
aws iam create-role \
--role-name "${name}" \
--assume-role-policy-document "${trust}" \
--description "AgentCore Payments tutorial role" >/dev/null
fi
aws iam put-role-policy \
--role-name "${name}" \
--policy-name "${policy_name}" \
--policy-document "${policy_doc}" >/dev/null
echo " ↳ policy ${policy_name} applied"
}
# ── Create / update roles ─────────────────────────────────────────────
echo "=== Creating / updating IAM roles ==="
create_or_update_role "${CP_ROLE}" "${ACCOUNT_TRUST_POLICY}" "ControlPlanePolicy" "${CP_POLICY}"
create_or_update_role "${MGMT_ROLE}" "${ACCOUNT_TRUST_POLICY}" "ManagementPolicy" "${MGMT_POLICY}"
create_or_update_role "${PP_ROLE}" "${ACCOUNT_TRUST_POLICY}" "ProcessPaymentPolicy" "${PP_POLICY}"
create_or_update_role "${RR_ROLE}" "${SERVICE_TRUST_POLICY}" "ResourceRetrievalPolicy" "${RR_POLICY}"
echo
CP_ROLE_ARN="arn:aws:iam::${ACCOUNT_ID}:role/${CP_ROLE}"
MGMT_ROLE_ARN="arn:aws:iam::${ACCOUNT_ID}:role/${MGMT_ROLE}"
PP_ROLE_ARN="arn:aws:iam::${ACCOUNT_ID}:role/${PP_ROLE}"
echo "=== Role ARNs ==="
echo " CONTROL_PLANE_ROLE_ARN: ${CP_ROLE_ARN}"
echo " MANAGEMENT_ROLE_ARN: ${MGMT_ROLE_ARN}"
echo " PROCESS_PAYMENT_ROLE_ARN: ${PP_ROLE_ARN}"
echo " RESOURCE_RETRIEVAL_ROLE_ARN: ${RR_ROLE_ARN}"
echo
# ── Write ARNs back to .env ───────────────────────────────────────────
# Only set values for keys that are empty or have the <ACCOUNT_ID> placeholder.
# Never clobber a hand-edited value.
if [ ! -f "${ENV_FILE}" ]; then
if [ -f "${TEMPLATE}" ]; then
cp "${TEMPLATE}" "${ENV_FILE}"
echo " Seeded ${ENV_FILE} from env-sample.txt"
else
touch "${ENV_FILE}"
echo " Created empty ${ENV_FILE}"
fi
fi
write_env_var() {
local key="$1"
local value="$2"
# Match KEY=, KEY=<…>, or KEY=arn:aws:iam::<ACCOUNT_ID>:…
local current
current="$(awk -F '=' -v k="${key}" '$1 == k { sub(/^[^=]+=/, ""); print; exit }' "${ENV_FILE}" 2>/dev/null || true)"
case "${current}" in
"" | "<"* | *"<ACCOUNT_ID>"*)
if grep -q "^${key}=" "${ENV_FILE}"; then
# in-place update using a tmp file so we don't depend on sed -i flavour
awk -F '=' -v k="${key}" -v v="${value}" \
'{ if ($1 == k) print k "=" v; else print $0 }' "${ENV_FILE}" > "${ENV_FILE}.tmp"
mv "${ENV_FILE}.tmp" "${ENV_FILE}"
else
echo "${key}=${value}" >> "${ENV_FILE}"
fi
echo " ✅ Wrote ${key} to .env"
;;
*)
echo "${key} already set — leaving alone (${current})"
;;
esac
}
echo "=== Updating ${ENV_FILE} ==="
write_env_var "CONTROL_PLANE_ROLE_ARN" "${CP_ROLE_ARN}"
write_env_var "MANAGEMENT_ROLE_ARN" "${MGMT_ROLE_ARN}"
write_env_var "PROCESS_PAYMENT_ROLE_ARN" "${PP_ROLE_ARN}"
write_env_var "RESOURCE_RETRIEVAL_ROLE_ARN" "${RR_ROLE_ARN}"
echo
echo "✅ Done. Next: run the §2 setup cell in the notebook to fill in credentials"
@@ -0,0 +1,138 @@
"""Env-file plumbing for the pay-for-api tutorial.
Provides a small set of helpers the notebook + utility scripts use to
seed ``.env`` from ``env-sample.txt`` and write non-secret values
(``USER_ID``, role ARNs, manager IDs, etc.) into it.
User-supplied wallet-provider secrets (Coinbase / Privy keys, Privy
authorization private key) are pasted into ``.env`` by hand. The
notebook's §2 cell opens ``.env`` in the editor for the user and lists
the keys that still need values. The notebook's §4 then reads those
secrets once, passes them to ``CreatePaymentCredentialProvider``, and
AgentCore Identity stores them in AWS Secrets Manager under KMS
encryption and surfaces only the secret ARN to the agent. The local
``.env`` copy is no longer needed at runtime after that point and can
be cleared by hand. Nothing in this module ever logs, transmits, or
reads back secret material.
Entry points:
- ``python3 test/integration/setup_env.py`` — CLI; seeds ``.env`` and
generates a fresh ``USER_ID`` if missing.
- ``from setup_env import seed_env, write_env_var`` — programmatic API.
"""
from __future__ import annotations
import pathlib
import shutil
import sys
import uuid
# ── Path plumbing ─────────────────────────────────────────────────────
# The Python module lives at test/integration/setup_env.py; walk up two
# levels to land at the use-case root where env-sample.txt and .env live.
HERE = pathlib.Path(__file__).resolve().parent
USE_CASE_ROOT = HERE.parent.parent
TEMPLATE = USE_CASE_ROOT / "env-sample.txt"
ENV_FILE = USE_CASE_ROOT / ".env"
# Tokens that mean "this slot has not been filled yet" — treat like empty.
PLACEHOLDER_PREFIXES = ("<",)
PLACEHOLDER_SUBSTRINGS = ("<ACCOUNT_ID>",)
def _is_empty(value: str) -> bool:
"""True if the value is unset, blank, or a template placeholder."""
if not value:
return True
if any(value.startswith(p) for p in PLACEHOLDER_PREFIXES):
return True
if any(s in value for s in PLACEHOLDER_SUBSTRINGS):
return True
return False
def _read_env_lines() -> list[str]:
return ENV_FILE.read_text().splitlines() if ENV_FILE.exists() else []
def _current_value(key: str) -> str:
for line in _read_env_lines():
if line.startswith(f"{key}="):
return line.split("=", 1)[1]
return ""
def write_env_var(key: str, value: str) -> None:
"""Update or append KEY=VALUE in .env without touching other lines.
Only intended for non-secret values written programmatically by the
notebook (USER_ID, role ARNs, manager IDs, instrument IDs, session
IDs, wallet addresses). Wallet-provider secrets (Coinbase / Privy
keys, Privy authorization private key) are pasted into ``.env`` by
the user manually and never flow through this function. Once §4 of
the notebook calls ``CreatePaymentCredentialProvider``, those
secrets are stored in AWS Secrets Manager under AgentCore Identity
and only the credential-provider ARN remains in ``.env``.
"""
lines = _read_env_lines()
replaced = False
out: list[str] = []
for line in lines:
if line.startswith(f"{key}="):
out.append(f"{key}={value}")
replaced = True
else:
out.append(line)
if not replaced:
out.append(f"{key}={value}")
ENV_FILE.write_text("\n".join(out) + "\n")
def seed_env() -> bool:
"""Create .env from env-sample.txt if it doesn't exist and ensure
USER_ID is set to a unique UUID.
Returns True if .env was created on this call, False if it was
already there.
"""
seeded = False
if not ENV_FILE.exists():
if not TEMPLATE.exists():
raise FileNotFoundError(
f"env-sample.txt not found at {TEMPLATE}. Run this from the use-case root with the template in place."
)
shutil.copy2(TEMPLATE, ENV_FILE)
seeded = True
# Auto-generate USER_ID on first run. The notebook uses USER_ID as
# the operator identifier on CreatePaymentSession headers. A fixed
# value across runs caused collisions in the service's vendor-user
# mapping, so each fresh .env gets its own UUID.
#
# The `pay-for-api-` prefix marks this as a tutorial-scoped
# identifier; production code should generate USER_IDs from your
# own auth system rather than reusing this format.
if _is_empty(_current_value("USER_ID")):
write_env_var("USER_ID", f"pay-for-api-{uuid.uuid4()}")
return seeded
def _cli() -> int:
if seed_env():
print(f"✅ Seeded {ENV_FILE} from env-sample.txt.")
else:
print(f"↷ Found existing {ENV_FILE} — left in place.")
print()
print(
"Open .env in your editor and fill in any missing values "
"(secrets are paste-only; non-secrets are written for you by "
"later notebook cells)."
)
return 0
if __name__ == "__main__":
sys.exit(_cli())
@@ -0,0 +1,156 @@
"""
utils.py — shared helpers for the pay-for-api notebook.
Small wrappers around boto3 for pretty-printing responses, assuming IAM
roles, polling for status transitions, and tolerating idempotent
create calls.
"""
import json
import time
import boto3
import botocore.exceptions
def pp(label: str, response: dict) -> None:
"""Pretty-print an API response, stripping ResponseMetadata."""
data = {k: v for k, v in response.items() if k != "ResponseMetadata"}
print(f"\n{'=' * 60}")
print(f" {label}")
print(f"{'=' * 60}")
print(json.dumps(data, indent=2, default=str))
def assume_role(
session: boto3.Session,
role_arn: str,
session_name: str = "tutorial-session",
) -> boto3.Session:
"""Assume an IAM role and return a boto3 Session with auto-refreshing credentials.
Uses botocore's ``RefreshableCredentials`` under the hood so sessions
stay valid past the default 1-hour STS expiry without the caller
having to rebuild clients. This matters for the notebook, where a
user can leave §5.1's session sitting for hours before coming back
to §7 / §9.
Immediately verifies the assumed identity by calling
get_caller_identity(); raises if the assumption fails outright.
"""
from botocore.credentials import RefreshableCredentials
from botocore.session import Session as BotocoreSession
sts = session.client("sts")
def _refresh() -> dict:
creds = sts.assume_role(
RoleArn=role_arn,
RoleSessionName=session_name,
)["Credentials"]
return {
"access_key": creds["AccessKeyId"],
"secret_key": creds["SecretAccessKey"],
"token": creds["SessionToken"],
"expiry_time": creds["Expiration"].isoformat(),
}
refreshable_creds = RefreshableCredentials.create_from_metadata(
metadata=_refresh(),
refresh_using=_refresh,
method="sts-assume-role",
)
botocore_session = BotocoreSession()
botocore_session._credentials = refreshable_creds
botocore_session.set_config_variable("region", session.region_name)
new_session = boto3.Session(botocore_session=botocore_session)
assumed_arn = new_session.client("sts").get_caller_identity()["Arn"]
print(f" Assumed: {assumed_arn}")
return new_session
def wait_for_status(
client_fn,
expected_status: str,
poll_interval: int = 5,
timeout: int = 120,
**kwargs,
) -> dict:
"""Poll a Get* API until the resource reaches expected_status.
Resolves status from these response shapes (checked in order):
- Top-level ``status`` field (Manager, Connector responses)
- ``paymentInstrument.status`` (GetPaymentInstrument response)
Raises TimeoutError if the resource has not reached expected_status
within ``timeout`` seconds.
Raises RuntimeError immediately if the resource enters a terminal
failure state (any status ending in ``_FAILED``).
"""
deadline = time.time() + timeout
while True:
resp = client_fn(**kwargs)
status = resp.get("status") or resp.get("paymentInstrument", {}).get("status")
print(f" Status: {status}")
if isinstance(status, str) and status.endswith("_FAILED"):
raise RuntimeError(f"Resource reached failure state: '{status}'")
if status == expected_status:
return resp
if time.time() >= deadline:
raise TimeoutError(f"Resource still in '{status}' after {timeout}s — check the console for errors")
time.sleep(poll_interval)
def idempotent_create(create_fn, conflict_msg: str = "Resource already exists", **kwargs) -> dict | None:
"""Call create_fn; handle ConflictException gracefully.
Returns the API response on success, or None if the resource already exists.
Re-raises any other ClientError.
"""
try:
return create_fn(**kwargs)
except botocore.exceptions.ClientError as exc:
if exc.response["Error"]["Code"] == "ConflictException":
print(f" ⚠️ {conflict_msg} — skipping create")
return None
raise
def write_env_updates(updates: dict, env_path: str = ".env") -> None:
"""Upsert key=value pairs into a dotenv file, preserving other lines.
Updates in-place — matching keys are replaced, new keys are appended,
comments and blank lines are preserved. Values are written verbatim
(no quoting), matching the existing .env style in this tutorial.
Used only for non-secret values written by the notebook at runtime
(USER_ID, role ARNs, manager IDs, instrument IDs, session IDs,
wallet addresses). Wallet-provider secrets (Coinbase / Privy keys,
Privy authorization private key) are pasted into ``.env`` by the
user manually and never flow through this function. After §4 of
the notebook calls ``CreatePaymentCredentialProvider``, AgentCore
Identity stores those secrets in AWS Secrets Manager under KMS
encryption and only the credential-provider ARN remains in
``.env`` for runtime use. The ``.env`` file itself is gitignored
from use-case creation.
"""
import pathlib
path = pathlib.Path(env_path)
existing = path.read_text().splitlines() if path.exists() else []
seen = set()
out = []
for line in existing:
key = line.split("=", 1)[0].strip()
if key in updates:
out.append(f"{key}={updates[key]}")
seen.add(key)
else:
out.append(line)
for key, value in updates.items():
if key not in seen:
out.append(f"{key}={value}")
path.write_text("\n".join(out) + "\n")