使用AgentCore Payments构建可为API付费的AI代理

本文深入探讨了如何为AI代理构建安全的付费API访问机制,重点介绍了Amazon Bedrock AgentCore Payments与OpenAI Agents SDK的集成。通过一个供应商研究示例,详细阐述了x402支付协议、应用程序控制的支出审批流程、收据验证以及审计追踪。文章提供了完整的代码实现,包括本地模拟和AWS测试网连接,并强调了模型、应用程序、支付基础设施和商户之间的职责分离。读者将学习如何设置支付策略、定义代理工具、处理支付挑战,以及如何验证交易和防止滥用。

技术背景

AI代理在执行任务时,经常需要访问外部付费服务,如风险报告、市场数据或搜索API。这些服务按请求收费,因此应用程序必须建立明确的规则,控制代理可以请求哪些购买以及由谁批准支出。传统的API密钥管理方式无法满足代理动态决策的需求,且存在安全风险。AgentCore Payments提供了有界支付会话的基础设施,而OpenAI Agents SDK允许模型请求应用程序定义的工具。这种组合实现了代理商务工作流,其中代理可以请求付费资源,但应用程序保留支付权限,确保每次购买都经过审批和审计。

核心原理

核心原理是x402支付协议,它利用HTTP 402 Payment Required响应来传达付费请求的条款。当代理请求付费资源时,商户返回一个包含金额、资产、网络、收款人和到期时间的挑战。应用程序根据预定义的策略(如批准的商户、用途、最高金额和预算)检查这些条款。如果通过,应用程序获取有界支付证明(由AgentCore Payments生成),重试请求,商户返回200和内容。整个过程由应用程序控制,模型仅提供资源URL和目的,而审批、支出限额和支付凭证在模型上下文之外。

架构分析

系统架构将四个职责分离:1) AI代理识别何时需要付费服务并调用工具;2) 应用程序检查商户、用途、金额、审批和支出限额,并记录收据和审计;3) AgentCore Payments在批准的会话限制内生成支付证明;4) 付费API验证证明并返回内容。应用程序是模型与支付基础设施之间的边界,确保模型无法直接访问支付凭证。本地模拟使用合成证明,而AWS工作流使用AgentCore Payments生成有界证明。

完整实战教程

AI代理通常需要从外部服务获取信息以完成任务。例如,采购代理可能会在制裁数据库中检查供应商,获取当前的风险报告,或使用付费搜索API。当这些服务按请求收费时,应用程序需要明确规则,规定代理可以请求哪些购买以及由谁批准支出。

代理商务描述了这样的工作流:代理可以请求付费资源,而应用程序代码保留支付权限。Amazon Bedrock AgentCore Payments为受限支付会话提供了基础设施,而OpenAI Agents SDK允许模型请求应用程序定义的工具。应用程序在支付发生前检查商户、用途、金额和审批。

本笔记本通过供应商研究示例解释了这一工作流。它涵盖了AgentCore Payments流程、x402支付交换、应用程序拥有的支出控制以及收据验证。相同的支付控制首先通过本地模拟进行说明,然后连接到AWS测试网上的AgentCore Payments。

为什么智能体需要访问付费服务

设想一个采购智能体正在审查一家虚构的供应商 Northstar Components。智能体拥有基本的供应商信息,但在向人工审核员总结潜在问题之前,它需要从付费 API 获取一份最新的风险报告。

应用程序批准该数据提供商,定义供应商研究目的,并设定报告的最高价格。它还记录了与请求关联的人工审批。当模型决定报告有用时,它会使用供应商报告 URL 和已批准的目的请求 x402_fetch 工具。

商家返回一个 HTTP 402 Payment Required 挑战,描述其支付条款。应用程序根据已批准的商家、金额、网络、到期时间和预算检查这些条款。如果请求通过,应用程序获取有界支付证明,检索报告,并记录收据和审计历史。然后智能体根据工具返回的报告和收据证据准备其答案。

