AI Agent也能自己“花钱”了：Cloudflare与Stripe的自主部署协议实践

import requests
import os
import sys

# 环境变量中获取 Agent 的基础授权 Token
# 实际场景中，这些 Token 由上层调度平台（如 LangChain/CrewAI）注入
STRIPE_AGENT_TOKEN = os.getenv("STRIPE_AGENT_TOKEN", "sk_test_your_stripe_key")
CF_AGENT_TOKEN = os.getenv("CF_AGENT_TOKEN", "your_cloudflare_api_token")

def agent_provision_and_deploy(domain_name: str, worker_file_path: str):
    """
    模拟 AI Agent 自主开户、买域名、部署 Worker 的完整链路
    """
    headers_stripe = {"Authorization": f"Bearer {STRIPE_AGENT_TOKEN}", "Content-Type": "application/json"}
    headers_cf = {"Authorization": f"Bearer {CF_AGENT_TOKEN}", "Content-Type": "application/json"}

    # 步骤 1: Stripe - 创建账户并启动订阅，设定 $100/月 硬顶
    print("[Agent] 正在通过 Stripe 创建云账户并绑定支付...")
    # 注：此处端点为协议逻辑的假设映射，实际端点以 Stripe 官方 Agent SDK 为准
    stripe_resp = requests.post(
        "https://api.stripe.com/v1/agent/accounts",
        headers=headers_stripe,
        json={
            "action": "create_and_subscribe",
            "subscription_cap_cents": 10000  # 100 USD = 10000 cents，协议默认上限
        }
    )

    if stripe_resp.status_code != 200:
        print("开户或订阅失败，Agent 终止流程。")
        sys.exit(1)

    account_id = stripe_resp.json().get("id")
    print(f"[Agent] 账户创建成功，ID: {account_id}，月度预算已锁定 $100。")

    # 步骤 2: Cloudflare - 注册域名
    print(f"[Agent] 正在注册域名: {domain_name}...")
    cf_domain_resp = requests.post(
        "https://api.cloudflare.com/client/v4/domains/register",
        headers=headers_cf,
        json={"domain": domain_name, "account_id": account_id}
    )

    if cf_domain_resp.status_code != 200:
        print("域名注册失败，可能已被占用。")
        sys.exit(1)

    print(f"[Agent] 域名 {domain_name} 注册成功。")

    # 步骤 3: Cloudflare - 部署 Worker 到生产环境
    print("[Agent] 正在打包并部署 Worker 到生产环境...")
    with open(worker_file_path, "rb") as f:
        worker_code = f.read()

    # 使用 Cloudflare 标准 Workers API 进行部署
    cf_deploy_resp = requests.put(
        f"https://api.cloudflare.com/client/v4/accounts/{account_id}/workers/scripts/my-agent-app",
        headers={"Authorization": f"Bearer {CF_AGENT_TOKEN"}, "Content-Type": "application/javascript"},
        data=worker_code
    )

    if cf_deploy_resp.status_code != 200:
        print("Worker 部署失败。")
        sys.exit(1)

    print(f"[Agent] 部署成功！应用已上线: https://{domain_name}")
    return {"domain": domain_name, "status": "live"}

# 运行示例：准备一个简单的 worker.js 文件后执行
# result = agent_provision_and_deploy("my-ai-startup-site.com", "./worker.js")