Add a complete crypto financial stack to your AWS Bedrock Agent as an Action Group

# Lambda handler for Purple Flea Bedrock Action Group
import os, json, urllib.request, urllib.error

PF_KEY  = os.environ["PURPLEFLEA_API_KEY"]
PF_BASE = "https://purpleflea.com/api/v1"

# Map Bedrock action names to Purple Flea REST endpoints
ROUTES = {
    "GetWalletBalance":   ("GET",  "/wallet/balance"),
    "SendCrypto":         ("POST", "/wallet/send"),
    "CreateWallet":       ("POST", "/wallet/create"),
    "OpenPosition":       ("POST", "/trading/position"),
    "ClosePosition":      ("POST", "/trading/position/close"),
    "GetMarketPrice":     ("GET",  "/trading/price"),
    "PlayDice":           ("POST", "/casino/dice"),
    "PlayCrash":          ("POST", "/casino/crash"),
    "CreateEscrow":       ("POST", "/escrow/create"),
    "ReleaseEscrow":      ("POST", "/escrow/release"),
    "RegisterDomain":     ("POST", "/domains/register"),
    "ClaimFaucet":        ("POST", "/faucet/claim"),
}

def lambda_handler(event, context):
    action_group = event["actionGroup"]
    api_path     = event["apiPath"]       # e.g. /GetWalletBalance
    http_method  = event["httpMethod"]
    params       = event.get("requestBody", {}).get("content", {})

    # Extract JSON body from Bedrock action group format
    body = {}
    if "application/json" in params:
        props = params["application/json"].get("properties", [])
        body  = {p["name"]: p["value"] for p in props}

    # Resolve route
    action_name = api_path.lstrip("/")
    if action_name not in ROUTES:
        return _response(404, {"error": f"Unknown action: {action_name}"})

    method, path = ROUTES[action_name]
    url          = f"{PF_BASE}{path}"
    payload      = json.dumps(body).encode()

    req = urllib.request.Request(
        url, data=payload, method=method,
        headers={
            "Authorization": f"Bearer {PF_KEY}",
            "Content-Type": "application/json",
        },
    )
    try:
        with urllib.request.urlopen(req, timeout=15) as r:
            data = json.loads(r.read())
        return _response(200, data)
    except urllib.error.HTTPError as e:
        return _response(e.code, {"error": e.read().decode()})

def _response(status: int, body: dict) -> dict:
    return {
        "messageVersion": "1.0",
        "response": {
            "actionGroup": "PurpleFleasActions",
            "httpStatusCode": status,
            "responseBody": {
                "application/json": {"body": json.dumps(body)},
            },
        },
    }

# pip install boto3
import boto3

AGENT_ID      = "YOUR_BEDROCK_AGENT_ID"
AGENT_VERSION = "DRAFT"
LAMBDA_ARN    = "arn:aws:lambda:us-east-1:123456789:function:purpleflea-handler"

client = boto3.client("bedrock-agent", region_name="us-east-1")

response = client.create_agent_action_group(
    agentId=AGENT_ID,
    agentVersion=AGENT_VERSION,
    actionGroupName="PurpleFleasActions",
    description="Crypto wallets, trading, casino, escrow, and domains via Purple Flea",
    actionGroupExecutor={
        "lambda": LAMBDA_ARN,
    },
    apiSchema={
        "s3": {
            # Or pass inlinePayload if you prefer
        },
        # Use hosted spec — always up to date
    },
    # Alternatively, inline the OpenAPI spec:
    # apiSchema={"payload": open("purpleflea-openapi.json").read()},
)

print(f"Action group created: {response['agentActionGroup']['actionGroupId']}")

# Prepare and deploy agent
client.prepare_agent(agentId=AGENT_ID)
print("Agent prepared and ready for alias creation")

import boto3, uuid

runtime = boto3.client("bedrock-agent-runtime", region_name="us-east-1")

response = runtime.invoke_agent(
    agentId="YOUR_AGENT_ID",
    agentAliasId="YOUR_ALIAS_ID",
    sessionId=str(uuid.uuid4()),
    inputText=(
        "Claim from the faucet, then open a $500 long on ETH-PERP with 5x leverage. "
        "Once confirmed, register agent-task-001.xyz as our operational domain."
    ),
)

for event in response["completion"]:
    if "chunk" in event:
        print(event["chunk"]["bytes"].decode(), end="")