架构与支付授权

系统将四个职责分离:

  1. AI 代理 识别何时付费服务可以帮助完成用户的请求,并调用应用程序定义的工具。
  2. 应用程序 在记录收据和审计历史之前,检查商户、购买目的、金额、审批和支出限额。
  3. AgentCore Payments 在批准的支付会话限制内生成支付证明。
  4. 付费 API 检查支付证明并返回请求的内容。

本地示例模拟了 AgentCore Payments 步骤。连接的 AWS 工作流使用 AgentCore Payments 生成有界证明。

应用程序是模型与支付基础设施之间的边界。模型接收资源 URL 和研究目的,而审批记录、支出限额、支付凭证和审计状态则在其控制之外。

An agent requests a paid supplier report through application controls and AgentCore Payments*应用控制审批和支出,而 AgentCore Payments 为连接的 AWS 工作流提供有界的支付会话。*

1. 设置供应商研究示例

该示例需要 Python 3.11 至 3.13 和 uv。请遵循 README.md 中的仓库根目录 Jupyter 命令,以便笔记本能够加载其包并显示架构图。

第一部分使用脚本化模型、模拟商户和合成支付证明。此设置允许您检查相同的请求、批准、收据和审计序列,而无需配置 AWS 凭证或转移资金。最后一部分展示了真实 Amazon Bedrock 模型和 AgentCore Payments 测试网资源如何进入工作流。

以下单元格定位示例包,并使其应用程序、商户和策略模块可用于笔记本。

import sys
from pathlib import Path

cwd = Path.cwd().resolve()
candidates = [
    cwd,
    cwd / "examples/partners/AWS/controlled_agentic_commerce_with_agentcore_payments",
    *cwd.parents,
]
ROOT = next(
    (path for path in candidates if (path / "src/agentic_commerce").is_dir()),
    None,
)
if ROOT is None:
    raise FileNotFoundError("Run this notebook from its example directory.")
sys.path.insert(0, str(ROOT / "src"))

print(f"Example root: {ROOT}")

2. 定义供应商请求和支出限额

应用程序需要在模型请求工具之前建立其支付规则。在此示例中,供应商报告的费用为 0.25 个合成 USDC。应用程序还定义了批准的商户、允许的研究目的、本次购买的最高金额以及代理运行可用的总预算。

ApprovalGrant 将购买与特定的请求 ID 和资源 URL 绑定。其有效期限制了该授权的有效时长。模型从不提供或更改这些值;它们由应用程序代码在创建工具时提供。

from datetime import UTC, datetime, timedelta
from decimal import Decimal

from agentic_commerce.demo import build_demo
from agentic_commerce.merchant import RESOURCE_URL
from agentic_commerce.models import ApprovalGrant

now = datetime.now(UTC)
application, merchant, local_payments = build_demo(now)
approval = ApprovalGrant(
    approval_id="approval-cookbook-001",
    request_id="request-cookbook-001",
    resource_url=RESOURCE_URL,
    purpose="supplier_due_diligence",
    maximum_amount=Decimal("0.25"),
    approved_by="synthetic-cookbook-reviewer",
    approved_at=now,
    expires_at=now + timedelta(minutes=10),
)

{
    "approved_merchant": str(RESOURCE_URL),
    "purpose": approval.purpose,
    "approved_amount": str(approval.maximum_amount),
    "run_budget": str(application.policy.policy.per_run_limit),
    "currency": approval.currency,
    "approval_expires_at": approval.expires_at.isoformat(),
}

3. 为已批准的购买定义代理工具

x402_fetch 函数是代理可用的唯一与支付相关的工具。模型可以提供资源 URL 和业务目的,但应用程序提供请求 ID、幂等键、支出策略和批准授权。

在开始购买之前,该工具会消耗一次性能力。该检查可防止模型在同一代理运行期间发出多个经济请求。然后,application.purchase 在生成收据之前验证商家和支付挑战。

