View a markdown version of this page

将 AgentCore 付款与浏览器工具集成 - Amazon Bedrock AgentCore
浏览器工具集成（剧作家）Non-browser HTTP 集成

将 AgentCore 付款与浏览器工具集成

您的代理可以通过两种方式访问 x402 付费内容：通过浏览器工具或通过标准 HTTP 调用。

浏览器工具集成（剧作家）

AgentCore 浏览器工具为您的代理提供了 AgentCore 运行时内托管的无头 Chromium 浏览器会话。您的代理 WebSocket 使用 Chrome DevTools 协议 (CDP) 进行连接，使其能够在网络级别浏览页面 JavaScript、执行和拦截 HTTP 响应。要开始使用，请参阅浏览器入门指南。

此选项使用Strands SDK以及浏览器工具和付款工具来浏览网站和处理付款。当代理使用浏览器工具浏览网站时，它会使用 Playwright 的响应拦截功能来自动检测付费专区网站：

  1. 浏览器导航 — 代理使用由 Playwright 提供支持的browse_with_payment工具。

  2. 自动 402 检测 — Playwright 拦截所有 HTTP 响应并触发代理的响应处理程序来检测 402 状态码。

  3. x402 提取 — 支付工具会自动解析响应中的 x402 付款明细。 headers/body这也可以通过代理业务逻辑来完成。

  4. 付款处理 — AgentCore 支付处理器使用配置的 CDP 钱包创建加密签名。

  5. 重试-支付工具注入授权标头并在同一个浏览器会话中重试请求。

from agentcore.payments import PaymentClient
from agentcore.browser import browser_tool

@tool
async def browse_with_payment(url: str, auto_pay: bool = True) -> dict:
    payment_requirements = None

    async def handle_response(response):
        nonlocal payment_requirements
        if response.status == 402:
            payment_requirements = await extract_x402_requirements(response)

    async def add_payment_header(route, request):
        headers = {**request.headers, **payment_result["authorization"]}
        await route.continue_(headers=headers)

    async with async_playwright() as p:
        # 1. Browser setup
        browser = await p.chromium.connect_over_cdp(AGENTCORE_BROWSER_WS_URL)
        page = await browser.contexts[0].new_page()

        # 2. Register response interceptor
        page.on("response", handle_response)

        # 3. Initial navigation (handle_response runs automatically here)
        response = await page.goto(url)

        # 4. Payment processing logic (runs AFTER navigation completes)
        if response.status == 402 and payment_requirements and auto_pay:
            payment_result = payment_client.process_payment(
                payment_session,
                payment_instrument,
                payment_requirements=payment_requirements
            )

            if payment_result["success"]:
                await page.route("**/*", add_payment_header)
                response = await page.goto(url)  # Retry with payment
            else:
                raise PaymentError(f"Payment failed: {payment_result['error_message']}")

        # 5. Capture content, cleanup, and return
        content = await page.content()
        await browser.close()
        return {"content": content, "status": response.status}

Non-browser HTTP 集成

对于直接 API 调用和无头场景，开发人员可以使用 Strands SDK 以及标准 HTTP 请求库编写自定义工具逻辑，以便通过 x402 检测处理付款：

  1. 代理使用自定义工具进行直接 HTTP 调用。

  2. 该工具监控 HTTP 响应状态代码以检测 402。

  3. 工具提取 x402 需求并处理付款。

  4. 工具使用付款授权标头重试HTTP请求。

@tool
async def make_payment(method, url, headers, body) -> dict:
    # Make initial HTTP request
    response = requests.request(method, url, headers=headers, data=body)

    # Check for 402 Payment Required
    if response.status_code == 402:
        payment_requirements = extract_x402_requirements(response)
        payment_result = payment_client.process_payment(
            payment_session,
            payment_instrument,
            payment_requirements=payment_requirements
        )

        if payment_result["success"]:
            # Retry request with payment authorization
            headers.update(payment_result["authorization"])
            response = requests.request(method, url, headers=headers, data=body)
        else:
            raise PaymentError(f"Payment failed: {payment_result['error_message']}")