Setting the file. One moment.
Setup Payment User · Agents Build · aws/agent-toolkit-for-aws · Skills Docs
Back to the top of the page Teardown
69
Creating Amazon Aurora Db Cluster With Instances
92
Routing Traffic With Route53 And CloudFront
Resilience Program Design
Debugging Lambda Timeouts
scripts/setup_payment_user.py
scripts/ setup_payment_user.py
Python · 120 lines · 5 KB
argparse
16 import json
17 import os
18 import sys
19 from pathlib import Path
20
21 from bedrock_agentcore.payments import PaymentManager
22
23
24 def _from_deployed_state ():
25 """Best-effort: read manager ARN + connector ID from the CLI's deployed state.
26
27 CLI 0.20.x writes targets.<target>.resources.payments[]; older shapes used a
28 top-level payments[]. Handle both.
29 """
30 path = Path( "agentcore/.cli/deployed-state.json" )
31 if not path.exists():
32 return None , None
33 try :
34 data = json.loads(path.read_text())
35 payments = None
36 targets = data.get( "targets" ) or {}
37 target = targets.get( "default" ) or ( next ( iter (targets.values()), {}) if targets else {})
38 if isinstance (target, dict ):
39 payments = (target.get( "resources" ) or {}).get( "payments" )
40 if not payments:
41 payments = data.get( "payments" ) # legacy/top-level fallback
42 if not payments:
43 return None , None
44 pay = payments[ 0 ]
45 connectors = pay.get( "connectors" ) or []
46 return pay.get( "managerArn" ), (connectors[ 0 ].get( "connectorId" ) if connectors else None )
47 except Exception :
48 return None , None
49
50
51 def main ():
52 ap = argparse.ArgumentParser( description = "Provision a per-user AgentCore Payments instrument" )
53 ap.add_argument( "--user-id" , required = True , help = "Stable end-user identifier" )
54 ap.add_argument( "--email" , required = True , help = "End-user email (linked to the wallet; required for delegation)" )
55 ap.add_argument( "--budget" , default = None , help = "Optional session spend cap in USD, e.g. 5" )
56 ap.add_argument( "--expiry-minutes" , type = int , default = 60 , help = "Session expiry, 15-480" )
57 ap.add_argument( "--network" , default = "ETHEREUM" , help = "Wallet network family: ETHEREUM or SOLANA" )
58 ap.add_argument( "--manager-arn" , default = os.environ.get( "PAYMENT_MANAGER_ARN" ))
59 ap.add_argument( "--connector-id" , default = os.environ.get( "PAYMENT_CONNECTOR_ID" ))
60 ap.add_argument( "--region" , default = os.environ.get( "AWS_REGION" , "us-east-1" ))
61 args = ap.parse_args()
62
63 manager_arn, connector_id = args.manager_arn, args.connector_id
64 if not manager_arn or not connector_id:
65 ds_arn, ds_conn = _from_deployed_state()
66 manager_arn = manager_arn or ds_arn
67 connector_id = connector_id or ds_conn
68 if not manager_arn or not connector_id:
69 sys.exit( "Could not resolve manager ARN / connector ID. Pass --manager-arn and --connector-id, "
70 "or run from the project dir with agentcore/.cli/deployed-state.json present." )
71
72 manager = PaymentManager( payment_manager_arn = manager_arn, region_name = args.region)
73
74 # Data plane: per-user instrument (wallet). Email -> linkedAccounts.
75 instrument = manager.create_payment_instrument(
76 user_id = args.user_id,
77 payment_connector_id = connector_id,
78 payment_instrument_type = "EMBEDDED_CRYPTO_WALLET" ,
79 payment_instrument_details = {
80 "embeddedCryptoWallet" : {
81 "network" : args.network,
82 "linkedAccounts" : [{ "email" : { "emailAddress" : args.email}}],
83 }
84 },
85 )
86 instrument_id = instrument[ "paymentInstrumentId" ]
87 wallet = instrument.get( "paymentInstrumentDetails" , {}).get( "embeddedCryptoWallet" , {})
88 wallet_address = wallet.get( "walletAddress" )
89 redirect_url = wallet.get( "redirectUrl" ) # Coinbase delegation URL; None for Privy
90
91 # Data plane: optional budget-bounded session. NOTE : cap key is "value", not "amount".
92 session_id = None
93 if args.budget:
94 session = manager.create_payment_session(
95 user_id = args.user_id,
96 expiry_time_in_minutes = args.expiry_minutes,
97 limits = { "maxSpendAmount" : { "value" : str (args.budget), "currency" : "USD" }}, # cap currency is USD
98 )
99 session_id = session[ "paymentSessionId" ]
100
101 print ( "Instrument ID :" , instrument_id)
102 print ( "Wallet address:" , wallet_address)
103 print ( "Session ID :" , session_id or "(none - use `agentcore invoke --auto-session`)" )
104 print ( " \n Export these for the x402 tool (Step 8):" )
105 print ( f " export PAYMENT_MANAGER_ARN= { manager_arn } " )
106 print ( f " export PAYMENT_INSTRUMENT_ID= { instrument_id } " )
107 if session_id:
108 print ( f " export PAYMENT_SESSION_ID= { session_id } " )
109 print ( f " export PAYMENT_USER_ID= { args.user_id } " )
110 print ( f " export AWS_REGION= { args.region } " )
111 print ( " \n One-time per wallet:" )
112 if redirect_url:
113 print ( f " 1. Delegation (Coinbase): visit { redirect_url } , log in, grant access to { wallet_address } " )
114 else :
115 print ( " 1. Delegation (Privy): approve delegation via the Privy frontend SDK" )
116 print ( f " 2. Funding: send testnet USDC to { wallet_address } via https://faucet.circle.com/ (Base Sepolia)" )
117
118
119 if __name__ == "__main__" :
120 main()