AgentPurchaseEvidence 仅返回模型准备回答所需的报告和收据字段。钱包材料、支付证明头和完整的应用程序审计跟踪不包含在模型上下文中。

from threading import Lock

from agents import Agent, ModelSettings, function_tool
from openai.types.shared import Reasoning

from agentic_commerce.agent import PurchaseResultRecorder, SupplierResearchOutput
from agentic_commerce.errors import AgentResultInvalid
from agentic_commerce.learning_model import ScriptedCommerceLearningModel
from agentic_commerce.models import AgentPurchaseEvidence, PurchaseRequest

recorder = PurchaseResultRecorder()
model = ScriptedCommerceLearningModel()
tool_capability = {"used": False}
tool_lock = Lock()


@function_tool
def x402_fetch(resource_url: str, purpose: str) -> str:
    """Fetch one paid resource using application-owned approval."""

    with tool_lock:
        if tool_capability["used"]:
            raise AgentResultInvalid(
                "access_count_invalid",
                "The economic tool permits exactly one invocation.",
            )
        tool_capability["used"] = True

    purchase = application.purchase(
        PurchaseRequest(
            request_id="request-cookbook-001",
            resource_url=resource_url,
            purpose=purpose,
            idempotency_key="purchase-cookbook-001",
        ),
        approval=approval,
    )
    recorder.record(purchase)
    return AgentPurchaseEvidence(
        status=purchase.status,
        report=purchase.report,
        receipt_id=purchase.receipt.receipt_id,
        amount=purchase.receipt.amount,
        currency=purchase.receipt.currency,
        requires_human_approval=purchase.authorization.requires_human_approval,
    ).model_dump_json()


agent = Agent(
    name="Supplier research agent",
    model=model,
    model_settings=ModelSettings(reasoning=Reasoning(effort="low"), store=False),
    output_type=SupplierResearchOutput,
    tools=[x402_fetch],
    instructions=(
        "Call x402_fetch exactly once for the approved supplier report. "
        "Return only the typed summary supported by its receipt."
    ),
)

{"agent": agent.name, "tools": [tool.name for tool in agent.tools]}

4. 运行 x402 支付工作流

x402 使用 HTTP 402 Payment Required 响应来传达付费请求的条款。当应用首次请求供应商报告时,商户会返回一个包含金额、资产、网络、收款人和到期时间的挑战。

应用会根据其批准的策略检查这些条款。如果商户、价格、请求目的和人工批准均有效,AgentCore Payments 会在连接的 AWS 工作流中创建一个有界证明。本地示例使用合成证明来演示相同的顺序。应用使用该证明重试请求,商户返回 HTTP 200 和供应商报告。

The x402 request sequence between the agent, application, merchant, and AgentCore Payments*商家的402响应提供了支付条款。应用审批发生在请求支付证明之前。*

Runner.run 执行 Agents SDK 循环。脚本化模型在第一次轮次时请求 x402_fetch,并在工具返回后准备类型化的供应商摘要。应用将该摘要与实际记录的购买进行核对。

from agents import RunConfig, Runner

from agentic_commerce.agent import (
    DEFAULT_PROMPT,
    SupplierResearchRun,
    validate_supplier_research_output,
)

result = await Runner.run(
    agent,
    DEFAULT_PROMPT,
    max_turns=4,
    run_config=RunConfig(
        tracing_disabled=True,
        workflow_name="Synthetic Agentic Commerce",
    ),
)
purchase = validate_supplier_research_output(result.final_output, recorder)
run = SupplierResearchRun(output=result.final_output, purchase=purchase)

{
    "model_turns": model.model_turns,
    "tool_calls": model.requested_tools,
    "merchant_requests": merchant.request_count,
    "synthetic_charges": local_payments.charge_count,
    "model_inference_calls": 0,
    "aws_calls": 0,
    "value_transferred": False,
}

直接使用 Responses API

