Connect AWS Bedrock AI agents to Purple Flea's 6-service financial platform. Casino, Trading, Wallet, Domains, Faucet, and Escrow β all accessible via standard REST APIs your Bedrock Lambda actions can call in seconds.
AWS Bedrock is where enterprise AI agents live: Claude 3, Titan, Llama 3, and Mistral models running inside your VPC with full IAM control, CloudTrail audit logs, and GuardDuty threat detection. It is production-grade AI infrastructure trusted by thousands of enterprises.
Purple Flea is where those agents earn money. Six financial services purpose-built for autonomous agents β no KYC, no minimum balances, no human approval required. REST APIs that respond in under 200ms. USDC-native, provably fair, trustless.
Together, they give you the full stack: AWS Bedrock handles the intelligence and compliance; Purple Flea handles the economics. Your Bedrock agent can reason about markets, place trades, manage a casino bankroll, and pay other agents for services β all within a single orchestrated workflow.
The fastest path to a Bedrock agent with financial capabilities. This example creates an agent session, invokes a prompt that triggers a Purple Flea API call, and processes the response. Total setup time: under 15 minutes.
import boto3, json, uuid # Initialize Bedrock Agents Runtime client client = boto3.client("bedrock-agent-runtime", region_name="us-east-1") # IDs from your Bedrock console after creating agent + action group AGENT_ID = "YOUR_BEDROCK_AGENT_ID" AGENT_ALIAS = "TSTALIASID" # or your production alias SESSION_ID = str(uuid.uuid4()) def invoke_financial_agent(prompt: str) -> str: """Invoke Bedrock agent with a financial task prompt.""" response = client.invoke_agent( agentId=AGENT_ID, agentAliasId=AGENT_ALIAS, sessionId=SESSION_ID, inputText=prompt, ) # Stream the response chunks completion = "" for event in response["completion"]: chunk = event.get("chunk", {}) if "bytes" in chunk: completion += chunk["bytes"].decode() return completion # Example: Ask the agent to check balance and place a casino bet result = invoke_financial_agent( "Check my Purple Flea wallet balance, then place a $0.50 coin flip bet. " "Report the outcome and my new balance." ) print(result)
The agent automatically decides which Purple Flea API to call based on the prompt. The action group definition (below) tells Bedrock exactly which operations are available and what parameters they require.
Bedrock Agents use OpenAPI schemas to understand what APIs they can call. The following schema covers the core Purple Flea operations. Upload this to your Bedrock Agent action group in the AWS Console or via the CLI.
openapi: "3.0.0" info: title: Purple Flea Financial API version: "1.0" description: Financial services for AI agents - casino, trading, wallet, escrow, faucet paths: /wallet/balance: get: operationId: getWalletBalance summary: Get agent USDC wallet balance responses: "200": description: Current balance content: application/json: schema: type: object properties: usdc: {type: number} agent_id: {type: string} /casino/bet: post: operationId: placeCasinoBet summary: Place a bet on a casino game requestBody: required: true content: application/json: schema: type: object required: [game, amount, choice] properties: game: type: string enum: [coin, dice, roulette, crash] amount: type: number minimum: 0.01 choice: type: string responses: "200": description: Bet result content: application/json: schema: type: object properties: outcome: {type: string} payout_usdc: {type: number} new_balance: {type: number} proof: {type: string} /trading/open: post: operationId: openTradePosition summary: Open a trading position requestBody: required: true content: application/json: schema: type: object required: [symbol, side, size_usd] properties: symbol: {type: string, example: BTCUSDT} side: {type: string, enum: [LONG, SHORT]} size_usd: {type: number, minimum: 1} stop_loss_pct: {type: number} take_profit_pct: {type: number} responses: "200": description: Position opened content: application/json: schema: type: object properties: position_id: {type: string} entry_price: {type: number} size_usd: {type: number} /escrow/create: post: operationId: createEscrow summary: Create trustless USDC escrow for agent-to-agent payment requestBody: required: true content: application/json: schema: type: object required: [recipient, amount, condition] properties: recipient: {type: string} amount: {type: number, minimum: 0.01} condition: {type: string} deadline_hours: {type: integer, default: 24} responses: "200": description: Escrow created content: application/json: schema: type: object properties: escrow_id: {type: string} locked_amount: {type: number} fee_usdc: {type: number} /faucet/claim: post: operationId: claimFaucet summary: Claim $1 free USDC (new agents only) responses: "200": description: Faucet claimed content: application/json: schema: type: object properties: claimed: {type: boolean} amount_usdc: {type: number} balance: {type: number}
In the Bedrock console, go to your Agent β Action Groups β Create action group. Select "Define with API schemas" and upload this YAML. Set the Lambda function to your Purple Flea handler (see next section). The agent will automatically route relevant prompts to the appropriate API operation.
Bedrock Agents call Lambda functions to execute action group operations. The following handler processes all Purple Flea operations and retrieves the API key securely from AWS Secrets Manager.
import json, boto3, requests, os # Lazy-load secret to take advantage of Lambda execution environment reuse _api_key: str | None = None def get_api_key() -> str: global _api_key if _api_key is not None: return _api_key sm = boto3.client("secretsmanager") secret = sm.get_secret_value(SecretId=os.environ["PF_SECRET_ARN"]) _api_key = json.loads(secret["SecretString"])["api_key"] return _api_key PF_HOSTS = { "wallet": "https://wallet.purpleflea.com/api/v1", "casino": "https://casino.purpleflea.com/api/v1", "trading": "https://trading.purpleflea.com/api/v1", "escrow": "https://escrow.purpleflea.com/api/v1", "faucet": "https://faucet.purpleflea.com/api/v1", } OPERATION_MAP = { "getWalletBalance": ("wallet", "GET", "/balance"), "placeCasinoBet": ("casino", "POST", "/bet"), "openTradePosition": ("trading", "POST", "/open"), "createEscrow": ("escrow", "POST", "/create"), "claimFaucet": ("faucet", "POST", "/claim"), } def lambda_handler(event: dict, context) -> dict: operation = event["actionGroup"] + "/" + event["function"] function = event["function"] body = {p["name"]: p["value"] for p in event.get("parameters", [])} if function not in OPERATION_MAP: return {"statusCode": 400, "body": f"Unknown operation: {function}"} service, method, path = OPERATION_MAP[function] url = PF_HOSTS[service] + path headers = {"X-API-Key": get_api_key(), "Content-Type": "application/json"} if method == "GET": resp = requests.get(url, headers=headers) else: resp = requests.post(url, json=body, headers=headers) resp.raise_for_status() data = resp.json() # Bedrock expects this exact response structure return { "messageVersion": "1.0", "response": { "actionGroup": event["actionGroup"], "function": function, "functionResponse": { "responseBody": {"TEXT": {"body": json.dumps(data)}} }, }, }
AWS Bedrock Knowledge Bases allow agents to query documents using semantic search. Connecting your Purple Flea transaction history to a Bedrock Knowledge Base lets your agent answer questions like "What was my win rate in casino games last month?" or "Which trading symbols have generated the most profit?" using retrieval-augmented generation.
import boto3, json, requests from datetime import datetime def export_history_to_s3(api_key: str, s3_bucket: str, s3_prefix: str): """Export Purple Flea trade/bet history to S3 for Bedrock RAG indexing.""" resp = requests.get( "https://wallet.purpleflea.com/api/v1/transactions", headers={"X-API-Key": api_key}, params={"days": 90}, ) transactions = resp.json()["transactions"] # Convert to natural language chunks for better RAG retrieval s3 = boto3.client("s3") chunks = [] for tx in transactions: chunk = ( f"On {tx['timestamp']}, {tx['type']} transaction: " f"amount ${tx['amount_usdc']} USDC, service: {tx['service']}, " f"outcome: {tx.get('outcome', 'N/A')}, balance after: ${tx['balance_after']} USDC." ) chunks.append(chunk) # Upload as single document (Bedrock will chunk it automatically) content = "\n\n".join(chunks) key = f"{s3_prefix}/transactions_{datetime.utcnow().strftime('%Y%m%d')}.txt" s3.put_object(Bucket=s3_bucket, Key=key, Body=content.encode()) print(f"Uploaded {len(chunks)} transactions to s3://{s3_bucket}/{key}") def trigger_kb_sync(knowledge_base_id: str, data_source_id: str): """Trigger Bedrock Knowledge Base re-index after new data is uploaded.""" bedrock = boto3.client("bedrock-agent") bedrock.start_ingestion_job( knowledgeBaseId=knowledge_base_id, dataSourceId=data_source_id, ) print("Knowledge base sync triggered.")
Schedule this sync daily using EventBridge. Your agent will then be able to answer performance questions using its own trading and betting history as context, without needing to call the Purple Flea API on every query.
Never hardcode your Purple Flea API key. Store it in AWS Secrets Manager and reference it by ARN. Grant your Lambda function's execution role the minimum permissions required to read the secret and call Purple Flea APIs.
{
"Version": "2012-10-17",
"Statement": [
{
"Sid": "ReadPurpleFleavKey",
"Effect": "Allow",
"Action": ["secretsmanager:GetSecretValue"],
"Resource": "arn:aws:secretsmanager:us-east-1:ACCOUNT:secret:purpleflea-api-key-*"
},
{
"Sid": "AllowBedrockInvoke",
"Effect": "Allow",
"Action": [
"bedrock:InvokeAgent",
"bedrock:InvokeModel"
],
"Resource": "*"
},
{
"Sid": "S3KnowledgeBase",
"Effect": "Allow",
"Action": ["s3:PutObject", "s3:GetObject"],
"Resource": "arn:aws:s3:::your-kb-bucket/purpleflea/*"
},
{
"Sid": "LogsWrite",
"Effect": "Allow",
"Action": ["logs:CreateLogGroup", "logs:CreateLogStream", "logs:PutLogEvents"],
"Resource": "arn:aws:logs:*:*:*"
}
]
}
# Create the secret (run once) aws secretsmanager create-secret \ --name purpleflea-api-key \ --description "Purple Flea API key for Bedrock agent" \ --secret-string '{"api_key":"pf_live_your_key_here"}' \ --region us-east-1 # Rotate the key (run anytime you issue a new key) aws secretsmanager update-secret \ --secret-id purpleflea-api-key \ --secret-string '{"api_key":"pf_live_new_key_here"}'
Purple Flea API keys use the prefix pf_live_ for production and pf_test_ for sandbox. Never use generic key names like sk_*. The pf_live_ prefix makes it immediately clear in logs and code reviews that the credential is a Purple Flea key.
Bedrock charges per input and output token. Every tool call result that gets included in the conversation context consumes tokens. If your agent checks its wallet balance 50 times per day, that is 50 redundant API calls and 50 * ~200 tokens of context each time the balance appears in the prompt.
A simple in-memory cache with TTL reduces both Bedrock token costs and Purple Flea API calls. For data that changes slowly (balance, open positions), a 60-second TTL is typically sufficient.
import time from functools import wraps class TTLCache: def __init__(self): self._store: dict = {} def get(self, key: str, ttl: int) -> tuple[bool, object]: if key not in self._store: return False, None value, ts = self._store[key] if time.time() - ts > ttl: del self._store[key] return False, None return True, value def set(self, key: str, value: object): self._store[key] = (value, time.time()) _cache = TTLCache() def cached_wallet_balance(api_key: str, ttl: int = 60) -> float: """Balance cached for 60s. Avoids redundant API calls in rapid agent loops.""" cache_key = f"balance_{api_key[-8:]}" hit, value = _cache.get(cache_key, ttl) if hit: return value import requests resp = requests.get( "https://wallet.purpleflea.com/api/v1/balance", headers={"X-API-Key": api_key}, ) balance = resp.json()["usdc"] _cache.set(cache_key, balance) return balance
In Lambda, execution environment reuse means that module-level variables persist between warm invocations. The TTLCache above lives at module level and survives across multiple Lambda invocations in the same execution environment β giving you free in-memory caching without ElastiCache or DynamoDB for low-concurrency functions.
The following ASCII diagram shows the recommended production architecture for a Bedrock agent integrated with Purple Flea. Each component maps to AWS managed services, minimizing operational overhead.
βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
β AWS VPC β
β β
β βββββββββββββββ invoke ββββββββββββββββββββββββ β
β β EventBridge ββββββββββββββββΆ Bedrock Agent β β
β β (scheduler) β β (Claude 3 Sonnet) β β
β βββββββββββββββ β β β
β β Orchestration Loop β β
β βββββββββββββββ query KB β 1. Reason about taskβ β
β β Bedrock βββββββββββββββββ 2. Select action β β
β β Knowledge β β 3. Call Lambda β β
β β Base (S3) β β 4. Process result β β
β βββββββββββββββ ββββββββββββ¬ββββββββββββ β
β β action group call β
β βββββββββββββββ get secret ββββββββββββΌββββββββββββ β
β β Secrets ββββββββββββββββ Lambda Function β β
β β Manager β β (pf_handler.py) β β
β β pf_live_* β ββββββββββββ¬ββββββββββββ β
β βββββββββββββββ β HTTPS β
β β β
β βββββββββββββββ β β
β β CloudWatch ββββββββ logs ββββββββββββββ β
β β (metrics, β β
β β alerts) β β
β βββββββββββββββ β
ββββββββββββββββββββββββββββββββββββ¬βββββββββββββββββββββββββββββββ
β HTTPS / REST API
βΌ
βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
β purpleflea.com services β
β β
β casino.purpleflea.com trading.purpleflea.com β
β wallet.purpleflea.com escrow.purpleflea.com β
β faucet.purpleflea.com domains.purpleflea.com β
βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
| AWS Component | Purpose | Purple Flea Interaction |
|---|---|---|
| Bedrock Agent | Orchestrates reasoning and tool use | Decides which PF service to call |
| Lambda | Executes Purple Flea API calls | All 6 services via HTTP |
| Secrets Manager | Stores pf_live_* API key |
Key injected at Lambda runtime |
| Bedrock Knowledge Base | RAG over transaction history | Fed by PF Wallet transaction export |
| EventBridge | Triggers scheduled agent runs | Daily strategy execution, KB sync |
| CloudWatch | Metrics, logs, alerts | PF API response times, error rates |
Before your Bedrock agent can use Purple Flea services, it needs a registered agent ID and an API key. The faucet registration endpoint handles both in a single call. Add this as an initialization step in your Lambda function β it is idempotent, so calling it on every cold start is safe.
import boto3, json, os, requests def initialize_agent(agent_name: str) -> dict: """ Register agent and claim $1 USDC on first run. Returns: {"agent_id": ..., "api_key": ..., "balance": ...} """ # Check if already registered (stored in SSM Parameter Store) ssm = boto3.client("ssm") param_name = f"/purpleflea/agents/{agent_name}/api_key" try: existing = ssm.get_parameter(Name=param_name, WithDecryption=True) api_key = existing["Parameter"]["Value"] print(f"Agent already registered: {agent_name}") return {"agent_id": agent_name, "api_key": api_key, "fresh": False} except ssm.exceptions.ParameterNotFound: pass # Register and claim faucet resp = requests.post("https://faucet.purpleflea.com/api/v1/register", json={ "agent_id": agent_name, "description": f"AWS Bedrock agent: {agent_name}", }) data = resp.json() api_key = data["api_key"] # e.g. pf_live_xxxxxxxxxxxx # Claim $1 USDC claim = requests.post( "https://faucet.purpleflea.com/api/v1/claim", headers={"X-API-Key": api_key}, ).json() # Store API key in SSM for future runs ssm.put_parameter( Name=param_name, Value=api_key, Type="SecureString", Overwrite=False, ) print(f"Registered {agent_name}. Claimed ${claim['amount_usdc']} USDC.") return {"agent_id": agent_name, "api_key": api_key, "fresh": True, "balance": claim.get("balance")}
Get your API key, explore the docs, and claim your $1 free USDC in under 2 minutes.