OpenAI Agents SDK 协调了前面示例中的工具循环,但支付工作流并不需要该编排层。Responses API 可以将 x402_fetch 暴露为一个函数,返回模型提出的参数,并在应用程序处理购买后接受 function_call_output。

应用程序在付款前仍会检查商家、用途、审批和支出限额。AgentCore Payments 仍负责在连接的 AWS 工作流中生成有界证明。下面的示例使用与本地演练相同的模拟商家和支付处理器。

此替代方案调用真实的 Bedrock 托管的 GPT-5.6 Sol 模型,因此仅在设置 ALLOW_PAID_INFERENCE=1 并配置有效 AWS 凭证时运行。它不会启用实时支付或转移资金。

import json
import os

from openai import OpenAI
from openai.providers import bedrock

responses_tool = {
    "type": "function",
    "name": "x402_fetch",
    "description": "Fetch one approved supplier report.",
    "parameters": {
        "type": "object",
        "properties": {
            "resource_url": {"type": "string"},
            "purpose": {"type": "string"},
        },
        "required": ["resource_url", "purpose"],
        "additionalProperties": False,
    },
    "strict": True,
}
responses_result = {
    "result": "SKIPPED",
    "reason": "Set ALLOW_PAID_INFERENCE=1 to call the Bedrock model.",
    "model_calls": 0,
    "live_payment_calls": 0,
}

if os.environ.get("ALLOW_PAID_INFERENCE") == "1":
    responses_client = OpenAI(
        provider=bedrock(
            region=os.environ.get("AWS_REGION", "us-east-2"),
            profile=os.environ.get("BEDROCK_AWS_PROFILE") or None,
            api_key=None,
        )
    )
    prompt = (
        f"Call x402_fetch for {RESOURCE_URL} with purpose "
        "supplier_due_diligence. After the tool returns, provide JSON "
        "containing supplier, receipt_id, amount, and currency."
    )
    first_response = responses_client.responses.create(
        model="openai.gpt-5.6-sol",
        input=prompt,
        tools=[responses_tool],
        tool_choice={"type": "function", "name": "x402_fetch"},
        parallel_tool_calls=False,
        reasoning={"effort": "low"},
        max_output_tokens=128,
        store=False,
    )
    requested_tools = [
        item for item in first_response.output if item.type == "function_call"
    ]
    if len(requested_tools) != 1 or requested_tools[0].name != "x402_fetch":
        raise AgentResultInvalid(
            "access_count_invalid", "Expected exactly one x402_fetch request."
        )

    arguments = json.loads(requested_tools[0].arguments)
    responses_now = datetime.now(UTC)
    responses_application, responses_merchant, responses_payments = build_demo(
        responses_now
    )
    responses_request_id = "request-responses-cookbook-001"
    responses_approval = approval.model_copy(
        update={
            "approval_id": "approval-responses-cookbook-001",
            "request_id": responses_request_id,
            "approved_at": responses_now,
            "expires_at": responses_now + timedelta(minutes=10),
        }
    )
    responses_purchase = responses_application.purchase(
        PurchaseRequest(
            request_id=responses_request_id,
            resource_url=arguments["resource_url"],
            purpose=arguments["purpose"],
            idempotency_key="purchase-responses-cookbook-001",
        ),
        approval=responses_approval,
    )
    responses_evidence = AgentPurchaseEvidence(
        status=responses_purchase.status,
        report=responses_purchase.report,
        receipt_id=responses_purchase.receipt.receipt_id,
        amount=responses_purchase.receipt.amount,
        currency=responses_purchase.receipt.currency,
        requires_human_approval=(
            responses_purchase.authorization.requires_human_approval
        ),
    )
    response_input = [
        {"role": "user", "content": prompt},
        *(
            item.model_dump(mode="json", exclude_none=True)
            for item in first_response.output
        ),
        {
            "type": "function_call_output",
            "call_id": requested_tools[0].call_id,
            "output": responses_evidence.model_dump_json(),
        },
    ]
    final_response = responses_client.responses.create(
        model="openai.gpt-5.6-sol",
        input=response_input,
        tools=[responses_tool],
        tool_choice="none",
        parallel_tool_calls=False,
        reasoning={"effort": "low"},
        max_output_tokens=256,
        store=False,
    )
    final_output = json.loads(final_response.output_text)
    expected_output = {
        "supplier": responses_purchase.report.supplier,
        "receipt_id": responses_purchase.receipt.receipt_id,
        "amount": str(responses_purchase.receipt.amount),
        "currency": responses_purchase.receipt.currency,
    }
    if any(final_output.get(key) != value for key, value in expected_output.items()):
        raise AgentResultInvalid(
            "agent_output_mismatch", "The model response did not match its receipt."
        )
    responses_result = {
        "result": "PASSED",
        "responses_api_calls": 2,
        "tool_calls": [requested_tools[0].name],
        "merchant_requests": responses_merchant.request_count,
        "synthetic_charges": responses_payments.charge_count,
        "receipt_verified": True,
        "live_payment_calls": 0,
        "value_transferred": False,
    }

responses_result

5. 验证供应商报告、收据和审计历史

模型返回结构化的供应商摘要,但有效的模式仅确保响应具有预期的形状。应用程序仍需检查所报告的收据、金额、货币和供应商详细信息是否与已完成的工具请求匹配。

第一个单元格将代理提议的结果与应用程序拥有的收据进行比较。第二个单元格显示从初始资源请求到授权、支付证明创建、商家重试和返回内容的审计序列。

proposal = run.output.model_dump(mode="json")
receipt = run.purchase.receipt.model_dump(mode="json")

{
    "typed_agent_proposal": proposal,
    "application_receipt": {
        "receipt_id": receipt["receipt_id"],
        "amount": receipt["amount"],
        "currency": receipt["currency"],
        "network": receipt["network"],
        "reused": receipt["reused"],
    },
}
[
    {
        "sequence": event.sequence,
        "event_type": event.event_type.value,
    }
    for event in run.purchase.audit_events
]

6. 检查被拒绝的购买和伪造的收据

一个有用的支付示例应该展示当其控制措施拒绝请求时会发生什么。第一次检查提交了另一个供应商报告请求,但没有所需的人工审批。应用程序可以检查商家的挑战,但必须在创建另一个支付证明或收费之前拒绝该购买。

在请求被拒绝后,现有的合成收费计数应保持不变。

from agentic_commerce.errors import PolicyDenied

blocked_request = PurchaseRequest(
    request_id="request-cookbook-002",
    resource_url=RESOURCE_URL,
    purpose="supplier_due_diligence",
    idempotency_key="purchase-cookbook-002",
)

try:
    application.purchase(blocked_request, approval=None)
except PolicyDenied as exc:
    blocked_result = {
        "status": "DENIED",
        "reason": exc.code,
        "synthetic_charges": local_payments.charge_count,
    }
else:
    raise AssertionError("Unapproved purchase unexpectedly succeeded")

blocked_result

第二个检查在原本有效的供应商摘要中更改了收据标识符。输出验证器将该响应与应用程序观察到的购买进行比较,并拒绝不匹配的情况。

这两个检查覆盖了不同的边界:支付前的授权和模型响应后的证据验证。

from agentic_commerce.agent import (
    PurchaseResultRecorder,
    validate_supplier_research_output,
)

fabricated = run.output.model_copy(update={"receipt_id": "receipt-fabricated"})
try:
    validate_supplier_research_output(
        fabricated,
        PurchaseResultRecorder(results=[run.purchase]),
    )
except AgentResultInvalid as exc:
    negative_result = {"status": "REJECTED", "code": exc.code}
else:
    raise AssertionError("Fabricated output unexpectedly passed validation")

negative_result

7. 将 AgentCore Payments 连接到 AWS 测试网

所连接的供应商研究工作流使用 Bedrock 托管的 OpenAI 模型、AgentCore Payments 以及 Base Sepolia 上经过批准的 x402 商户。该应用程序保持了与本地模拟中演示的相同的商户策略、支出审批、请求身份和收据检查。AgentCore Payments 会创建一个具有自身最大支出限额的短期会话。

Local supplier research and an AWS testnet request share the same application approval and spending controls*本地模拟和 AWS 测试网工作流共享相同的应用控制。AgentCore Payments 在连接的环境中提供有界支付会话。*

在启用 AWS 路径之前,操作员在笔记本之外配置支付管理器、连接器、工具和精确钱包签名权限。模型推理、会话管理和支付执行使用不同的职责。README.md 描述了设置、支持的配置、低于预算的拒绝检查以及清理过程。

此笔记本不使用 AgentCore Runtime。笔记本进程仍然是应用程序运行时,并且除非操作员明确启用,否则所有实时支付网关均被禁用。

在网络调用前检查测试网就绪状态

就绪检查在不联系 AWS、模型、商家或钱包的情况下检查本地配置。它确认所需的支付选择加入项已存在,AWS 角色已分离,会话预算有限,并且本地状态文件中没有记录支付会话。

该检查还会拒绝与商家连接的 DNS 和 TLS 保护不兼容的代理设置。使用默认配置时,预期结果为 NOT_READY,因为实时支付网关已禁用。

from agentic_commerce.agentcore_e2e import (
    readiness_report,
    run_managed_e2e,
)

agentcore_readiness = readiness_report()
agentcore_readiness

检查支付工具和网络

当显式启用 ALLOW_AGENTCORE_READ_ONLY=1 时,应用程序可以检查配置的支付工具,而无需创建支付会话。该工具必须处于 ACTIVE 状态,使用 ETHEREUM 钱包网络,并报告精确的 BASE_SEPOLIA 和 USDC 余额范围。

缺少门控返回 SKIPPED。网络或资产不匹配返回 NOT_READY。这些检查不会暴露余额金额、生成支付证明、联系商家或转移价值。

from agentic_commerce.agentcore_infrastructure import (
    inspect_payment_instrument,
)

agentcore_infrastructure = inspect_payment_instrument()
agentcore_infrastructure

运行一次已批准的测试网交易

操作员只有在检查了已批准的商户、收款方、资产、金额、钱包和短期会话配置后,才能启用受管理的测试网运行。该工作流要求所有五个显式门控:RUN_AGENTCORE_E2E、ALLOW_AGENTCORE_SESSION_ADMIN、ALLOW_PAID_INFERENCE、ALLOW_AGENTCORE_TESTNET 和 APPROVE_AGENTCORE_TESTNET_PURCHASE。

组合执行路径在开始模型推理或支付活动之前再次检查这些门控。它创建一个有界会话,通过应用程序策略发送代理请求,请求 AgentCore 支付证明,并重试一次已批准的商户。

会话创建是串行化的。如果本地会话持久化失败,应用程序会尝试补偿性清理。成功或失败的受管理运行也会在 finally 中尝试删除会话。如果清理失败,操作员必须在开始另一次运行之前遵循 README.md 中的恢复程序。

使用默认笔记本配置,单元格返回 SKIPPED 并且不执行任何实时操作。

import json

agentcore_e2e_result = await run_managed_e2e()
if agentcore_e2e_result["result"] not in {"PASSED", "SKIPPED"}:
    print(json.dumps(agentcore_e2e_result, indent=2, sort_keys=True))
    raise RuntimeError(
        "Managed AgentCore test did not complete safely; review the "
        "sanitized report before any rerun."
    )

agentcore_e2e_result

解读支付与结算证据

托管结果记录了模型是否完成了其工具调用、AgentCore Payments 是否生成了支付证明、商户是否返回了 HTTP 200,以及临时会话是否被删除。应用程序对原始 HTTP 响应字节进行哈希处理,并在报告中排除提供商凭据、支付头和内部标识符。

商户接受表明已支付的请求已完成。独立的结算和交易最终性需要单独的证据。因此,该示例报告 settlement_verified=false。

if agentcore_e2e_result["result"] == "PASSED":
    assert agentcore_e2e_result["model_run_completed"] is True
    assert agentcore_e2e_result["agentcore_payment_path_completed"] is True
    assert agentcore_e2e_result["merchant_paid_retry_completed"] is True
    assert agentcore_e2e_result["status_code"] == 200
    assert agentcore_e2e_result["payment_attempts"] == 1
    assert agentcore_e2e_result["session_cleanup"] == "DELETED"
    assert agentcore_e2e_result["identifiers_logged"] is False
    assert agentcore_e2e_result["settlement_verified"] is False

{
    "result": agentcore_e2e_result["result"],
    "session_cleanup": agentcore_e2e_result.get("session_cleanup", "NOT_CREATED"),
    "settlement_verified": agentcore_e2e_result.get("settlement_verified", False),
    "interpretation": (
        "Combined testnet path completed; settlement still needs separate evidence."
        if agentcore_e2e_result["result"] == "PASSED"
        else "Live path was not enabled; no testnet transaction was attempted."
    ),
}

可选的 CloudWatch 生命周期证据

此公开笔记本不包含 CloudWatch 截图。如需经过审查的实时运行,请在运行前配置 AgentCore Payments 日志交付,并仅将 CloudWatch 用作经过清理的生命周期证据。不要共享账户 ID、请求 ID、钱包数据、支付标识符或证明头。

生命周期事件有助于确认会话创建、支付处理和会话删除。它们不能独立证明支付金额、区块链结算、交易最终性或生产就绪性。

将支付工作流应用于其他服务

相同的结构可以支持其他付费数据源,包括高级搜索、制裁筛查、市场数据或供应商验证。每个集成都应定义其批准的商户、允许的用途、最大金额、审批要求以及会话时长,之后代理才能请求购买。

模型会选择付费工具何时可能有助于完成用户的任务。应用程序代码决定该请求是否被授权,记录由此产生的交易证据,并根据实际工具结果检查模型的最终响应。当工作流连接到批准的 AWS 测试网会话时,AgentCore Payments 提供了额外的支出边界。

参考资料

常见问题

常见错误包括:1) 未设置ALLOW_PAID_INFERENCE导致模型调用被跳过;2) 未配置AWS凭证导致AgentCore连接失败;3) 代理多次调用工具,违反一次性能力检查;4) 模型输出与收据不匹配,导致验证失败。排查方法:检查环境变量、使用readiness_report检查配置、查看审计事件序列。生产环境优化:使用CloudWatch监控生命周期事件,确保会话清理,分离IAM角色,并定期审计支付记录。

企业应用扩展

该工作流可应用于其他付费数据源,如高级搜索、制裁筛查、市场数据或供应商验证。每个集成需定义批准的商户、允许的用途、最大金额、审批要求和会话时长。模型选择工具,应用程序决定授权,AgentCore Payments提供支出边界。未来可扩展至多商户聚合、动态定价策略、更细粒度的审批流程,以及与其他支付协议(如Stripe)的集成。

总结

本文深入探讨了如何为AI代理构建安全的付费API访问机制,重点介绍了Amazon Bedrock AgentCore Payments与OpenAI Agents SDK的集成。通过一个供应商研究示例,详细阐述了x402支付协议、应用程序控制的支出审批流程、收据验证以及审计追踪。文章提供了完整的代码实现,包括本地模拟和AWS测试网连接,并强调了模型、应用程序、支付基础设施和商户之间的职责分离。读者将学习如何设置支付策略、定义代理工具、处理支付挑战,以及如何验证交易和防止滥用。

内容来源与验证

继续系统学习

站内还有更多完整实战教程

加入会员,解锁全部教程、完整代码与持续更新内容。

加入会员
查看全部API 与工程实践教程