Skip to main content

支持支付的 Agent：ACP、AP2、x402 与 MPP 在生产环境中的组合

这是一门关于四个协议的速成课。它们让 OpenAI Agents SDK 系统可以在商家处、针对 API、与其他 agent 之间，以及在开放经济中花钱。目标读者是已经交付过 agent、现在需要让 agent 付款的工程师。

19 个概念。5 个决策。3 张图。四条学习路径。 Reader 路径是 2–3 小时的纯阅读：四层 stack、每个协议的深入讲解、组合规则，不需要环境搭建。Beginner、Intermediate 和 Advanced 路径会逐步增加动手深度：把 agent 接到协议上，让组合 durable 运行，并管理身份、花费和争议。它们大约分别需要 1 天、2–3 天和 4–5 天。诚实估计：读完需要 2–3 小时；一个团队要把这套 stack 变成稳定工作习惯，需要 4–5 天。进入第 5 部分的 decision lab 之前，请先选择你的路径。

本课建立在一个核心判断上：四个协议不是竞争对手。它们是层。 许多文章会问「ACP 还是 x402？」这个问题把这些协议的性质问错了。一个 2026 年真正上线的系统通常会把其中几个协议组合起来，因为每个协议解决 agent-commerce 问题中的不同层。consumer shopping agent 在 commerce 层使用 ACP，在 settlement 层使用 card rails。API-paying agent 在两层都使用 x402，因为在 machine-to-machine micropayments 中，这两层会合并为一层。enterprise procurement agent 使用 AP2 mandates 证明人类授权了花费，然后在 settlement 层使用 Stripe MPP。你会学会读懂 use case，并选出正确的组合。

如果你还没学过其他 Agent Factory 课程

本课会提到几个相邻课程：Build AI Agents 速成课（SDK 基础）、Production Worker 速成课（用 Inngest durable 地运行 agent）、Eval-Driven Development 速成课，以及 Choosing Agentic Architectures 速成课。没学过也可以读这门课。四个协议、分层、SDK 代码和决策纪律本身都能独立成立。唯一有帮助的前提是：第 3 部分的代码假设你能读懂 Agent、Runner.run() 和 @function_tool。如果这些还陌生，先快速浏览 Build AI Agents 速成课 或 OpenAI Agents SDK 文档，再回来继续。

如果你使用不同 stack，每个协议对应什么​

如果你不是用 OpenAI Agents SDK 加 Stripe、Coinbase 和 Cloudflare，这张表会把每个 reference implementation 映射到常见替代方案。协议 spec 与 stack 无关；只有 primitive 名称会变。

协议主要 reference SDK（2026）常见替代方案许可证 / 治理
ACP（Agentic Commerce Protocol）Stripe SDK（stripe）加 OpenAI Agents SDK；PayPal ACP server tools；Worldpay credentialsAdyen ACP、Shopify-native checkout APIApache 2.0，OpenAI 加 Stripe，位于 github.com/agentic-commerce-protocol/agentic-commerce-protocol
AP2（Agent Payments Protocol）Google ADK 加 Python、TypeScript、Kotlin、Go 的 reference implementationsLangGraph 加 custom mandate signing、AutoGen 加 a2a-x402Apache 2.0，Google 加 60+ partners，位于 github.com/google-agentic-commerce/AP2
x402x402-client（Python）、@x402/client（JS/TS）、Coinbase Developer Platform、Cloudflare withX402Client、AgentPay MCPDirect EIP-3009 implementations；Lobster.cash；Crossmint agent walletsApache 2.0，由 Coinbase 创建，现在由 Linux Foundation 的 x402 Foundation 治理
MPP（Machine Payments Protocol）Stripe PaymentIntents 加 MPP extensions；Tempo blockchain SDKDirect Lightning Network；Tempo native SDKsApache 2.0，Stripe 加 Tempo；spec 位于 mpp.dev
A2A（Agent2Agent，AP2 扩展的层）Google ADKCustom A2A implementationsApache 2.0，Google 加 Linux Foundation
MCP（Model Context Protocol，discovery layer）Anthropic MCP servers and clients；openai-agents MCP supportLangChain MCPMIT，Anthropic

这张表的读法是：当课程说「把 agent 的 @function_tool 接到 Stripe ACP endpoint」时，如果你的 stack 是 Adyen 加 LangGraph，就读作「把等价的 LangGraph tool 接到 Adyen ACP endpoint」。论证相同，名称不同。读这门课不需要先学 Stripe stack。映射 primitive，遵循 framework，再应用到你自己的 stack。

术语表​

📖 本课使用的术语，第一次阅读建议展开，后面需要时回来查

四个 headline protocols

  • ACP（Agentic Commerce Protocol）。 consumer-shopping protocol，由 OpenAI 和 Stripe 构建。它管理 agent 如何代表一个人在真实商家完成 checkout。ChatGPT Instant Checkout 就基于它。主要位于 commerce layer。Apache 2.0。
  • AP2（Agent Payments Protocol）。 authorization protocol，由 Google 与 60+ partners 构建。它生成签名的「mandates」，证明人类允许 agent 花钱。它本身不移动资金；它证明这笔花费经过授权。Apache 2.0。
  • x402。 HTTP-native settlement protocol，由 Coinbase 构建，现在由 Linux Foundation 治理。它复活了未被使用的 HTTP 402「Payment Required」状态码，让 agent 能用 stablecoin 在 1–2 秒内为一次 API call 付款。Apache 2.0。
  • MPP（Machine Payments Protocol）。 Stripe 和 Tempo 的 settlement protocol。它的关键招式是「session」：agent 先预授权一个 spending cap，然后在 cap 内持续流式支付许多小额款项。multi-rail（stablecoin、Lightning、cards）。Apache 2.0。

这些层（整门课的脊柱）

  • Discovery layer。 agent 找到自己可以买什么的层。答案可能来自 MCP servers、A2A、agent directories 或 AI shopping surfaces。它回答的问题是：「有什么可买？」
  • Authorization layer（也叫 Identity and Authorization）。 在资金移动前证明两件事的层：人类允许这次花费，并且 agent 确实是它声称的那个 agent。它回答的问题是：「我是否被允许花这笔钱？」
  • Commerce layer。 执行完整购买的层：cart、checkout、fulfillment、dispute、refund。它回答的问题是：「完整购买 lifecycle 是什么？」普通 API call 会完全跳过这一层。
  • Settlement layer。 资金真正换手的层。它回答的问题是：「钱到底在哪里移动？」
  • Settlement rail（或 rail）。 资金实际通过的管道：card networks（通过 Stripe 的 Visa/Mastercard）、区块链上的 stablecoin、bank transfer（ACH/SEPA）或 Lightning。「Pick a rail」的意思是「选择钱如何移动」。

授权 primitives

  • Mandate（AP2）。 一份签名的数字证明，表示人类授权了某类具体花费。AP2 有三种：Intent、Cart 和 Payment（见下文）。它们共同形成一条之后可以审计的链。
  • Intent Mandate。 第一份 mandate，在 agent 开始前由用户签署：任务规则。示例：「买 $120 以下的鞋」。它设置 agent 必须遵守的边界。
  • Cart Mandate。 中间的 mandate，在 agent 建好具体 cart 后由用户签署：「是的，就是这些商品、这个价格。」用于有人类在场审批的 flow。
  • Payment Mandate。 最后一份 mandate，在付款时签署（或根据 Intent Mandate 自动生成）：「授权在这条 rail 上完成这笔具体付款。」
  • SPT（Shared Payment Token）。 ACP 的 primitive。由 payment processor（Stripe）发出的一次性 token，锁定到一个 merchant、一个 amount cap 和一个短时间窗。如果 agent 被授权花 $50，却试图花 $1,000，SPT 会直接失败。它是 AP2 Payment Mandate 在 card rail 上的近亲。
  • Non-repudiable。 签署者之后不能否认自己签过的记录。AP2 mandate chain 是 non-repudiable 的：「我从未授权」无法对抗用户自己的 cryptographic signature。

Settlement 和 crypto primitives

  • Stablecoin。 锚定稳定价值的 cryptocurrency，通常锚定 1 美元。agent 使用它，是为了让「$0.05 payment」在发送和结算之间仍然价值 5 美分。
  • USDC。 大多数 x402 payment 使用的美元锚定 stablecoin。1 USDC 设计上等于 1 美元。由 Circle 发行。
  • HTTP 402 Payment Required。 1997 年起保留但长期未使用的 HTTP status code，直到 x402 重新启用。server 回复「402」以及 payment requirements；client 附带签名的付款证明重试。
  • EIP-3009（transferWithAuthorization）。 x402 构建其上的 Ethereum 标准。它让 buyer 离线签署付款，由其他人提交到链上，因此 buyer 不需要自己付 gas fee，也不需要直接接触区块链。
  • Facilitator（x402）。 可选第三方，替 merchant 检查签名并提交链上付款，这样 merchant 不必运行自己的区块链 plumbing。Coinbase 和 Cloudflare 都运行 facilitators。
  • CAIP-2。 一种标准化命名区块链的方式，让协议保持 chain-agnostic。x402 用 CAIP-2 形式写 chain。
  • EIP-155 / chain id。 CAIP-2 内部用于 Ethereum-style chains 的编号方案。eip155:8453 表示 Base chain；这个数字就是 chain id。（你会在 x402 代码中看到 eip155:8453；它只是表示「Base」。）
  • Smart-contract wallet。 一种 crypto wallet，它的花费规则（per-transaction cap、daily cap、per-recipient cap）由区块链上的代码本身强制执行。因为是 chain 执行这些限制，即使 agent 自己的代码失控，这些 cap 仍然有效。
  • MPP session。 MPP 的标志性动作。agent 不为每笔付款单独签名，而是打开一个带 spending cap 和 time limit 的「session」，然后在关闭前持续用它结算许多小额 metered payments。可以把它想成「agent 的预付标签页」。
  • Sessions model（MPP）。 上述模式的通称：预授权一个 cap 和 duration，然后对许多小额收费计量。调用频繁时，它比为每一笔 micropayment 都签名更便宜。

Commerce 和资金概念

  • Settlement。 资金实际从 buyer 移动到 seller 的时刻。在 settlement 之前的一切都只是编排；只有 settlement 完成，交易才真正完成。
  • Merchant of record（MoR）。 在法律上对交易负责的业务主体：处理税务、争议和客户支持。在 ACP 中，merchant 仍然是 MoR。在普通 machine-to-machine x402 call 中，通常没有 MoR。
  • Chargeback。 buyer 的银行撤销一笔 card payment，通常发生在争议之后。card rails 支持 chargebacks；纯 x402 不支持。是否需要 chargeback 往往决定你使用哪个协议。
  • Dispute。 buyer 对收费提出正式挑战（「我没有授权这笔交易」/「商品从未送达」）。不同协议用不同方式处理 disputes，这种差异常常会迫使协议选择。
  • Idempotency。 同一操作执行两次，效果与执行一次相同的性质。它很重要，因为 Stripe 会用同一个 event id 重试 webhooks；没有 idempotency key，你的代码可能会退款两次或收费两次。

Agent stack 和相邻协议

  • Agent commerce。 autonomous AI agent 作为 buyer、seller 或两者之一参与的任何交易，并且当下没有人类点击购买。它不同于「AI-assisted」shopping，后者仍然由人按下按钮。
  • OpenAI Agents SDK。 用于构建 agent loop 的 Python/JavaScript toolkit，包含 Agent、Runner.run()、@function_tool 和 guardrails。在本课中它是「universal client」：每个 payment protocol 都会变成 agent 可以调用的一个或多个工具。
  • MCP（Model Context Protocol）。 Anthropic 用于向 agent 暴露工具和上下文的开放标准。在 agent commerce 中，它通常是 discovery layer：agent 通过 MCP servers 找到可以买的服务。MIT licensed。
  • A2A（Agent2Agent）。 Google 让 agent 互相通信和发现彼此的协议。AP2 构建在 A2A 之上：mandate 作为 A2A message 传递。Apache 2.0。
  • UCP（Universal Commerce Protocol）。 Google 的 commerce-layer protocol，是 ACP 的 peer，围绕 Google 的 shopping surfaces（Gemini、Google AI Mode）构建。它在 commerce layer 与 ACP 竞争。
  • TAP（Trusted Agent Protocol）。 Visa 和 Cloudflare 用于在 HTTP request headers 中证明 agent 身份的协议。它验证 agent 是谁，而不是它被允许花多少钱，所以通常是叠加到另一个 auth protocol 上，而不是替代它。
  • ERC-8004。 agent identity 和 reputation 的 on-chain standard：一个公开的 agent registry，加上 agent 的交易历史，让没有既有关系的 agent 可以在信任对方之前检查其记录。
  • tool_input_guardrail。 OpenAI Agents SDK 的 guardrail，在 tool 执行之前运行并可以拒绝调用。它是 SDK-native 的「付款发生前阻止付款」方式。本课的脊柱：它出现在 7 个付款工具中。（对比 output_guardrail，后者运行在 agent 的最终回复上，太晚，无法阻止付款。）

先修条件​

如果你具备下面这些，本课收益最大：

  1. Build AI Agents 速成课，或等价 SDK 经验。 protocol integrations 会以 SDK code 展示，所以你需要能顺畅阅读 Agent、Runner.run() 和 @function_tool。参见 Build AI Agents 速成课。
  2. Choosing Agentic Architectures 速成课，或等价设计判断。 第 5 部分的「哪个 use case 该用哪个 composition」framework 建立在这种 pattern-selection 纪律之上。参见 Choosing Agentic Architectures 速成课。
  3. 基本 HTTP。 status codes、request/response cycle、headers。x402 尤其是在 HTTP 层工作。
  4. 基本支付词汇。 Merchant、settlement、dispute、chargeback。本课会解释 agent-specific 的部分，但假设你已经知道「merchant of record」是什么。

你不需要 blockchain 或 smart-contract 经验（本课会教足够的 x402 和 EIP-3009，让你能跟上；不需要 Solidity），也不需要之前接触过四个协议中的任何一个。它们都会从 primary sources 开始讲。

四条学习路径​

本课有四种深度。请在第 5 部分之前选择你的路径。

Track时间你会做什么最适合谁
Reader2–3 小时阅读所有 Concepts 和 Decisions；跳过运行代码。正在判断是否继续深入投入的工程师；需要 framework 来评估 vendor proposal 的 PM 和 architects。
Beginner约 1 天Reader 加运行 x402 client examples，并在 Stripe test mode 中完成一笔 ACP test transaction。刚接触 agent commerce、想动手体验最简单协议（x402）和最 production-ready 协议（ACP）的工程师。
Intermediate2–3 天Beginner 加构建一个 agent：一种交易用 ACP，另一种用 x402，并接上 AP2 Intent Mandate checks。正在交付真实系统、必须组合多个协议的工程师。
Advanced4–5 天Intermediate 加 durable 地运行组合系统（Inngest envelope），接 spend limits 和 human-approval gates，度量 trace、cost 和 dispute metrics，并处理完整 refund cycle。负责让生产系统为真实用户移动真实资金的工程师。

一个有用的自检问题是：「我心里的 use case，能交付价值的最小 protocol composition 是什么？」如果读完第 4 部分还答不上来，重读第 4 部分。如果能答上，你的路径只是在「最小 composition」和「production-grade composition」之间走多远的问题。

四层 stack​

这是整门课挂靠的图。

agent commerce 的四层 stack，自上而下。Layer 1 Discovery：agent 使用 MCP、A2A 和 agent directories 找到可用内容。Layer 2 Identity and Authorization：agent 使用 AP2 mandates、ACP shared payment tokens、TAP 和 ERC-8004 证明自己被允许花钱。Layer 3 Commerce：cart、dispute、refund 的完整购买 lifecycle，使用 ACP、UCP 或 direct API。Layer 4 Settlement：资金真正移动，使用 x402、MPP、card rails，或 bank 和 Lightning。右侧有两个示例路径：consumer shopping agent 运行全部四层；API-paying agent 跳过 commerce，并收缩为 discovery 和 settlement。协议是层，不是替代品。

每个 agent-commerce use case 都会触碰四层，但不同 use case 会在每层组合不同协议。consumer shopping agent：MCP 做 discovery，ACP shared payment token 做 authorization，ACP 做 commerce，card rails 做 settlement。API-paying agent：agent directory 做 discovery，EIP-3009 signature 做 authorization，完全没有 commerce layer，x402 做 settlement。enterprise procurement agent：A2A directory、AP2 mandate、ACP 或 UCP 做 commerce、Stripe MPP 做 settlement。规则很简单：每层选一个协议，并让 use case 为每个选择负责。

第 1 部分：为什么 agent commerce 需要新协议​

第 1 部分轻读可以吗？

第 1 部分快速引入六个名字（ACP、AP2、x402、MPP、MCP、A2A）和三层。这是有意的：这一部分是 brief，不是 deep dive。你至少要记住：ACP 是 consumer shopping（Stripe 加 OpenAI），AP2 是 authorization mandates（Google 加 60 partners），x402 是 HTTP 上的 per-request stablecoin payment（Coinbase），MPP 是 session-based multi-rail payment（Stripe 加 Tempo）。MCP 和 A2A 是 agent 发现彼此、互相通信的方式，不是 payment protocols。先宽松地记住名字。第 2 部分会逐一深入，第二遍自然会牢。

概念 1：坏掉的假设​

一句话： payment systems 默认人类在点击 buy，而 agent 会同时从三个方向打破这个假设。

Payment systems 建立在一个安静的假设上：键盘前有一个人，在点击 buy。每个屏幕、每个 fraud check、每个 dispute process、每个 signup form 都是为人设计的。AI agents 会同时从三个方向打破这个假设。

Break 1：agent 没有 email address。 consumer payment flow 想要 account。account 想要 email、phone number，通常还要 name。autonomous agent 没有这些东西。伪造这些字段，就会创建一个在 fraud detection 一看就无法通过 KYC 的实体。signup flow 假设是人类在申请一种关系；agent 需要别的东西。

Break 2：agent 每秒可以行动上千次。 fraud detection 会按 rate、location 和 pattern 标记异常行为。agent 在一分钟内发起 1,000 个 API calls，看起来就像 credential-stuffing attack。对 agent 正常的行为，对人类 rails 来说是警报；而这些 rails 是为人调校的。

Break 3：agent 不会接电话。 dispute resolution 假设能联系到 buyer：「这笔是你授权的吗？」授权了一笔 charge 的 agent 无法回答，背后的人类甚至可能不知道这笔 charge 已经发生。当 buyer 是软件时，dispute 需要不同模型。

每一种 break 都需要 protocol level 的修复，而不是给 UI 刷一层漆：

Breakhuman rail 的修复做不到什么agent rail 的修复给你什么
No email or account强迫 agent 创建 fake accountsCryptographic identity（TAP、ERC-8004）或 scoped tokens（ACP SPT、AP2 Mandate）
High-frequency behavior阻断看起来像 attack 的 trafficHTTP-native per-request payment（x402）或 pre-authorized sessions（MPP）
No phone for disputes给 agent 读不了的 disputes 发 emailMandate-based authorization（AP2）和 non-repudiable audit trail

「把旧支付包进更好的 agent UX」这条路已经失败了。2024 和 2025 年，一些 startup 试过：给 agent 类人的账号和编造的身份。fraud detection 抓住它们，chargebacks 堆起来，merchant relationships 崩掉。结果证明，protocol-level fixes 是必需的，不是可选项。这也是为什么 ACP、AP2、x402 和 MPP 都在同一个 12 个月里出现。

概念 2：为什么一个协议不可能赢​

一句话： 这些 break 发生在四个不同层，每层都有不同 incumbents，所以协议按层分工，而不是由一个协议吞掉全部。

一个合理问题是：如果四个协议都出现来修复同样的三个 break，为什么不是其中一个直接胜出？答案是结构性的。break 发生在不同层，而试图修复全部层的单一协议会大到没有人愿意采用。

想想一个统一协议必须规定什么：

  1. agent 如何找到可用 merchant 和 services。这是 discovery layer。
  2. agent 如何证明自己是谁，以及人类授权了这笔花费。这是 authorization layer。
  3. agent 如何执行完整购买，包括 disputes 和 refunds。这是 commerce layer。
  4. 钱如何在各方之间真正移动。这是 settlement layer。

每一层已经有强大的 incumbents。Discovery 属于 search engines 和 APIs。Identity 属于 OAuth 和 certificate authorities。Commerce 属于 Stripe、Adyen 和 Shopify。Settlement 属于 Visa、Mastercard 和 ACH，再加上新的 crypto rails。一个统一协议需要每一层的每个 incumbent 都同意它。这从来都不现实。

最终发生的是：每个协议都选择了 sponsor 最有 leverage 的那一层：

Protocolsponsor 拥有 leverage 的地方它占据的层
ACPOpenAI 拥有 shopping channel（ChatGPT）；Stripe 拥有 merchant integrationCommerce，用于 human-buyer-via-AI flows
AP2Google 拥有 Android wallets 和 60-partner coalitionAuthorization，把 mandates 作为 signed credentials
x402Coinbase 拥有 stablecoin infrastructure；Cloudflare 拥有 HTTP edgeSettlement，用于 machine-to-machine micropayments
MPPStripe 拥有 merchant relationships；Tempo 拥有 blockchainSettlement，用于 enterprise 和 multi-rail flows

所以这些协议不是在一个层内互相打架；它们是在竞争定义某一层。在 settlement 层，x402 和 MPP 确实竞争，而多数「x402 vs MPP」文章漏掉了：这才是它们真正重叠的唯一地方。在 commerce 层，ACP 和 UCP 竞争。在 authorization 层，AP2、TAP 和 ERC-8004 竞争。

这给了我们整门课所依托的核心想法。同一层内，选择一个协议。跨层时，组合多个协议。 四个协议不是让你二选一的 alternatives；它们是要叠起来的 layers。consumer shopping agent 在 commerce 层运行 ACP，在 settlement 层运行 card rails。API-paying agent 在 settlement 层运行 x402，并完全跳过 commerce。enterprise agent 在 authorization 层运行 AP2 mandates，在 settlement 层运行 MPP。第 5 部分的 decision tree 会逐层走完。请记住这一点：后面的每一节都以它为前提。

概念 3：OpenAI Agents SDK 作为 universal client​

一句话： 你不会用四种方式分别对接四个协议；每个协议都会变成 agent 调用的一个工具，而 SDK 是把它们全部接起来的单一 client。

一个实际问题是：既然有四个层、四个协议，agent 到底如何使用它们？2026 年的答案是：agent 的 framework 会成为 universal client。每个协议暴露 SDK 或 HTTP endpoint，而 framework 把它接成 tool。OpenAI Agents SDK、LangGraph、AutoGen 和 CrewAI 都是这个形状。本课全程使用 OpenAI Agents SDK。

对 SDK 来说，每个 protocol integration 都遵循同一个形状：把协议包装成一个或多个 @function_tool 函数，交给一个 Agent，再让 Runner.run 驱动 loop。

note

下面的 stripe.PaymentTokens.create、X402Client(wallet=...) 和 from ap2 import ... 调用都是 illustrative：它们展示 protocol integration 的形状。围绕它们的 Agent、Runner 和 @function_tool scaffolding 是真实的，并且可以运行。代码块之后会说明哪些是 stand-in。

from agents import Agent, Runner, function_tool
import stripe  # for ACP/MPP
from x402_client import X402Client  # for x402
from ap2 import IntentMandate, CartMandate  # for AP2
from decimal import Decimal
from .models import PaymentToolResult, X402PaymentResult  # the shared result models

# Each protocol becomes one or more @function_tool decorated functions
@function_tool
async def acp_checkout(merchant_id: str, items: list, max_amount: Decimal) -> PaymentToolResult:
    """Complete an ACP checkout at a merchant with the given items."""
    # Mint a one-time payment token scoped to this merchant, then POST the order
    spt = stripe.PaymentTokens.create(
        amount=int(max_amount * 100),   # cents as int
        currency="usd",
        merchant_id=merchant_id,
        max_uses=1,
    )
    response = await acp_post(merchant_id, items, spt.token)
    return PaymentToolResult(
        status="success" if response.status == "confirmed" else "failed",
        details={"order_id": response.order_id, "merchant_status": response.status},
    )

@function_tool
async def x402_fetch(url: str, max_payment_usdc: Decimal) -> X402PaymentResult:
    """Fetch a URL that may require x402 payment up to max_payment_usdc."""
    client = X402Client(wallet=agent_wallet, max_per_request=max_payment_usdc)
    response = await client.get(url)
    return X402PaymentResult(
        content=response.content,
        amount_paid_usdc=response.amount_paid_usdc,
        tx_hash=response.payment_proof,
    )

# Compose the tools into an agent
shopping_agent = Agent(
    name="ShoppingAgent",
    instructions="Help the user find and purchase items. Use acp_checkout for retail goods, x402_fetch for paid APIs.",
    tools=[acp_checkout, x402_fetch],
    model="gpt-5.5",
)

# Run the agent. The SDK handles tool selection, the loop, and retries.
result = await Runner.run(shopping_agent, "Buy me a red t-shirt under $30")

这里真正能运行的部分是：Agent、Runner.run 和 @function_tool wiring，它们就是真实 SDK，按写法可以工作。payment clients 是 stand-ins。stripe.PaymentTokens.create 不是一个真实 Stripe 调用（production ACP 会通过 live Stripe ACP endpoints 集成），而丰富的 X402Client(wallet=..., max_per_request=...) constructor 也是 illustrative。真实 buyer-side package 是 x402-client，live call 需要一个有资金的 account 和一个真实 402 endpoint，本课不会移动真实资金。第 3 部分会为每个协议给出 runnable mock backend，让你在不移动真钱的情况下看到 harness 端到端工作。

四个协议的形状完全相同。SDK 是 universal client；每个协议只是 agent 能推理并在需要时调用的 tool。SDK 结构中有三部分对支付尤其重要：

  1. 支付结果的 typed return value。 当 payment tool 返回 Pydantic model 时，agent 的 reasoner 会获得清晰的类型信息：什么成功、什么失败、下一步该做什么。第 3 部分会一次性定义这些 shared result models。
  2. Runner.run(..., context=...) 用于 payment context。 agent 常常需要用户身份、spending limits 和 wallet handle。请通过 SDK 的 context 参数传入，而不是烘进 instructions。context 是 per-run、per-user 的。
  3. tool_input_guardrail 用于 spend limits。 tool input guardrail 在每个 tool 执行前运行，可以拒绝调用。这是 SDK-native 的「在付款发生前阻止付款」方式，而不是事后阻止。概念 15 会讲完整的三层 enforcement。agent-level output_guardrail 解决不了这个问题，因为它在 agent 最终回复时才触发，而此时任何 payment tool 都已经运行过。

第 2 部分：四层深入​

第 1 部分已经见过四层；这里逐一近看。每一层回答一个不同问题，有自己的 competing protocols，并迫使你做一个决定：对这个 use case 来说，哪个协议最适合这一层？请先读第 2 部分，再读第 3 部分。这里逐层 framing，正是让第 3 部分的 protocol details 读起来像一个整体，而不是一串 rivals 的原因。

概念 4：Layer 1，Discovery（agent 如何找到可以买什么）​

一句话： Discovery 是 agent 查明到底有哪些东西可以买的地方，而正确机制取决于这些服务实际在哪里。

agent 交易之前，必须先发现外面有什么。shopping agent 需要找到卖相关商品的 merchants。API-paying agent 需要知道哪些 endpoints 有数据、价格多少。procurement agent 需要找到符合 compliance rules 的 suppliers。Discovery 回答「有什么可用？」

2026 年有四个严肃选项在这里竞争：

Protocol工作方式最适合
MCP（Anthropic）Tool servers 暴露可调用函数；agent 连接、列出 tools 并调用对开发者已经接入的具体服务进行 programmatic access；高频 agent work；主流 agent-tooling discovery layer
A2A（Google）agents 用标准 envelopes 发布自己提供的能力；其他 agents 发现它们agent 需要寻找 peer agents 的 multi-agent ecosystems；AP2 扩展的 discovery layer
Agent directories（Agent.market、lobster.cash、Tenzro）列出 paid APIs 和 services 的公开 marketplaces；agent 像查 catalog 一样查询runtime 中发现的 third-party services；agent commerce 的「Yellow Pages」
AI shopping surfaces（ChatGPT Instant Checkout、Google AI Mode、Walmart-in-ChatGPT）consumer AI products 内置 product discovery 加 ACP checkout用户正在与 AI 对话、AI inline 展示产品的 consumer flows

SDK 对 MCP 有 first-class support：几行代码把 MCP server 接到 agent 上，它的所有 tools 就能被 agent reasoning 使用。对于非 MCP discovery，把它包装成普通的 @function_tool，查询 directory 并返回 structured listings。

note

下面的 MCPServerStreamableHttp wiring 真实且可 import。agent_market_client 调用是 illustrative：它代表 directory client。@function_tool 和 Agent scaffolding 是真实的。

from agents import Agent, function_tool
from agents.mcp import MCPServerStreamableHttp
from decimal import Decimal

# Wire an MCP discovery server: all its tools become available
research_mcp = MCPServerStreamableHttp(
    name="research-services",
    params={"url": "https://research-services.example.com/mcp"},
)

# Wire a non-MCP directory as a regular tool
@function_tool
async def search_agent_market(query: str, max_price_usdc: Decimal) -> list[dict]:
    """Search Agent.market for x402-paid services matching the query."""
    return await agent_market_client.search(query, max_price_usdc=max_price_usdc)

agent = Agent(
    name="ResearchAgent",
    instructions="Find and use research services. Prefer MCP-discovered tools; fall back to Agent.market for niche needs.",
    mcp_servers=[research_mcp],
    tools=[search_agent_market],
)

你在这里做的选择不是「MCP vs A2A vs directories」。真正的问题是：我的 agent 需要的服务实际在哪里？在组织内部，用 MCP。跨 partner agents 的 network，用 A2A。runtime 中发现第三方 APIs，用 directories。consumer products，用 AI shopping surface（并在 commerce layer 使用 ACP）。这些不是互斥关系；真实 agent 往往会同时使用几个。

概念 5：Layer 2，Identity and Authorization（证明 agent 被允许）​

一句话： Authorization 是 agent 在任何资金移动之前，证明人类允许这笔花费、且自己确实是它声称的 agent 的地方。

资金移动前，必须同时满足两件事：agent 是它声称的那个 agent，人类授权了这笔花费。这是两个不同问题，有不同解决方案，Layer 2 会在 settlement 发生前解决两者。跳过 Layer 2 会导致两种失败之一：fraud（任何人的 agent 都能花任何人的钱）或 paralysis（每笔交易都需要人点击确认）。

这是 2026 年竞争最激烈的一层。四个选项，四种哲学：

Protocol工作方式最强场景
AP2 Mandates（Google）Signed credentials：Intent Mandate（「买 $120 以下的鞋」）、Cart Mandate（「这个 cart、这个价格」）、Payment Mandate（「授权这条 rail」）需要 non-repudiable consent proof 的 audit-heavy flows；merchant 之前没见过 buyer agent 的 multi-agent flows
ACP SPT（OpenAI 加 Stripe）Stripe mint 一个 Shared Payment Token，scope 到一个 merchant、amount 和 time window；agent 出示它；merchant 验证并收费Stripe 是 processor 且 card rails 承载 chargeback discipline 的 consumer shopping
TAP（Visa 加 Cloudflare）agent 的 identity signature 放在 HTTP headers 中；merchants 通过 Visa directory 验证专门做 identity verification（不是 authorization）；通常叠加到另一个 auth protocol 上，不单独使用
ERC-8004 plus on-chain reputationagent identities 和 transaction history 的 on-chain registry，并用过去交易形成 reputation score没有既有信任的纯 multi-agent flows；值得检查 reputation 的 high-stakes B2B

Layer 2 必须回答两个问题，以及每个协议如何回答：

  1. 「人类授权了吗？」 AP2 用用户在委托前签署的 mandate 回答。ACP 用 Stripe 在用户于 account level 授权后 mint 的 SPT 回答。TAP 不回答这个问题；它只处理 identity。ERC-8004 用签名的 on-chain transactions 回答。
  2. 「agent 是它声称的那个 agent 吗？」 AP2 用 signing key 回答（只有真实 agent 能签名）。ACP 用 SPT 的 merchant-scoped 属性回答（只有被授权 merchant 能 redeem）。TAP 用 Visa directory lookup 回答。ERC-8004 用 on-chain identity record 回答。

SDK 给你两个 integration points：一个在 payment tool 之前运行的 tool input guardrail，以及把 per-user state 带进两者的 run context。

note

下面的 guardrail、@function_tool 和 Agent wiring 都是真实 SDK，可以运行（attribute paths data.context.tool_arguments 和 data.context.context 已按已安装 SDK 确认）。tool 内部的 stripe.PaymentTokens.create 调用是 illustrative；production ACP 使用 live Stripe ACP endpoints。

from agents import Agent, function_tool, RunContextWrapper
from agents.tool_guardrails import (
    tool_input_guardrail,
    ToolInputGuardrailData,
    ToolGuardrailFunctionOutput,
)
from decimal import Decimal
import json
import stripe

# Pattern 1: a tool input guardrail. Runs BEFORE the payment tool executes.
# This is the SDK-native way to block a payment before it happens.
@tool_input_guardrail
def block_over_user_cap(data: ToolInputGuardrailData) -> ToolGuardrailFunctionOutput:
    """Reject any payment tool call where the request would exceed the user's per-run cap."""
    args = json.loads(data.context.tool_arguments or "{}")          # raw JSON args -> dict
    requested = Decimal(str(args.get("max_amount_usd", 0)))
    ctx = data.context.context                                       # the run context (a dict)
    user_cap = Decimal(str(ctx["user_session"].per_run_spend_cap_usd))
    run_spent = Decimal(str(ctx.get("run_spend_usd", 0)))
    if run_spent + requested > user_cap:
        return ToolGuardrailFunctionOutput.reject_content(
            f"Refusing payment tool: would spend ${run_spent + requested}, exceeds run cap ${user_cap}"
        )
    return ToolGuardrailFunctionOutput.allow()

# Pattern 2: the payment tool itself, guarded at the function-tool level
@function_tool(tool_input_guardrails=[block_over_user_cap])
async def purchase_with_acp(
    ctx: RunContextWrapper,
    merchant_id: str,
    items: list,
    max_amount_usd: Decimal,
) -> PaymentToolResult:
    """Use ACP to buy items from the merchant up to max_amount_usd.
    The guardrail has already verified spend is within bounds before we reach here."""
    user_session = ctx.context["user_session"]
    spt = stripe.PaymentTokens.create(
        amount=int(max_amount_usd * 100),   # Stripe expects cents as int
        currency="usd",
        merchant_id=merchant_id,
        user_session_id=user_session.id,
        max_uses=1,
    )
    response = await acp_post(merchant_id, items, spt.token)
    return PaymentToolResult(
        status="success" if response.status == "confirmed" else "failed",
        details={"order_id": response.order_id, "merchant_status": response.status},
    )

agent = Agent(
    name="ShoppingAgent",
    instructions="Help the user shop. Always verify authorization before any purchase.",
    tools=[purchase_with_acp],
    # No output_guardrails for spend control. Those run on the agent's FINAL output,
    # not on individual tool calls. See Concept 15 for the full three-level enforcement.
)
哪个 guardrail 能阻止付款

SDK 有三种 guardrail types，它们触发的时机不同。input_guardrail 运行在第一个 agent 的 initial input 上。output_guardrail 运行在最终 agent 对用户的 response 上。tool_input_guardrail 和 tool_output_guardrail 会在每一次 function-tool call 前后运行。对于 payment safety，你需要 tool input guardrail：它在 payment tool 运行前触发，并可以拒绝调用。output guardrail 太晚，无法阻止付款；它适合清理最终回复（例如 redact sensitive data），不适合阻止 tool。概念 15 会回到这一点；这是最常见的错误。

这里的选择取决于你的 trust model。如果用户已登录你的 app，并且你能通过 Stripe mint SPT，ACP 给出最 production-ready 的方案。如果你需要 non-repudiable audit trails，比如 regulated industries 或 B2B procurement，AP2 mandates 更合适。如果你需要单独的 cryptographic identity，而不是 authorization，请加 TAP。在没有 shared trust 的纯 multi-agent 设置中，ERC-8004 填补空缺。这些协议不可互换。

概念 6：Layer 3，Commerce（完整购买 lifecycle）​

一句话： Commerce 是真实购买中不属于 authorization 或 settlement 的所有内容：cart、order、fulfillment、dispute、refund；普通 API call 会完全跳过它。

Authorization 加 settlement 覆盖的是「资金在许可下移动」。Commerce 覆盖购买中的其他全部：structured carts、order confirmation、fulfillment tracking、dispute resolution、refunds、chargebacks、returns。这一层把 checkout 和 money transfer 区分开。普通 machine-to-machine API call 不需要 commerce layer；它只是 API call。consumer purchase 显然需要它。

三个有实际差异的选项：

Protocol它做什么最适合
ACP（OpenAI 加 Stripe）structured flow：cart format、order confirmation、fulfillment status、dispute escalation、refunds。merchant 仍是 merchant of record。Consumer shopping：retail goods、subscriptions、physical fulfillment。支撑 ChatGPT Instant Checkout。
UCP（Google）类似 lifecycle，围绕 Google shopping surfaces（Gemini、Google AI Mode）和 Google Pay 构建Google Shopping 上的 merchants；Google AI surfaces 上的 agents
Direct API（machine-to-machine）没有 commerce protocol，只有 HTTP API。通过 x402 或 MPP 付款。没有 cart、disputes、refunds。API access、compute、data feeds：购买物是 stateless API response 的场景

所以 ACP 和 UCP 竞争；「direct API」是这一层的缺席，不是第三个 competitor，而且它常常与「什么都不做」相处得很好。consumer platform 会选 ACP 或 UCP（如果横跨 ChatGPT 和 Gemini，也可能两者都选）。API marketplace 在这一层根本不选择，因为它的购买没有 lifecycle 要管理。

工程师最容易低估的部分是 refunds 和 disputes。买错尺码 t-shirt 的 customer 期望能退货。commerce protocol 必须说明退货如何开始、agent 如何得知，以及 refund 如何通过 settlement 回流。ACP 通过让 merchant 保持 merchant of record 正确处理这一点：既有 dispute machinery（Stripe 的 chargeback flows、retailer 的 return policy）可以直接工作。Direct-API approaches 会因为忽视这一点而出错。它通常完全没有 refund path，这对 $0.0001 的 API call 没问题，但对 $500 的 API credits 就不对。

Commerce flows 通常需要多个 tools 顺序工作：

note

下面的 acp_client 调用是 illustrative；它们代表 ACP commerce backend。Pydantic models、@function_tool 和 Agent wiring 是真实的。

from agents import Agent, function_tool
from pydantic import BaseModel
from decimal import Decimal

class CartItem(BaseModel):
    sku: str
    quantity: int
    unit_price: Decimal

class OrderResult(BaseModel):
    order_id: str
    status: str  # "confirmed", "fulfilled", "shipped", "delivered"
    tracking_url: str | None = None
    estimated_arrival: str | None = None

@function_tool
async def acp_create_cart(merchant_id: str, items: list[CartItem]) -> PaymentToolResult:
    """Create a cart at an ACP merchant. Does NOT charge yet."""
    cart = await acp_client.cart.create(merchant_id=merchant_id, items=items)
    return PaymentToolResult(
        status="success",
        details={"cart_id": cart.id, "merchant_id": merchant_id, "item_count": len(items)},
    )

@function_tool
async def acp_checkout(cart_id: str, spt_token: str) -> OrderResult:
    """Complete the checkout for a previously-created cart."""
    return await acp_client.checkout.complete(cart_id=cart_id, spt_token=spt_token)

@function_tool
async def acp_check_order_status(order_id: str) -> OrderResult:
    """Get the current status of an order. The agent calls this to follow up."""
    return await acp_client.order.status(order_id=order_id)

@function_tool
async def acp_initiate_refund(order_id: str, reason: str) -> RefundResult:
    """Start a refund for an order. Returns refund_id for follow-up."""
    response = await acp_client.refund.create(order_id=order_id, reason=reason)
    return RefundResult(
        refund_id=response.refund_id,
        order_id=order_id,
        status=response.status,
        amount_refunded_usd=response.amount_refunded_usd,
    )

shopping_agent = Agent(
    name="ShoppingAgent",
    instructions="Help the user shop. Create the cart first, confirm items with the user, then check out. Handle refund requests with an ACP refund.",
    tools=[acp_create_cart, acp_checkout, acp_check_order_status, acp_initiate_refund],
)

这里要问的问题是：你的 use case 是否需要 commerce lifecycle。如果需要（cart、refund、dispute），为了 ChatGPT reach 选 ACP，为了 Google reach 选 UCP，或两者都选。如果不需要，跳过这一层，直接从 authorization 走到 settlement。两种常见错误是：把 consumer-commerce protocol 强行套到 machine-to-machine call 上（过重），或者在真实 consumer purchase 中跳过 commerce，之后再糟糕地重新实现 disputes（过轻）。

概念 7：Layer 4，Settlement（钱真正移动）​

一句话： Settlement 是 dollars 真正换 custody 的地方，选择主要由交易经济性决定。

把价值从 buyer 移到 seller。这一层以上的所有内容都只是 choreography；settlement 是 dollars（或 stablecoins，或任何计价单位）实际换手的地方。只有 settlement 完成，agent 才完成了一笔交易。

四个严肃选项，各有经济性和边界：

Protocol工作方式经济性最适合
x402HTTP-native；通过 Base/Solana/EVM 上的 stablecoin transfer settlement，并用 EIP-3009 签名sub-cent gas、1–2 秒 finality、无 protocol feesMachine-to-machine micropayments；高频低价值（API access、per-call billing）
MPPSessions：agent 预授权 cap 和 duration，然后流式 metered payments。Multi-rail（Tempo 上的 stablecoin、Lightning、cards）card rails 上有 Stripe fees；stablecoin 近零；subscription-friendlyEnterprise 和 multi-rail flows；recurring subscriptions；需要 fiat 和 crypto 共存于一个 envelope 的场景
Card rails（Stripe/Adyen/Worldpay）通过 Stripe 接入 Visa、Mastercard、Amex；agent 出示 SPT（ACP）或 Payment Mandate（AP2）；processor 向卡收费cards 约 2.9% 加 $0.30；成熟的 dispute machineryConsumer flows；有 chargeback exposure 的交易；international card acceptance
Bank transfer / LightningACH、SEPA、Bitcoin LightningACH 固定约 $0.25；Lightning sub-cent；SEPA 约 €0.202.9% card fees 很痛的 high-value flows；通过 Lightning 的 cross-border micropayments

settlement 的选择主要是钱的问题。sub-dollar payments 走 x402，因为 stablecoin gas 低于 1 美分。约 $1,000 以下的 consumer purchases 通过 ACP 走 card rails，因为 chargeback protection 值得 2.9%。recurring subscriptions 走 MPP sessions。大型 B2B transfers 走 bank rails 或 Lightning。上层通常会迫使这个选择：commerce 层的 ACP 大多意味着 settlement 层的 card rails；discovery 层 x402-paywalled MCP server 大多意味着 settlement 层的 x402。

小心 headline-number trap。引用 x402 transaction volume 或 MPP integration count 的文章可能误导你。正确问题不是「哪个 volume 最大？」而是「哪个适合这笔交易的经济性？」一笔 $0.001 API call 如果用 card rails settlement，手续费比调用本身还贵。一笔 $5,000 procurement 如果用 x402 stablecoin settlement，就放弃了 card 本可给你的 chargeback protection。

Settlement 通常作为 higher-layer tool 的 side effect 触发：agent 很少直接调用 settle_payment()；settlement 发生在 acp_checkout() 或 x402_fetch() 内部。不过 SDK 层的 spend limits 仍然重要：

note

下面的 x402_client.get 调用是 illustrative；它代表 buyer-side x402 fetch。@function_tool wiring 和 spend check 是真实 Python。

from agents import function_tool, RunContextWrapper
from decimal import Decimal
from .models import X402PaymentResult, PaymentToolResult

@function_tool
async def x402_fetch(
    ctx: RunContextWrapper,
    url: str,
    max_payment_usdc: Decimal,
) -> X402PaymentResult | PaymentToolResult:
    """Fetch a paid URL via x402. Settlement is automatic if cost <= max_payment_usdc."""
    # Check the SDK-level spend tracker before initiating the request
    spent_so_far = Decimal(str(ctx.context.get("session_x402_spend_usdc", Decimal(0))))
    session_cap = ctx.context["user_session"].x402_session_cap_usdc
    if spent_so_far + max_payment_usdc > session_cap:
        return PaymentToolResult(
            status="rejected",
            error=f"Would exceed session spend cap (already spent ${spent_so_far})",
        )

    # Initiate the x402 flow: the server returns 402, the agent retries with a signed payment
    response = await x402_client.get(url, max_payment_usdc=max_payment_usdc)

    # Update the spend tracker, kept in ctx.context for cross-tool visibility
    ctx.context["session_x402_spend_usdc"] = spent_so_far + response.amount_paid_usdc
    return X402PaymentResult(
        content=response.content,
        amount_paid_usdc=response.amount_paid_usdc,
        tx_hash=response.tx_hash,
    )

有一点必须讲清楚：if spent_so_far + ... 检查只是 tool body 内的 soft guard，用来快速、友好地失败，但不是真正安全边界。真正保护你的是 agent 的 smart-contract wallet。即使你删掉 tool 内检查，概念 15 中的 wallet caps 仍会拒绝链上 transfer。tool 检查用于 UX；wallet caps 才是安全。

这里的动作是：选择符合交易经济性的 rail，再确认它与你的 commerce choice 兼容。sub-dollar machine-to-machine 走 x402。consumer purchases 通过 ACP 走 card rails。enterprise subscriptions 走 MPP。不要孤立地选择 settlement；把它作为 composed stack 的底部来选。

第 3 部分：四个协议深入，以及 OpenAI Agents SDK integration​

第 1 和第 2 部分建立了 framing：四个层、每层若干协议、SDK 作为 universal client。第 3 部分会近距离讲四个 headline protocols。每个协议都包括：它是什么、关键 primitives、SDK integration code，以及团队如何部署和运行它的简短说明。

请把它们当成四个平行 deep-dives 来读。每个协议都有相同形状，所以可以并排比较。

OpenAI Agents SDK 如何接入四个 protocol layers。顶部是 SDK 和三个工具：function-tool decorator、run-context wrapper、tool input guardrail。中间四个框展示每个 SDK piece 连接到哪一层：discovery 通过 MCP servers 和 tools，authorization 通过 function tools 和 run context，commerce 通过 function tools 序列，settlement 作为 commerce tools 的 side effect 加 guardrail spend caps。底部是三层 spend-limit stack：Level 1 wallet 和 payment-method caps，任何东西都绕不过去；Level 2 SDK tool input guardrail，在 tool 执行前运行（不是 output guardrail，后者太晚）；Level 3 application business rules。SDK 不是 protocol；它是把 protocols 干净组合起来的 orchestrator。

概念 8 到 11 中的每个协议都会通过这张图里的某个 pattern 接入 SDK。底部的三层 spend-limit stack 预告了第 6 部分的概念 15。阅读时把它放在视野边缘：这张图展示的 safety discipline，正是把这段代码变成可部署系统的关键。

🧰 Pydantic 作为 contract layer：读一次，适用于下面所有协议

第 3 部分的每个代码示例都使用 Pydantic models，而不是普通 Python dicts，来表示 protocol payloads、tool returns、FastAPI request/response bodies 和 Inngest event payloads。这一部分不能跳过。它把整个系统粘在一起。

典型 agent-commerce flow 会跨过四个边界：

  1. SDK 的 @function_tool 向 agent 的 reasoner 返回一个值。
  2. 这个值跨 wire 进入 protocol endpoint（ACP、AP2、x402、MPP）。
  3. protocol 返回一个 response，再跨回来。
  4. 有时 webhook 会稍后到达并跨入 FastAPI handler。

在每个边界上，untyped dict 都会悄悄丢字段、丢 coercions，并发送错误 shape。Pydantic models 会在边界处抓住这四类 failure，并给出精确指向字段的错误。

下面是之后每个 concept 都会反复出现的 pattern：

from pydantic import BaseModel, Field
from decimal import Decimal
from typing import Literal
from agents import function_tool, RunContextWrapper

class CartItem(BaseModel):
    sku: str
    quantity: int = Field(ge=1)
    unit_price_usd: Decimal

class CheckoutRequest(BaseModel):
    merchant_id: str
    items: list[CartItem]
    max_total_usd: Decimal = Field(gt=0)

class CheckoutResult(BaseModel):
    order_id: str
    status: Literal["confirmed", "failed", "pending_user_confirmation"]
    total_charged_usd: Decimal
    estimated_delivery: str | None = None

@function_tool
async def acp_checkout(ctx: RunContextWrapper, request: CheckoutRequest) -> CheckoutResult:
    # Pydantic has already validated the request shape before this line runs.
    # Returning a CheckoutResult means the agent's reasoner gets typed feedback.
    ...

在这个规模下，它有三个具体意义：

  1. reasoner 会把 return type 当作 feedback。 当 acp_checkout 返回 typed CheckoutResult 时，agent 的下一步 reasoning 会拿到清晰的 field names 和 types，而不是 stringified dict。tool-selection accuracy 会明显提高。
  2. FastAPI 原生使用 Pydantic。 Stripe webhooks、AP2 mandate callbacks 和 MPP session events 都会在 FastAPI handler 中 deserialize 成 Pydantic models：与 agent tools 返回的 model 是同一套。一个 contract，两个 endpoints。
  3. Inngest events 携带 Pydantic payloads。 FastAPI webhook handler 触发 Inngest event 时，payload 是 Pydantic model。suspended step.wait_for_event 会直接收到 typed payload。workflow 中不需要 JSON parsing。

钱永远用 Decimal。 本课每个 monetary amount 都使用 Decimal，绝不使用 float。money 上的 floating-point math 会丢精度，并在数千次 micropayments 中累积。Stripe SDK 两者都接受，但会用 Decimal 报回。x402 amounts 以 on-chain integers 到达（USDC 有 6 位小数），你再包成 Decimal。除非序列化到 wire format，否则 money 始终保持 Decimal。

shared result models。 课程不会在每个 concept 中重新定义 result types，而是一次性定义一小组 result models。下面的每个代码块都会按名称 import 它们。把这些放进你的 models.py：

from pydantic import BaseModel, Field
from decimal import Decimal
from typing import Literal
from datetime import datetime

# --- Tool-result models (returned by @function_tool functions) ---

class PaymentToolResult(BaseModel):
    """Generic envelope for any payment-related tool action."""
    status: Literal["success", "failed", "rejected", "pending"]
    error: str | None = None
    details: dict | None = None  # protocol-specific details

class MandateResult(BaseModel):
    """Result of creating an AP2 mandate (Intent / Cart / Payment)."""
    mandate_id: str | None = None
    status: Literal["signed", "declined", "pending", "failed"]
    expires_at: str | None = None  # ISO 8601
    error: str | None = None

class OrderStatusResult(BaseModel):
    """Result of fetching ACP order status."""
    order_id: str
    status: Literal["confirmed", "shipped", "delivered", "cancelled", "refunded", "pending"]
    tracking_url: str | None = None
    estimated_delivery: str | None = None

class RefundResult(BaseModel):
    """Result of initiating an ACP refund."""
    refund_id: str
    order_id: str
    status: Literal["initiated", "processing", "completed", "failed"]
    amount_refunded_usd: Decimal | None = None

class DiscoveryResult(BaseModel):
    """Result of an Agent.market or similar agent-directory search."""
    service_id: str
    name: str
    description: str
    price_per_call_usdc: Decimal
    endpoint_url: str

class X402PaymentResult(BaseModel):
    """Result of an x402-paid fetch."""
    content: str
    amount_paid_usdc: Decimal
    tx_hash: str | None = None

class MPPSessionResult(BaseModel):
    """Result of creating or closing an MPP session."""
    session_id: str
    status: Literal["active", "closed", "expired", "failed"]
    total_charged_usd: Decimal | None = None  # only populated on close
    expires_at: datetime | None = None
    rail_breakdown: dict[str, Decimal] | None = None  # only on close

class MPPMeteredCallResult(BaseModel):
    """Result of a metered call within an active MPP session."""
    session_id: str
    cost_usd: Decimal
    response_payload: dict
    accumulated_session_spend_usd: Decimal

# --- FastAPI handler models (request/response bodies for webhooks/callbacks) ---

class WebhookAck(BaseModel):
    """Generic webhook handler ack."""
    received: bool = True
    event_id: str | None = None

# --- Inngest workflow result models (return types of @inngest_client functions) ---

class WorkflowResult(BaseModel):
    """Generic workflow completion envelope."""
    status: Literal["completed", "abandoned", "failed", "partial"]
    reason: str | None = None
    output: dict | None = None

这些 models 会贯穿第 3 和第 6 部分。下面的代码块会按名称 import 它们，不再重复定义。这一 sidebar 也不重复。

概念 8：ACP（Agentic Commerce Protocol），consumer-shopping protocol​

一句话： ACP 让 agent 可以代表一个人在真实 merchant 完成真实 checkout，同时 merchant 仍然对这笔销售负责。

它是什么。 ACP 是 OpenAI 和 Stripe 构建的开放 specification，于 2025 年 9 月 29 日随 Etsy 和 Shopify 作为 launch partners 推出。到 2026 年初，它支撑 ChatGPT Instant Checkout，覆盖 Shopify-integrated merchants 和若干大型 retail brands。merchant counts 和 brand lists 会因来源不同而变化，所以要用 ACP integration directory 验证当前 adoption。协议是 Apache 2.0，通过 github.com/agentic-commerce-protocol/agentic-commerce-protocol 的 Specification Enhancement Proposal process 治理。repository 将 spec 标记为 beta，所以预期 course examples 与 live spec 之间会有 drift。

它所在的位置。 ACP 主要位于 Layer 3（Commerce），并通过 token mechanism 触及 Layer 2（Authorization）。它覆盖 cart formation、checkout、order management、fulfillment status 和 refund mechanics。merchant 仍然是 merchant of record：chargebacks、returns 和 customer service 都流经 merchant 既有系统。（merchant of record 是在法律上对交易负责的 business。）

两个重要 primitives。

  1. Shared Payment Token（SPT）。 payment processor（reference build 中是 Stripe）发出的一次性 token，锁定到一个 merchant、一个 amount cap、一个短时间窗，通常也只能使用一次。如果 agent 被放行花 $50，却试图花 $1,000，SPT 会在 protocol level 直接失败。SPT 是 ACP 让 agent 保持在用户授权范围内的方式。
  2. Cart Mandate（通过 AP2 extension）。 ACP 可以与 AP2 组合以获得额外 audit rigor：用户在 agent 提交 SPT 前签署 Cart Mandate。这在 ACP 中是可选项，但在 regulated flows 中越来越常见。（mandate 是人类授权某类具体花费的 signed proof。）

OpenAI Agents SDK integration。 ACP 是 SDK 上最 production-ready 的协议，因为 OpenAI 共同构建了它。Stripe Python SDK 加一层很薄的 ACP client wrapper，就能完成完整集成。

note

下面的 acp_client 调用和 stripe.PaymentTokens.create(...) 是 illustrative：它们展示 ACP integration 的形状。真实 SPT minting 会走 live Stripe ACP endpoints，而不是 stripe.PaymentTokens method。agent wiring、guardrail 和 result models 都是真实且今天可运行的。代码块之后还有说明。

from agents import Agent, Runner, function_tool, RunContextWrapper
from agents.tool_guardrails import (
    tool_input_guardrail,
    ToolInputGuardrailData,
    ToolGuardrailFunctionOutput,
)
from pydantic import BaseModel
from decimal import Decimal
from typing import Literal
import json
import stripe
import time
from .models import OrderStatusResult, RefundResult

class CartItem(BaseModel):
    sku: str
    name: str
    quantity: int
    unit_price_usd: Decimal

class CheckoutResult(BaseModel):
    order_id: str
    status: Literal["confirmed", "failed", "pending_user_confirmation"]
    total_charged_usd: Decimal
    estimated_delivery: str | None = None

# Tool input guardrail: refuse the checkout if the user can't authorize the spend
@tool_input_guardrail
def verify_user_can_spend(data: ToolInputGuardrailData) -> ToolGuardrailFunctionOutput:
    args = json.loads(data.context.tool_arguments or "{}")
    max_total = Decimal(str(args.get("max_total_usd", 0)))
    merchant_id = args.get("merchant_id", "")
    user_session = data.context.context["user_session"]
    if not user_session.can_spend(max_total, merchant_id):
        return ToolGuardrailFunctionOutput.reject_content(
            f"User cannot authorize ${max_total} at merchant {merchant_id}"
        )
    return ToolGuardrailFunctionOutput.allow()

@function_tool
async def acp_browse_merchant(merchant_id: str, query: str) -> list[dict]:
    """Search a merchant's catalog via their ACP catalog endpoint."""
    response = await acp_client.catalog.search(
        merchant_id=merchant_id,
        query=query,
        limit=20,
    )
    return [item.model_dump() for item in response.items]

@function_tool(tool_input_guardrails=[verify_user_can_spend])
async def acp_create_cart_and_checkout(
    ctx: RunContextWrapper,
    merchant_id: str,
    items: list[CartItem],
    max_total_usd: Decimal,
) -> CheckoutResult:
    """Create a cart at the merchant and complete checkout in one transaction.
    Mints an SPT scoped to max_total_usd; the merchant verifies and charges.
    The verify_user_can_spend guardrail above has already validated authorization."""

    user_session = ctx.context["user_session"]

    # Mint the Shared Payment Token via Stripe (Decimal to cents via quantize)
    cents = int((max_total_usd * 100).quantize(Decimal("1")))
    spt = stripe.PaymentTokens.create(
        amount=cents,
        currency="usd",
        merchant_id=merchant_id,
        user_session_id=user_session.id,
        max_uses=1,
        expires_at=int(time.time()) + 600,  # 10-minute window
    )

    # Submit the cart and SPT to the merchant's ACP endpoint
    result = await acp_client.checkout.complete(
        merchant_id=merchant_id,
        items=[i.model_dump() for i in items],
        spt_token=spt.token,
    )

    # Update the user's spend tracker (Decimal in, Decimal out)
    user_session.record_spend(
        amount_usd=Decimal(str(result.total_charged_usd)),
        merchant_id=merchant_id,
        order_id=result.order_id,
    )

    return CheckoutResult(**result.model_dump())

@function_tool
async def acp_check_order(order_id: str) -> OrderStatusResult:
    """Get the current status of an ACP order (fulfillment, shipping, delivery)."""
    raw = await acp_client.orders.get(order_id=order_id)
    return OrderStatusResult(
        order_id=raw.order_id,
        status=raw.status,
        tracking_url=raw.tracking_url,
        estimated_delivery=raw.estimated_delivery,
    )

@function_tool
async def acp_refund(
    order_id: str,
    reason: str,
    amount_usd: Decimal | None = None,
) -> RefundResult:
    """Start a refund for an ACP order. Returns refund_id for follow-up.
    If amount_usd is None, a full refund is requested."""
    raw = await acp_client.refunds.create(
        order_id=order_id,
        reason=reason,
        amount_usd=amount_usd,
    )
    return RefundResult(
        refund_id=raw.refund_id,
        order_id=order_id,
        status=raw.status,
        amount_refunded_usd=raw.amount_refunded_usd,
    )

shopping_agent = Agent(
    name="ShoppingAgent",
    instructions="""Help the user shop at ACP-enabled merchants. Workflow:
    1. Use acp_browse_merchant to find products matching the user's request
    2. Present the matched items to the user (via reasoning, not a tool)
    3. When the user confirms, use acp_create_cart_and_checkout to complete the purchase
    4. Use acp_check_order to report order status when the user asks
    5. Use acp_refund only when the user explicitly requests a return""",
    tools=[acp_browse_merchant, acp_create_cart_and_checkout, acp_check_order, acp_refund],
    model="gpt-5.5",
)

这里真实的部分是：Agent、Runner、@function_tool 和 tool_input_guardrail scaffolding，以及 typed result models。这一部分可以在每个协议中复用。stand-in 的部分是 acp_client 和 stripe.PaymentTokens.create 调用，它们代表 ACP-Stripe backend。在生产中，你把该 backend 换成 live Stripe ACP endpoints，其余保持不变。下面是一个同形状的 runnable mock，用来看到它如何工作：

class MockACPClient:
    """Illustrative backend. Real ACP uses live Stripe ACP endpoints, not stripe.PaymentTokens."""
    async def checkout(self, merchant_id: str, amount_usd) -> dict:
        return {"order_id": "ord_mock_1", "status": "confirmed", "total_charged_usd": str(amount_usd)}

acp_client = MockACPClient()   # stands in for the ACP/Stripe backend

standard harness（这里讲一次，后面三个概念引用）。 ACP 运行在普通 cloud harness 上，不需要额外部件。Stripe SDK 运行在 FastAPI handler 内；ACP calls 是 outbound HTTPS；SPTs 存在 agent 的 run-scoped context 中，通过 RunContextWrapper 传递。唯一规则是：把 Stripe API keys 放进 secret store（key vault），不要放环境变量，并按 Stripe 的 schedule 轮换。无需 sandbox（ACP 不运行代码），也不需要特殊 storage（orders 持久化在 Stripe 和 merchant 系统中，audit 可选 shadow records）。这个 deployment baseline 对 AP2、x402 和 MPP 也相同。后面三个概念只说明每个协议额外增加什么。cloud deployment 在 deploying-agents 速成课中覆盖（进行中）。

durable 地运行它（Inngest）。 ACP transactions 很短，通常端到端 5–30 秒。它们很适合放进 Inngest step.run blocks。当 agent 必须在 cart creation 和 checkout 之间暂停、等待用户确认 cart（human-in-the-loop pattern）时，step.wait_for_event primitive 才真正体现价值。Production Worker 速成课 会深入讲这个 durability layer；这里是形状：

import inngest
from datetime import timedelta
from .models import WorkflowResult

@inngest_client.create_function(
    fn_id="shopping-workflow",
    trigger=inngest.TriggerEvent(event="shopping/checkout.requested"),
    concurrency=[inngest.Concurrency(limit=5, key="event.data.user_id")],
)
async def shopping_workflow(ctx: inngest.Context) -> dict:
    # Run the agent to produce the cart proposal
    cart = await ctx.step.run(
        "agent-builds-cart", build_cart_fn, ctx.event.data["user_query"],
    )

    # Wait for the user to confirm the proposed cart (human-in-the-loop)
    confirmation = await ctx.step.wait_for_event(
        "wait-for-user-confirm",
        event="shopping/cart.confirmed",
        if_exp=f"async.data.cart_id == '{cart['cart_id']}'",
        timeout=timedelta(minutes=15),
    )
    if confirmation is None:                 # timeout returns None
        return {"status": "abandoned", "reason": "user did not confirm in time"}

    # The user confirmed; complete checkout with the now-valid SPT
    result = await ctx.step.run("complete-checkout", complete_checkout_fn, cart["cart_id"])
    return {"status": "completed", "output": result}

有两个细节经常绊倒人：wait_for_event 用 if_exp（一个小表达式）匹配 incoming payload，等待字段位于 async. prefix 下。timeout 会返回 None，所以继续前要先检查。

团队常犯的错。 他们跳过 user-confirmation step。ACP 同时支持「用户确认每个 cart」和「用户预授权这一类购买」。团队常为了速度选择后者，随后发现一个小的 SPT misconfiguration 会让 agent 买到稍微不对的商品，且没有好办法恢复。生产前一个月默认要求确认 cart。只有在你测量过 agent 多常把 cart 做对之后，才放宽这个 gate。

概念 9：AP2（Agent Payments Protocol），authorization layer​

一句话： AP2 生成签名证明，表明人类允许这笔花费；它不移动资金本身，而是证明这笔资金被允许移动。

它是什么。 AP2 是 Google 与 60+ partners 推出的开放 specification，于 2025 年 9 月发布（latest version v0.2.0，2026 年 4 月）。Apache 2.0，维护在 github.com/google-agentic-commerce/AP2，提供 Python、TypeScript、Kotlin 和 Go 的 reference implementations。AP2 是 authorization layer，不是 commerce 或 settlement protocol。它生成 signed mandates，证明 agent 被授权花钱，然后把实际 settlement 交给合适的 rail（cards、bank，或通过 a2a-x402 extension 使用 x402）。

它所在的位置。 AP2 位于 Layer 2（Identity and Authorization）。它构建在两个底层协议之上：A2A（Agent2Agent，用于 agent-to-agent messaging）和 MCP（用于 tool exposure）。AP2 mandate 会作为 signed credential 通过 A2A 传递，或附加到 MCP tool call 上。

三个重要 primitives，也就是三种 mandate types。

Mandate何时创建证明什么
Intent Mandate任务开始时，由用户在自己的 UI 中签署用户允许 agent 在固定规则内行动（price limits、time windows、allowed merchants）
Cart Mandateagent 建好具体 cart 后、checkout 前由用户签署（human-present flows）用户批准了这个确切 cart 和这个确切价格
Payment Mandate付款时，由用户签署或根据 Intent Mandate 自动生成用户授权在这条确切 rail 上完成这笔确切付款

audit trail。 三种 mandates 形成一条签署者之后不能否认的链：Intent（「买 $120 以下的鞋」）指向 Cart（「这双鞋，$110」），再指向 Payment（「向这个 stablecoin wallet 收费」）。每份 mandate 都指回前一份。如果之后在 dispute 或 fraud claim 中挑战任何一步，整条链都可审计。这个性质就是 non-repudiable：「我从未授权」无法对抗用户自己的 signature。这也是为什么 AP2 适合 healthcare 和 financial services 这类 regulated industries，因为「用户是否真的授权？」具有法律重量。

SDK 中的变化。 AP2 在 OpenAI Agents SDK 中没有 first-class 位置；它的 reference builds 使用 Google Agent Development Kit。你把它接成 @function_tool functions，用来 create、sign、validate 和 dispatch mandates。harness 与概念 8 相同。AP2 额外增加的一件事是：一个用户实际签署每份 mandate 的 signing surface。

note

下面的 from ap2 import ... imports、MandateSigner 和 ap2_x402 调用都是 illustrative。AP2 的真实 Python package 是 ap2，mandate models 位于 ap2.types.mandate，真实字段也不同于这里简化的 principal_did / agent_did / rules。没有 MandateSigner class；真实 signing 使用 A2A 上的 verifiable credentials。mandate-chain 这个概念是真实的；这段具体代码是 teaching stand-in。SDK scaffolding 和 typed results 是真实的。

from agents import Agent, function_tool, RunContextWrapper
from agents.tool_guardrails import (
    tool_input_guardrail,
    ToolInputGuardrailData,
    ToolGuardrailFunctionOutput,
)
from ap2 import IntentMandate, CartMandate, PaymentMandate, MandateSigner
from pydantic import BaseModel
from decimal import Decimal
from datetime import datetime
from .models import MandateResult, PaymentToolResult

class IntentRules(BaseModel):
    max_total_usd: Decimal
    allowed_merchants: list[str] | None = None
    allowed_categories: list[str] | None = None
    expires_at: str  # ISO 8601 datetime

# Guardrail: refuse any cart that has no preceding Intent Mandate
@tool_input_guardrail
def require_intent_mandate(data: ToolInputGuardrailData) -> ToolGuardrailFunctionOutput:
    intent = data.context.context.get("intent_mandate")
    if not intent:
        return ToolGuardrailFunctionOutput.reject_content(
            "No Intent Mandate found. Create one via ap2_create_intent_mandate first."
        )
    return ToolGuardrailFunctionOutput.allow()

@function_tool
async def ap2_create_intent_mandate(
    ctx: RunContextWrapper,
    task_description: str,
    rules: IntentRules,
) -> MandateResult:
    """Create an Intent Mandate at the start of a purchasing task.
    Needs the user's signature, so call this BEFORE the agent shops."""

    user_session = ctx.context["user_session"]
    mandate = IntentMandate(
        principal_did=user_session.did,  # decentralized identifier
        agent_did=ctx.context["agent_did"],
        task=task_description,
        rules=rules.model_dump(),
        issued_at=datetime.utcnow().isoformat(),
    )

    # Send to the user's signing UI; block until the user signs or rejects
    signed = await user_session.signer.request_signature(
        mandate,
        ui_prompt="Approve this shopping task?",
        timeout_seconds=300,
    )
    if not signed:
        return MandateResult(status="declined", error="User declined to sign Intent Mandate")

    # Store the mandate for later reference by Cart/Payment mandates
    ctx.context["intent_mandate"] = signed
    return MandateResult(mandate_id=signed.id, status="signed", expires_at=rules.expires_at)

@function_tool(tool_input_guardrails=[require_intent_mandate])
async def ap2_create_cart_mandate(
    ctx: RunContextWrapper,
    cart_items: list[dict],
    total_usd: Decimal,
    merchant_id: str,
) -> MandateResult:
    """Create a Cart Mandate that references the current Intent Mandate.
    Needs the user's signature in human-present flows.
    The require_intent_mandate guardrail above has verified an Intent Mandate exists."""

    intent = ctx.context["intent_mandate"]  # guaranteed present by the guardrail

    # Check that the cart fits inside the Intent Mandate's rules
    if total_usd > Decimal(str(intent.rules["max_total_usd"])):
        return MandateResult(
            status="failed",
            error=f"Cart total ${total_usd} exceeds Intent Mandate cap ${intent.rules['max_total_usd']}",
        )

    cart_mandate = CartMandate(
        parent_intent_id=intent.id,
        cart_items=cart_items,
        total_usd=str(total_usd),  # serialize Decimal as a string on the wire
        merchant_id=merchant_id,
        issued_at=datetime.utcnow().isoformat(),
    )

    user_session = ctx.context["user_session"]
    signed = await user_session.signer.request_signature(
        cart_mandate,
        ui_prompt=f"Approve this cart for ${total_usd}?",
        timeout_seconds=300,
    )
    if not signed:
        return MandateResult(status="declined", error="User declined to sign Cart Mandate")

    ctx.context["cart_mandate"] = signed
    return MandateResult(mandate_id=signed.id, status="signed")

@function_tool
async def ap2_settle_via_x402(
    ctx: RunContextWrapper,
    merchant_x402_url: str,
) -> PaymentToolResult:
    """Use the AP2 a2a-x402 extension to settle the current Cart Mandate via an x402 stablecoin payment."""
    cart = ctx.context.get("cart_mandate")
    if not cart:
        return PaymentToolResult(status="failed", error="No Cart Mandate to settle")

    # The a2a-x402 extension generates a Payment Mandate authorizing the x402 transfer
    payment_mandate = await ap2_x402.create_payment_mandate(
        cart_mandate=cart,
        rail="x402",
        chain="eip155:8453",  # Base
        asset="USDC",
    )

    # Dispatch the x402 payment with the Payment Mandate attached as proof
    result = await x402_client.pay(
        url=merchant_x402_url,
        amount_usdc=Decimal(str(cart.total_usd)),
        payment_mandate=payment_mandate,
    )
    return PaymentToolResult(status="success", details=result.model_dump())

procurement_agent = Agent(
    name="ProcurementAgent",
    instructions="""Enterprise procurement workflow:
    1. ALWAYS create an Intent Mandate first via ap2_create_intent_mandate
    2. Search merchants for matching items
    3. Build a cart and create a Cart Mandate via ap2_create_cart_mandate
    4. Settle via x402 (stablecoin) using ap2_settle_via_x402, or via card rails
    5. Record every mandate ID in the procurement system for audit""",
    tools=[ap2_create_intent_mandate, ap2_create_cart_mandate, ap2_settle_via_x402],
    model="gpt-5.5",
)

真实的部分是：与概念 8 相同的 SDK scaffolding 和 typed results。stand-in 的部分是 ap2 mandate classes、MandateSigner 和 ap2_x402 extension。mandate chain 本身是真实且可构建的想法；这里的 exact API 为教学而简化。下面是一个 runnable mock，让这个 pattern 能执行：

class MockSignedMandate:
    def __init__(self, mid, rules=None): self.id, self.rules = mid, (rules or {})
class MockSigner:
    """Illustrative. Real AP2 signing uses verifiable credentials over A2A; no MandateSigner class exists."""
    async def request_signature(self, mandate, ui_prompt="", timeout_seconds=300):
        return MockSignedMandate(getattr(mandate, "id", "mandate_mock_1"))
ap2_x402 = type("Mock", (), {"create_payment_mandate": staticmethod(lambda **k: MockSignedMandate("pm_mock_1"))})()

它给 harness 增加什么。 相比概念 8 baseline，多了两件事：一个让用户签署 mandates 的 signing surface（web app、mobile app 或 notification-based signing tool），以及可持久保存 signed mandates、覆盖完整 dispute windows 的 durable storage（financial mandates 的标准 retention 是 7 年）。signing surface 是与 ACP 的真正差异：ACP 可以复用现有 login sessions，但 AP2 需要专门的 signature flow。

durable 地运行它（Inngest）。 mandate signing 是 step.wait_for_event 的教科书用法。agent function 触发「signing requested」event，function suspend，用户在 UI 中签署并触发「signing signed」，function 恢复。等待期间不消耗 compute，这很重要，因为 enterprise signature 可能耗时数小时。Production Worker 速成课会深入讲这个 resume pattern。

团队常犯的错。 他们把 mandate creation 当成 checkout-time concern，在 agent 已经做了大量工作后才签 mandate。应该先创建 Intent Mandate，在 agent 开始 shopping 之前完成。这样能早早抓住 scope mismatch（用户想买鞋，但 Intent Mandate 只允许 office supplies），而不是等 agent 花 compute 建好一个永远无法付款的 cart 之后才发现。

概念 10：x402，HTTP-native settlement protocol​

一句话： x402 通过复活旧的 HTTP 402「Payment Required」状态码，让 agent 能用 stablecoin 在 1–2 秒内为一次 API call 付款。

它是什么。 x402 把沉睡的 HTTP 402 status code 变成 APIs 和 machine-to-machine commerce 的可用 payment layer。它由 Coinbase 创建（2025 年 5 月），V2 于 2025 年 12 月发布，现在由 Linux Foundation 的 x402 Foundation steward（2026 年 4 月），成员包括 Cloudflare、Stripe、AWS、Google 等。Apache 2.0。截至 2026 年初，x402 报告在 Base 和 Solana 上累计超过 1 亿笔 payments，并有持续增长的 facilitator ecosystem；数字变化很快，所以请在 x402.org 和 Coinbase 的 x402 launch pages 验证当前数据。

它所在的位置。 x402 主要位于 machine-to-machine flows 的 Layer 4（Settlement），但它也触及 Layer 1（通过 Agent.market 等 directories）和 Layer 3（对于没有真实购买 lifecycle 的普通 API access，它可以充当完整 commerce layer）。对纯 machine-to-machine flows，x402 往往是你唯一需要的协议。

四个重要 primitives。

  1. HTTP 402 status code。 当未付款 client 请求 paid resource 时，server 返回 402 Payment Required，加一个 header 携带 payment requirements：scheme（固定金额用 exact）、network（CAIP-2 形式，如 eip155:8453，表示「Base」）、asset（USDC）、recipient address、max amount 和 expiry。（CAIP-2 是一种标准区块链命名方式，让协议保持 chain-agnostic。）
  2. payment authorization header。 client 会带着一个携带 signed payment authorization 的 header 重试。signature 离线生成，所以 buyer 不付 gas。
  3. EIP-3009（transferWithAuthorization）。 x402 构建其上的 Ethereum 标准。它让 buyer 离线签署 payment，由其他人提交上链，所以 buyer 不需要直接接触区块链或支付 gas。
  4. Facilitator。 可选第三方，替 merchant 检查 signature 并提交 on-chain payment，因此 merchant 不需要运行区块链 plumbing。Coinbase 和 Cloudflare 都运行 facilitators。
关于 header 命名：请按当前文档验证

x402 经历了 V1 和 V2，header names 在 spec versions、facilitators 和 SDKs 之间会不同。Cloudflare 当前文档在 402 response 上使用 PAYMENT-REQUIRED，在 client retry 上使用 PAYMENT-SIGNATURE，在 success reply 上使用 PAYMENT-RESPONSE。一些早期示例使用 X-PAYMENT 和 X-PAYMENT-PROOF。flow 相同；只有 wire names 不同。写代码前，请按 x402.gitbook.io 和你的 facilitator 文档确认 exact names。下面的 trace 使用角色（request、带 requirements 的 402、signed retry、带 proof 的 success），不选择某一种命名 convention。

一条 trace 中的 flow。

1. Agent: GET https://api.example.com/data
2. Server: 402 Payment Required
   <payment-required header>: { network: "eip155:8453", asset: USDC,
                                recipient: 0xMerchant..., max_amount: 100000,
                                expiry: 1716304800 }
3. Agent (signs an EIP-3009 authorization off-chain):
   GET https://api.example.com/data
   <payment-signature header>: <base64-encoded signed authorization>
4. Server: 200 OK
   <payment-response header>: <transaction hash>
   { data: ... }

整笔交易需要 1–2 秒。没有 account creation，没有 API key，没有 session，没有 human in the loop。

SDK 中的变化。 x402 是四个协议中最简单的。Cloudflare 提供 helper，把 MCP client 包上 x402 payment ability；非 MCP 使用时，Python buyer-side library 加下面的 canonical pattern 就够了。harness 是概念 8 的同一个 baseline。

note

这里展示的 from x402_client import ... constructor、from cloudflare_agents import withX402Client import，以及 response.amount_paid_usdc field 都是 illustrative。真实 buyer-side package 是 x402-client（from x402_client import X402Client，真实 constructor 是 X402Client(account=...)，.get(url) 返回 httpx.Response）。这里展示的更丰富 wallet constructor 是 teaching stand-in，live call 需要 funded account 和真实 402 endpoint，本课不做。Cloudflare 的 withX402Client 是 JavaScript helper；没有 Python cloudflare_agents package。MCP server（MCPServerStreamableHttp）是真实 Python；这里的 payment wrapping 是 illustrative。不会移动真实资金。

from agents import Agent, function_tool, RunContextWrapper
from agents.mcp import MCPServerStreamableHttp
from agents.tool_guardrails import (
    tool_input_guardrail,
    ToolInputGuardrailData,
    ToolGuardrailFunctionOutput,
)
from x402_client import X402Client, X402Wallet
from cloudflare_agents import withX402Client
from decimal import Decimal
import json
from .models import DiscoveryResult, X402PaymentResult, PaymentToolResult

# Pattern 1: wrap an MCP server's tools with x402 payment ability
research_mcp = MCPServerStreamableHttp(
    name="research-services",
    params={"url": "https://research-services.example.com/mcp"},
)
research_mcp_with_payments = withX402Client(
    research_mcp,
    wallet=agent_wallet,  # smart-contract wallet the agent controls
    max_per_call_usdc=Decimal("0.10"),
    max_per_session_usdc=Decimal("10.00"),
)

# Pattern 2: direct x402 calls as @function_tool
x402_client = X402Client(wallet=agent_wallet)

# Guardrail: refuse any x402 call that would exceed the session cap
@tool_input_guardrail
def enforce_x402_session_cap(data: ToolInputGuardrailData) -> ToolGuardrailFunctionOutput:
    args = json.loads(data.context.tool_arguments or "{}")
    max_payment = Decimal(str(args.get("max_payment_usdc", 0)))
    ctx = data.context.context
    spent = Decimal(str(ctx.get("session_x402_spend_usdc", 0)))
    cap = Decimal(str(ctx["user_session"].x402_session_cap_usdc))
    if spent + max_payment > cap:
        return ToolGuardrailFunctionOutput.reject_content(
            f"x402 session cap would be exceeded: ${spent} spent + ${max_payment} requested > ${cap}"
        )
    return ToolGuardrailFunctionOutput.allow()

@function_tool(tool_input_guardrails=[enforce_x402_session_cap])
async def x402_fetch(
    ctx: RunContextWrapper,
    url: str,
    max_payment_usdc: Decimal = Decimal("0.10"),
) -> X402PaymentResult | PaymentToolResult:
    """Fetch a URL that may require an x402 payment up to max_payment_usdc.
    Signs the EIP-3009 authorization and retries with the payment-signature header.
    The enforce_x402_session_cap guardrail above has already checked the session bound."""

    try:
        response = await x402_client.get(url, max_payment_usdc=max_payment_usdc)
        # Update the spend tracker for later guardrail checks
        current = Decimal(str(ctx.context.get("session_x402_spend_usdc", 0)))
        ctx.context["session_x402_spend_usdc"] = current + Decimal(str(response.amount_paid_usdc))
        return X402PaymentResult(
            content=response.content,
            amount_paid_usdc=Decimal(str(response.amount_paid_usdc)),
            tx_hash=response.tx_hash,
        )
    except X402PaymentRequired as e:
        return PaymentToolResult(
            status="rejected",
            error=f"Resource requires ${e.required_usdc}, exceeds max ${max_payment_usdc}",
        )

@function_tool
async def x402_search_agent_market(
    query: str,
    max_price_per_call_usdc: Decimal = Decimal("0.05"),
) -> list[DiscoveryResult]:
    """Search Agent.market for x402-paywalled services matching the query."""
    results = await agent_market_client.search(
        query=query,
        max_price_per_call_usdc=max_price_per_call_usdc,
    )
    return [
        DiscoveryResult(
            service_id=r.service_id,
            name=r.name,
            description=r.description,
            price_per_call_usdc=Decimal(str(r.price_per_call_usdc)),
            endpoint_url=r.endpoint_url,
        )
        for r in results
    ]

research_agent = Agent(
    name="ResearchAgent",
    instructions="""Research user queries by paying for data sources via x402.
    Workflow:
    1. Use x402_search_agent_market to find relevant paid services
    2. Use x402_fetch to pull data from selected services (max $0.10 per call)
    3. Synthesize the findings into a final report
    Stay under $10 per research session.""",
    tools=[x402_fetch, x402_search_agent_market],
    mcp_servers=[research_mcp_with_payments],
)

真实的部分是：SDK scaffolding、typed results 和 MCPServerStreamableHttp（MCP server 是真实 Python）。stand-in 的部分是这里写出的 buyer-side x402 client、withX402Client wrapper（现实中只有 JavaScript）和 agent_market_client。下面的 runnable mocks 可以让 harness pattern 在不移动真实资金的情况下执行：

from decimal import Decimal

class MockX402Response:
    def __init__(self, content, amount, tx):
        self.content, self.amount_paid_usdc, self.tx_hash = content, amount, tx
class MockX402Client:
    """Illustrative. Real buyer-side package: x402-client (X402Client(account=...), .get() -> httpx.Response)."""
    async def get(self, url: str, max_payment_usdc: Decimal = Decimal("0.10")):
        return MockX402Response(content="<paid data>", amount=Decimal("0.02"), tx="0xmocktx")
x402_client = MockX402Client()
agent_wallet = object()   # stands in for the wallet handle

def withX402Client(mcp_server, **kwargs):
    """Illustrative: JavaScript-only in reality. Returns the server unchanged for the Python demo."""
    return mcp_server

它给 harness 增加什么。 几乎什么都不增加。x402 不需要概念 8 baseline 之外的额外基础设施。agent 的 smart-contract wallet 位于它交易的 chain address 上（多数情况下是 Base），wallet 的 signing key 存在 key vault 中，buyer-side library 处理 signing 和 HTTP retry。（smart-contract wallet 是一种 crypto wallet，它的 spending caps 由区块链代码强制执行，所以即使 agent 代码失控也仍然有效。）

durable 地运行它（Inngest）。 x402 calls 很短（1–2 秒）且 idempotent：同一个 request 和 signature 总会产生相同结果。它们很适合放进 step.run blocks，这里的 memoization 很有价值。如果 run 在 10 个 API calls 中已经付了 5 个之后崩溃，这 5 个 paid calls 会被 memoized，retry 只会为剩下 5 个付款。没有它，retry 会再次为全部 10 个付款。

wallet cap 才是保护你的安全边界

陷阱是把 x402 当成给 agent 一张信用卡。真正保护你的是 wallet 的 on-chain spend limit，而不是 per-request max_payment_usdc。跳过 on-chain limit，使用没有 cap 的 hot wallet，一个卡住的 agent loop 就能把它耗尽。请按 agent identity、session 和 merchant 配置 on-chain spend limits，形成三层独立限制，这样一个 bug 不会清空 wallet。

概念 11：MPP（Machine Payments Protocol），sessions-based settlement protocol​

一句话： MPP 让 agent 打开一个带 spending cap 的预付标签页，然后在关闭之前跨多条 rails 流式支付许多小额款项。

它是什么。 MPP 由 Stripe 和 Tempo 构建，并在 2026 年 3 月 公开发布。Tempo 是 Stripe 与 Paradigm 孵化的 layer-1 blockchain，用于 high-frequency machine payments。MPP 发布时 partners 覆盖 Stripe、Visa、Lightspark（Lightning Network）和其他 ecosystem players；partner list 会增长，所以请通过 mpp.dev 和 Cloudflare 的 MPP docs 验证当前参与者。Apache 2.0。MPP 是 Stripe 在 settlement layer 对 x402 的回答：use case 重叠，哲学不同。

它所在的位置。 MPP 位于 Layer 4（Settlement），并且在 machine-to-machine payments 上直接与 x402 竞争。关键区别是：MPP 是 multi-rail，并支持 per-charge 和 session-based authentication；x402 是 stablecoin-only 且 per-request。

HTTP shape。 x402 使用 402 code 和 custom headers，而 MPP 把 payment 叠到标准 HTTP authentication 上。server 返回带 requirements 的 WWW-Authenticate: Payment，client 带 Authorization: Payment <signed-payload> 重试，server 返回携带 settlement proof 的 Payment-Receipt。这让 MPP 更像熟悉的 HTTP auth flow，而不是自定义 protocol。

两种 intent types。

IntentLifecycle最适合
charge一次性 transfer，一轮往返内授权并 settlementsingle purchases、one-time API access、当你需要 multi-rail 时替代单个 x402 calls
sessionagent 预授权 cap 和 duration，然后持续流式 metered micropayments 直到关闭高频 micropayments，其中每个 request 都签名很昂贵；recurring subscriptions；「agents 的 Stripe Subscription」model

与 x402 的取舍。

Dimensionx402MPP
Authorization frequencyPer request（每次一个 EIP-3009 signature）Per charge 或 per session（一次 auth，许多 metered calls）
HTTP shapeCustom 402 加 payment headers标准 WWW-Authenticate / Authorization / Payment-Receipt
RailsStablecoin only（Base/Solana/EVM 上的 USDC）Multi-rail（Tempo stablecoin、Lightning、cards、ACH）
Feeszero protocol fee 加 sub-cent gascard rails 上有 Stripe processing fees；Tempo stablecoin 近零
Recurring support有限（需要 per-period signing）通过 session intent 原生支持
Best fitone-off API calls、public infrastructurerecurring subscriptions、enterprise 和 Stripe-integrated merchants、需要 fiat fallback 的 flows

SDK 中的变化。 MPP 通过 Stripe Python SDK 加 Stripe 的 MPP surface 集成。integration 重点在 session management。harness 是概念 8 的同一个 baseline。

note

下面的 from stripe.mpp import MPPSession import 和 stripe.MPPSession 调用是 illustrative。stripe.MPPSession 和 stripe.mpp 不在 Stripe Python SDK 中；MPP 是 Stripe 和 Tempo 标准（mainnet 2026 年 3 月 18 日），通过 Stripe 的 MPP surface 集成。session concept（预授权 cap、流式 metered calls）是真实的；这里的 exact API 是 teaching stand-in。SDK scaffolding 和 typed results 是真实的。

from agents import Agent, function_tool, RunContextWrapper
from agents.tool_guardrails import (
    tool_input_guardrail,
    ToolInputGuardrailData,
    ToolGuardrailFunctionOutput,
)
from decimal import Decimal
import json
import stripe
from stripe.mpp import MPPSession
from .models import MPPSessionResult, MPPMeteredCallResult, PaymentToolResult

@tool_input_guardrail
def verify_mpp_session_authorized(data: ToolInputGuardrailData) -> ToolGuardrailFunctionOutput:
    """Refuse session creation if the user has not pre-authorized this service."""
    args = json.loads(data.context.tool_arguments or "{}")
    service_id = args.get("service_id", "")
    max_total = Decimal(str(args.get("max_total_usd", 0)))
    user_session = data.context.context["user_session"]
    if not user_session.can_authorize_mpp_session(max_total, service_id):
        return ToolGuardrailFunctionOutput.reject_content(
            f"User has not authorized MPP sessions of ${max_total} for service {service_id}"
        )
    return ToolGuardrailFunctionOutput.allow()

@function_tool(tool_input_guardrails=[verify_mpp_session_authorized])
async def mpp_create_session(
    ctx: RunContextWrapper,
    service_id: str,
    max_total_usd: Decimal,
    duration_seconds: int = 3600,
) -> MPPSessionResult:
    """Create an MPP session for one service with a spending cap and duration.
    Returns session_id for the metered calls that follow.
    The verify_mpp_session_authorized guardrail above has already validated consent."""

    user_session = ctx.context["user_session"]
    cents = int((max_total_usd * 100).quantize(Decimal("1")))
    session = stripe.MPPSession.create(
        service_id=service_id,
        max_total_usd=cents,
        duration_seconds=duration_seconds,
        user_session_id=user_session.id,
        # The MPP server picks the rail (stablecoin/Lightning/card) by service preference
    )

    ctx.context.setdefault("mpp_sessions", {})[service_id] = session.id
    return MPPSessionResult(
        session_id=session.id,
        status="active",
        expires_at=session.expires_at,
    )

@function_tool
async def mpp_metered_call(
    ctx: RunContextWrapper,
    service_url: str,
    payload: dict,
    cost_estimate_usd: Decimal,
) -> MPPMeteredCallResult | PaymentToolResult:
    """Make a metered call inside an active MPP session.
    The session's running spend updates automatically; the cap is enforced server-side."""

    service_id = extract_service_id(service_url)
    sessions = ctx.context.get("mpp_sessions", {})
    session_id = sessions.get(service_id)
    if not session_id:
        return PaymentToolResult(
            status="failed",
            error=f"No active MPP session for service {service_id}. Create one first.",
        )

    response = await mpp_client.metered_call(
        url=service_url,
        payload=payload,
        session_id=session_id,
        cost_estimate_usd=cost_estimate_usd,
    )
    return MPPMeteredCallResult(
        session_id=session_id,
        cost_usd=Decimal(str(response.cost_usd)),
        response_payload=response.payload,
        accumulated_session_spend_usd=Decimal(str(response.accumulated_session_spend_usd)),
    )

@function_tool
async def mpp_close_session(ctx: RunContextWrapper, session_id: str) -> MPPSessionResult:
    """Close an MPP session and finalize payment. Returns the total charged and the breakdown by rail."""
    closed = await stripe.MPPSession.close(session_id)
    return MPPSessionResult(
        session_id=session_id,
        status="closed",
        total_charged_usd=Decimal(str(closed.total_charged_usd)),
        rail_breakdown={
            rail: Decimal(str(amount))
            for rail, amount in closed.rail_breakdown.items()
        },
    )

api_consumer_agent = Agent(
    name="APIConsumerAgent",
    instructions="""Consume third-party APIs efficiently using MPP sessions.
    Workflow:
    1. Identify the service to consume
    2. Create an MPP session with mpp_create_session ($X cap, Y seconds duration)
    3. Make metered calls via mpp_metered_call
    4. Close the session with mpp_close_session when done
    Sessions are cheaper than per-request payment for high-frequency calls.""",
    tools=[mpp_create_session, mpp_metered_call, mpp_close_session],
    model="gpt-5.5",
)

真实的部分是 SDK scaffolding 和 typed results。stand-in 的部分是 stripe.MPPSession 和 mpp_client。session lifecycle 是真实 pattern；exact Stripe API 为教学而简化。下面是一个 runnable mock：

from decimal import Decimal

class MockMPPSession:
    @staticmethod
    def create(**kw):
        return type("S", (), {"id": "mpp_sess_1", "expires_at": None})()
    @staticmethod
    async def close(session_id):
        return type("S", (), {"total_charged_usd": Decimal("4.20"),
                              "rail_breakdown": {"stablecoin": Decimal("4.20")}})()
mpp_client = type("Mock", (), {"metered_call": staticmethod(
    lambda **k: type("R", (), {"cost_usd": Decimal("0.05"), "payload": {},
                               "accumulated_session_spend_usd": Decimal("0.05")})())})()

它给 harness 增加什么。 一个配置步骤：你的 Stripe account 需要启用 MPP。Tempo blockchain integration 由 Stripe 的 MPP server 处理，所以 agent 永远不直接接触 Tempo。除了 Stripe SDK 和 key management（与 ACP 相同），没有额外内容。

durable 地运行它（Inngest）。 MPP sessions 很自然地映射到 Inngest 的 long-running function pattern。session lifecycle（create、use、close）会变成一串 step.run blocks，并可用 step.sleep 处理 time-based expiry。session model 和 durable execution 组合得很好：两者都是为 stateful、multi-step work 构建的。

团队常犯的错。 创建过大或过长的 sessions。session cap 是出错时的 loss limit。一个为了「convenience」设置的 $1,000 session，会让你暴露在 $1,000 的 agent-loop-gone-wrong 损失下。请按真实预期工作量调整 session：一个 30-call task，用持续 5 分钟的 $5 session，比用持续 1 小时的 $50 session 更安全。

第 4 部分：Composition rules，什么时候把哪些协议放在一起​

第 3 部分逐个讲了四个协议。现在把它们重新放回一起。第 1 部分已经见过这个想法：真实系统会同时使用多个协议，每层一个。第 4 部分给出规则：哪些组合可行，哪些会失败，以及如何让 stack 尽可能贴合任务、不多不少。

概念 12：minimum viable agent-payment stack​

一句话： 正确 stack 是能为一个 use case 交付价值的最小协议集合，不是每层都堆满四个协议。

组合任何东西前，先问一个问题：这里能交付价值的最小 stack 是什么？答案几乎从来不是「全部四个协议、每层都上」。多数 production systems 会从一层完整接好开始，只有 use case 迫使时才加其他层。

下面是每个常见 use case 的最小 stack。它也预告了第 5 部分的五个决策。

Use caseLayer 1（Discovery）Layer 2（Auth）Layer 3（Commerce）Layer 4（Settlement）为什么这是 MVP
Consumer shopping（agent 为人购买 retail goods）AI shopping surface，或带 merchant catalog 的 MCP serverACP SPT（regulated fields 中也可用 AP2 mandate）ACP通过 Stripe 的 card rails多数 buyer 想要 chargeback cover。ACP 加 Stripe 是今天可交付的路径。
API-paying agent（agent 调用会收费的 third-party APIs）支持 x402 的 MCP server，或 Agent.market directoryEIP-3009 signature（x402 默认）无：direct API callBase 或 Solana 上的 x402对 machine-to-machine，Layers 2 和 4 会合并。没有购买 lifecycle。
Enterprise procurement（agent 在规则下向 approved suppliers 采购）partner network 内的 A2A discoveryAP2 Intent Mandate（audit 必需）catalog suppliers 用 ACP 或 UCP；service buys 用 direct APIrecurring 用 MPP sessions；one-off 用 card rails 上的 ACP SPTaudit trail 是不能跳过的部分。AP2 mandates 必需。
Multi-agent marketplace（agent 雇用其他 agents）A2A 或 agent directoryAP2 mandate 加 ERC-8004 reputation check无：direct agent-to-agentx402（最常见），或 Stripe 已接入时用 MPPtrust 双向运行。双方都需要 verifiable identity 和 payment proof。

composition rule。 在每一层选择 use case 要求的协议。不要在 use case 根本不触碰的层增加协议。纯 API-paying agent 不需要 ACP。consumer-shopping agent 不该为了一件 $50 t-shirt 伸手拿 x402，因为 cards 上的 chargeback cover 值得 2.9% 的 fee。

陷阱。 一些团队会为了「flexibility」一次性接 ACP、AP2、x402 和 MPP。结果是 integration surface 扩大四倍，却无法清楚回答「什么时候触发哪个协议」。选一个 stack。交付它。只有第二个 use case 真的需要时，再增加第二个 stack。

概念 13：什么时候协议跨层组合，什么时候同层竞争​

一句话： 不同层的协议天生要 stack 在一起；同一层的协议天生要互相替代，所以唯一问题是两个协议是否位于同一层。

两个协议之间只有两种关系。要么它们 compose，因为位于不同层且为 stack 而设计；要么它们 compete，因为位于同一层且为替代而设计。多数架构混乱都来自把第二种当成第一种来读。

跨层时，天生可以 compose：

CompositionLayer mapping哪里会交付
AP2 + ACPAP2 在 Layer 2（audit-grade auth），ACP 在 Layer 3（commerce）ACP 正常 flow 需要额外 audit 的 regulated fields
AP2 + x402AP2 在 Layer 2（mandate auth），x402 在 Layer 4（stablecoin settlement）通过 a2a-x402 extension 实现仍需 audit 的 crypto-native flows
ACP + x402ACP 在 Layer 3（commerce），x402 在 Layer 4，用于 consumer purchase 内部的 machine-to-machine sub-flowsconsumer buy 内含一些 API spend 的 hybrid platforms
MCP + x402（通过 withX402Client）MCP 在 Layer 1（discovery），x402 在 Layer 4（settlement）Cloudflare 对 paid MCP tools 的 standard pattern

同一层内，天生竞争：

CompetitionLayer何时各自胜出
AP2 vs. ACP SPT vs. TAPLayer 2AP2 用于 audit-grade flows；ACP SPT 用于 Stripe-wired consumer flows；TAP 用于 identity-only checks
ACP vs. UCPLayer 3ACP 用于 ChatGPT reach；UCP 用于 Gemini reach；cross-surface sellers 可同时支持两者
x402 vs. MPPLayer 4x402 用于 one-off micropayments 和纯 stablecoin flows；MPP 用于 sessions、subscriptions 和 multi-rail flows

测试方法。 当你卡在两个协议之间时，只问一件事：它们在同一层吗？如果是，就选一个，或为不同 sub-flows 付出支持两者的成本。如果不是，它们很可能 compose，正确设计也常常会同时使用两者。

一个 worked composition：AP2 + x402 stack。 这是 crypto-native pattern，如今常见于 B2B 和 agent-to-agent flows：

Layer 1 (Discovery): A2A directory inside the partner network
Layer 2 (Auth):      AP2 Intent + Cart + Payment Mandates
Layer 3 (Commerce):  Often none (direct service request), or ACP for a catalog
Layer 4 (Settlement): x402 via the a2a-x402 extension

SDK assembly 只是组合第 3 部分中的 per-protocol tools。这些 tools（a2a_discover_partners、ap2_create_intent_mandate 等）已在概念 8 到 11 中定义；这里只是把它们接到一个 agent 上。

agent = Agent(
    name="EnterpriseB2BAgent",
    instructions="...",
    tools=[
        # Layer 1: Discovery
        a2a_discover_partners,
        # Layer 2: Authorization
        ap2_create_intent_mandate,
        ap2_create_cart_mandate,
        # Layer 4: Settlement (a2a-x402 composes Layers 2 and 4)
        ap2_settle_via_x402,
    ],
    model="gpt-5.5",
    # No commerce tools: direct B2B procurement.
)

概念 14：Cost and latency，真正迫使选择的因素​

一句话： transaction size 和可接受等待时间决定 settlement protocol，因为 card fees 会压垮小额支付，而 slow checkout 会破坏紧密 loop。

每种 composition 都有价格和速度。consumer-shopping flow 走 ACP 加 card rails，每笔约 2.9% + $0.30，端到端 5–30 秒。API-paying agent 走 x402，成本低于 1 美分，用时 1–2 秒。正确 composition 部分取决于你的 use case 能承受什么成本、什么等待。

按 composition 看每笔交易成本。

Composition典型每笔成本典型 latency
ACP + card rails（consumer shopping）2.9% + $0.30（Stripe rate）5–30 秒
ACP + MPP sessions（subscriptions）cards 上 2.9%，或 Tempo stablecoin 上约 0.5%，按 session每次 metered call 1–3 秒
AP2 + x402（B2B stablecoin）sub-cent gas、zero protocol fees2–5 秒（mandate signing 增加 1–3 秒）
x402 only（API-paying）sub-cent gas、zero protocol fees1–2 秒
MPP sessions only（recurring API）Tempo stablecoin 近零；cards 上按 Stripe rateactive session 内每次 metered call 50–500 ms

钱会迫使什么。 fees 超过 transaction 的约 5% 就有问题。一笔 $0.05 API call 如果支付 2.9% + $0.30 的 card fees，手续费比调用价值还高，这是使用 x402 或 MPP stablecoin 的明确信号。$50 t-shirt 付同样的 2.9% + $0.30 就可以接受。card rails 停止合理的金额线大约在 $5–$10。低于它，machine-payment rails 胜出；高于它，cards 的 chargeback cover 通常值得 fee。

latency 会迫使什么。 user-facing step 中超过 5 秒的等待就是问题。AP2 mandate signing 可增加 1–3 秒；如果等待 human signature 会更久；ACP checkout 增加 5–30 秒。对没有 human in the loop 的 agent-to-agent flows，预算更紧，常常是 sub-second，因此 Base 上的 x402 和 Tempo 上的 MPP 会成为默认选择。

压缩版 decision tree。

What is the transaction value?
├── Sub-dollar (per-call API, per-token billing)
│   → x402 only, or MPP sessions
├── $1 to $10 (small, low-stakes buys)
│   → x402 or MPP, with AP2 audit if you need it
├── $10 to $1,000 (consumer purchases)
│   → ACP + card rails (chargeback cover is worth the fee)
└── $1,000+ (B2B, enterprise procurement)
    → AP2 + ACP/UCP + MPP sessions, or bank rails

What is the latency budget?
├── Sub-second (multi-agent loops)
│   → MPP sessions on Tempo, or x402 on Base
├── 1 to 5 seconds (interactive)
│   → x402 or MPP; AP2 only if mandates are pre-signed
└── 5+ seconds (an acceptable user wait)
    → Full ACP checkout works

第 5 部分：decision lab，五个 worked examples​

第 2 到第 4 部分给了 framework：四层、每层几个协议、跨层 composition。第 5 部分走 5 个真实决策。每个都会展示完整 reasoning：use case 触及哪些层、每层选哪个协议、为什么，以及 agent code 长什么样。

四种 use case 映射到四层。行：consumer shopping、API-paying agent、enterprise procurement、multi-agent marketplace。列：Discovery、Authorization、Commerce、Settlement。consumer shopping 使用 AI shopping surface、ACP shared payment token、ACP 和 card rails。API-paying agent 使用 MCP 和 directory、EIP-3009 signature、无 commerce layer、Base 上的 x402。enterprise procurement 使用 internal MCP、AP2 mandates、ACP 加 direct B2B、MPP sessions 加 cards。multi-agent marketplace 使用 A2A、AP2 加 ERC-8004、无 commerce layer、x402。没有单一协议能覆盖全部四行；没有任何一行少于三个协议。正确问题不是「哪个协议」，而是「这个 use case 在每一层用哪个协议」。

横向读一行，看一个 use case 的完整 stack。纵向读一列，看一个层在 use case 变化时如何变化。下面五个 decisions 会详细走完这些行。前四个覆盖 matrix；第五个用完全不同 toolchain 重建其中一个，证明 framework 不绑定任何 vendor。

Decision 1：Consumer shopping agent（ChatGPT Instant Checkout pattern）​

use case。 你正在构建一个 agent，帮助人们在 Walmart、Etsy 和 Shopify sellers 这类 ACP-enabled merchants 上购物。用户说自己想要什么；agent 搜索 catalogs、展示 options，并在确认后 checkout。single buyer 到 single merchant，每单 $5–$500，需要 refunds 和 chargebacks。

走四层。

  • Layer 1（Discovery）。 agent 必须跨多个 merchants 找商品。你可以集成每个 merchant 的 MCP catalog，也可以使用已经聚合 ACP merchants 的 ChatGPT Shopping surface。选择：AI shopping surface，因为 discovery 工作已经完成，自己接入一百万个 catalogs 不现实。
  • Layer 2（Authorization）。 用户已登录 surface，并且每次购买都确认，所以 authorization 很简单。选择：ACP SPT，由 Stripe 按每笔 purchase mint，scope 到一个 merchant、一个 amount 和 10-minute window。
  • Layer 3（Commerce）。 完整 lifecycle 很重要：cart、checkout、fulfillment、disputes、refunds。选择：ACP，这正是它为之构建的协议。
  • Layer 4（Settlement）。 $5–$500 正好落在 card rail 的舒适区，chargeback cover 值得 fee。选择：通过 Stripe 的 card rails，这是 ACP 默认假设的 rail。

implementation。 这是概念 8 的代码；tools 来自概念 8。

shopping_agent = Agent(
    name="ShoppingAgent",
    instructions="""Help the user shop. Workflow:
    1. Use acp_browse_merchant to find products matching the request
    2. Show matched items; wait for the user to confirm
    3. On confirm, use acp_create_cart_and_checkout to buy
    4. Use acp_check_order for status
    5. Use acp_refund only when the user asks""",
    tools=[acp_browse_merchant, acp_create_cart_and_checkout, acp_check_order, acp_refund],
    model="gpt-5.5",
)

生产中最可能失败的地方。 cart mismatches：agent 构建的 cart 与 request 不一致（「我说红色，结果是粉色」）。修复：在 SPT mint 前要求用户确认 cart，记录 cart-accuracy，并在 accuracy 低于 95% 时调优 instructions。

durable 地运行它（Inngest）。 对 cart-confirmation gate 使用概念 8 的 step.wait_for_event pattern，再加 per-user concurrency cap。

先选这个。 这是今天已经 live 的 use case：ChatGPT Instant Checkout、ACP ecosystem、每个打开 ACP 的 Shopify merchant。如果你只能交付一个 composition，先交付它。

Decision 2：API-paying research agent（x402-only pattern）​

use case。 你正在构建一个 research agent，为 third-party APIs 付费获取数据：financial feeds、news、specialized search。它通过 Agent.market 这类 directory 在 runtime 中发现 paid APIs，权衡 cost 和 value，并按 call 付款。$0.001–$0.50 的高频 micropayments，任务开始后没有 human in the loop，也没有 commerce lifecycle。

走四层。

  • Layer 1（Discovery）。 Agent.market 和类似 x402-paywalled directories 让 agent 能在 runtime 中找到 services；支持 x402 的 MCP servers 处理预先集成的服务。选择：Agent.market 加 MCP-via-Cloudflare，runtime discovery 旁边保留预接好的 fallback set。
  • Layers 2 and 4（Authorization and Settlement，合并）。 任务开始后没有人类。wallet 的 on-chain caps 限制 per-transaction spend；user-level caps 通过每个 paying tool 上的 SDK tool_input_guardrail 运行。EIP-3009 signature 同时是 authorization 和 settlement。选择：Base（USDC）上的 x402，不使用单独 mandate protocol。
  • Layer 3（Commerce）。 「purchase」只是一次 API call。选择：无，通过 x402 direct API access。

implementation。 这是概念 10 的代码；tools 来自概念 10。

research_agent = Agent(
    name="ResearchAgent",
    instructions="""Research the user's query by paying for data via x402.
    1. Use x402_search_agent_market to find relevant paid services
    2. Use x402_fetch to pull data (max $0.10 per call)
    3. Write up the findings
    Stay under $10 per session.""",
    tools=[x402_fetch, x402_search_agent_market],
    mcp_servers=[research_mcp_with_payments],
    model="gpt-5.5",
    # Spend caps live on x402_fetch via tool_input_guardrails
    # (Concept 10's enforce_x402_session_cap), not on the agent.
)

生产中最可能失败的地方。 stuck loop 造成 runaway spend：agent 反复获取同一数据并烧掉 budget。修复：wallet 的 on-chain caps（真正保护你的 safety）、SDK session-spend guardrails，以及 dedup cache，让相同 fetch 不再重复付费。

durable 地运行它（Inngest）。 step.run memoization 在这里很有价值。session 中途 crash 时，retry 会带着已付款数据继续。per-user concurrency cap 防止一次 burst 接管系统。

纯 machine-to-machine。 Layers 2 和 4 合并为一个 signature，Layer 3 为空。这个 stack 在结构上比 Decision 1 简单：协议更少、integration points 更少、每次调用成本更低。取舍是没有 chargeback cover，也没有 commerce semantics；这没问题，因为 use case 两者都不需要。

Decision 3：Enterprise procurement agent（AP2 plus composed-stack pattern）​

use case。 你正在为金融服务领域的 regulated enterprise 构建 procurement agent。buyers 委托任务：「周五前，从 approved suppliers 购买 50 个 ergonomic keyboards，总额低于 $5,000。」audit trail 是法律要求，spend caps 在多个层级运行，suppliers 已预先批准，所以没有 runtime discovery。

走四层。

  • Layer 1（Discovery）。 supplier list 事先已知。选择：带 supplier catalogs 的 internal MCP server，因为 discovery scope 有边界。
  • Layer 2（Authorization）。 non-repudiable audit trail 是不能跳过的部分；non-repudiable record 是签署者之后不能否认的记录。AP2 mandates 正好提供这一点。选择：AP2，在 task creation 时有 Intent Mandate，checkout 前有 Cart Mandate，settlement 时有 Payment Mandate，每个都由 procurement officer 签署。
  • Layer 3（Commerce）。 大供应商暴露 ACP 或 UCP；小供应商暴露 direct B2B APIs。选择：ACP 用于 ACP-enabled suppliers，其余用 direct API；agent 两者都处理。
  • Layer 4（Settlement）。 recurring suppliers 偏向 MPP sessions；one-off buys 偏向 card rails，因为较高金额下需要 chargeback cover。选择：recurring 用 MPP sessions，one-off 用 ACP SPT 加 card rails，按 supplier history 选择。

implementation。 这会组合概念 8、9 和 11 中的 tools。

procurement_agent = Agent(
    name="ProcurementAgent",
    instructions="""Run procurement under audit-grade compliance:
    1. ALWAYS create an Intent Mandate first via ap2_create_intent_mandate
    2. Search approved suppliers via the internal MCP server
    3. Build the cart and create a Cart Mandate via ap2_create_cart_mandate
    4. Recurring suppliers: use an MPP session.
       One-off buys: use ACP plus card rails via ap2_settle_via_acp
    5. Record every mandate ID in the procurement audit log""",
    mcp_servers=[approved_suppliers_mcp],
    tools=[
        ap2_create_intent_mandate,
        ap2_create_cart_mandate,
        ap2_settle_via_acp,
        mpp_create_session,
        mpp_metered_call,
        mpp_close_session,
    ],
    model="gpt-5.5",
    # Caps and mandate rules run via tool_input_guardrails on the paying tools
    # (Concepts 9, 11, 15). require_intent_mandate on ap2_create_cart_mandate blocks
    # any cart with no prior Intent Mandate; enforce_per_run_spend_cap blocks any
    # payment over the user's run cap.
)

生产中最可能失败的地方。 Intent Mandate scope mismatches：officer 签署 mandate，agent 完成工作，然后 cart 不符合 mandate。修复：在 agent 开始 shopping 前验证 mandate scope（概念 9 的「先创建 Intent Mandates」），并拒绝 scope 超出用户可授权范围的 tasks。

durable 地运行它（Inngest）。 AP2 mandate signing 天然适合 step.wait_for_event。multi-stage tasks 对每个 stage 使用 step.run，再加 per-user concurrency cap。

regulated industry。 audit rules 迫使 Layer 2 使用 AP2；许多 suppliers 迫使 Layer 3 同时使用 ACP 和 direct APIs；recurring versus one-off 迫使 Layer 4 同时使用 MPP 和 cards。这个 composition 比 Decisions 1 和 2 更重，而 audit requirement 正是它的正当理由。

Decision 4：Multi-agent marketplace（AP2 plus x402 plus ERC-8004 pattern）​

use case。 你正在构建一个 agents 可以雇用其他 agents 的平台。Agent A 需要 research；Agent B 以 x402 service 的形式出售 research。双方尚未互信，transactions 必须可验证，payment 是 pure crypto-native，没有 cards。agent 到 agent，每笔 $0.10–$100，双方都需要 verifiable identity。

走四层。

  • Layer 1（Discovery）。 Agent B 通过 A2A 发布 capability；Agent A 找到它。选择：A2A，这正是为此构建的协议。
  • Layer 2（Authorization）。 transaction time 没有人类，但双方 trust 必须可验证。AP2 mandates 证明 user consent；ERC-8004 给 Agent B 一份 Agent A 可先检查的 on-chain reputation。选择：AP2 加 ERC-8004，组合出完整 bilateral verification。
  • Layer 3（Commerce）。 没有 carts，没有 refunds，只有「做这个 task 并交付 report」。选择：无，direct A2A request and response。
  • Layer 4（Settlement）。 crypto-native、sub-second、不需要 chargeback cover。选择：通过 AP2 的 a2a-x402 extension 使用 x402。

implementation。 这会把概念 9 和 10 的 tools 与 A2A discovery 组合起来。

researcher_hiring_agent = Agent(
    name="ResearcherHiringAgent",
    instructions="""Hire research-specialist agents to work for you:
    1. Use a2a_discover_researchers to find available agents
    2. Check ERC-8004 reputation (>50 successful jobs, no flagged disputes)
    3. Create an Intent Mandate scoped by amount and recipient
    4. Submit the task via A2A with the mandate attached
    5. Receive the result; settle via ap2_settle_via_x402""",
    tools=[
        a2a_discover_researchers,
        erc8004_check_reputation,
        ap2_create_intent_mandate,
        a2a_submit_task_with_mandate,
        ap2_settle_via_x402,
    ],
    model="gpt-5.5",
)

生产中最可能失败的地方。 信任了被刷出来的 reputation：ERC-8004 scores 可审计，但 operator 可以用小额成功 jobs 刷高。修复：把 reputation 与其他 signals 结合（operator identity、transaction-volume thresholds、dispute history），并对 first-time counterparties 超过固定金额时增加 human review。

durable 地运行它（Inngest）。 一次雇用多个 specialists 时 fan out；对每个 result 使用 step.wait_for_event，每个 stage 使用 step.run。这个 decision 会触碰每个 Inngest primitive。

纯 multi-agent economy。 transaction time 没有人类 in the loop；两个 agents 都在 pre-authorized scopes 内行动。这正是 agent economy 正在围绕构建的形状，也有最多 failure modes，因为没有既有关系的 bilateral trust 很难，protocol stack 也只能部分解决它。

Decision 5：Non-Stripe、non-OpenAI stack（证明 framework 可迁移）​

到目前为止，每个代码示例都用了 stripe.PaymentTokens.create(...) 和 OpenAI Agents SDK。它们是 ACP 和 SDK 最成熟的 integrations，SDK 也是本课 runtime，但四层 architecture 本来就是 stack-agnostic 的；只展示一个 stack 的课程还没有证明这一点。所以这个 decision 会用完全不同 toolchain 重建 Decision 2，也就是 API-paying research agent：用 Google Agent Development Kit（ADK）做 runtime，用 Coinbase smart-contract wallet 做 on-chain identity，用 AP2 mandates 做 authorization，用 direct x402 做 settlement。零 Stripe，零 OpenAI。

use case 仍然是 Decision 2 的：research agent 每次 call 付 $0.001–$0.10，每个 session cap 为 $10。sub-dollar、authorization 后无人类、machine-to-machine settlement。architecture 也保持 Decision 2：x402 合并 Layers 2 和 4，没有 commerce layer，用 MCP 做 discovery。只有 library 变了。

note

下面的代码块是 illustrative。Google ADK 结构（Agent、tool decorator）是真实的，google-adk 是真实 package。payment pieces 是 stand-ins：AP2 的真实 package 是 ap2，mandate fields 差异很大且没有 MandateSigner class；Coinbase wallet 通过 coinbase_agentkit 的 SmartWalletProvider 配置；live x402 call 需要 funded account 和真实 402 endpoint。不会移动真实资金。重点是形状，不是 runnable buyer。

from google.adk import Agent
from google.adk.tools import function_tool  # ADK's tool decorator
from coinbase_agentkit import AgentKit, SmartWalletProvider
from decimal import Decimal
from datetime import datetime, timedelta

# Shared result models come from the Pydantic sidebar in Part 3.
from .models import X402PaymentResult, PaymentToolResult, DiscoveryResult

# --- Illustrative stand-ins for the payment rails (real APIs differ) ---
class MockMandate:
    def __init__(self, mid, rules=None): self.id, self.rules = mid, (rules or {})
class MockSigner:
    async def sign(self, mandate): return mandate          # real AP2 signs over A2A
agent_market_client = type("Mock", (), {"search": staticmethod(lambda **k: [])})()
x402_client = type("Mock", (), {})()                       # real client: x402-client
# -----------------------------------------------------------------------

# Layer 2 (Authorization): a Coinbase smart-contract wallet gives the agent its
# on-chain identity and its spend caps. This is the analog of a Stripe customer
# plus per-customer caps, but enforced by the chain.
wallet_provider = SmartWalletProvider(
    config={
        "chain": "base-mainnet",
        "spend_limits": {
            "per_transaction_usdc": Decimal("0.50"),
            "per_session_usdc": Decimal("10.00"),
            "per_day_usdc": Decimal("100.00"),
        },
    },
)
agent_kit = AgentKit(wallet_provider=wallet_provider)

# Layer 2 (Authorization): an AP2 Intent Mandate, signed by the user at session start.
# The analog of a signed, revocable Stripe authorization, but declarative.
async def create_research_intent_mandate(user_did, user_signer, session_cap_usdc):
    mandate = MockMandate(
        "intent_mock_1",
        rules={
            "max_total_usd": str(session_cap_usdc),
            "allowed_categories": ["data-api", "research-service"],
            "expires_at": (datetime.utcnow() + timedelta(hours=1)).isoformat(),
        },
    )
    return await user_signer.sign(mandate)

# Layer 1 (Discovery) + Layer 4 (Settlement) as one tool.
# ADK's @function_tool is the analog of the SDK's @function_tool.
@function_tool
async def x402_paid_fetch(url: str, max_payment_usdc: Decimal) -> X402PaymentResult | PaymentToolResult:
    """Fetch a URL that may need x402 payment up to max_payment_usdc.
    The wallet handles the signature; the on-chain cap is the safety that protects you."""
    # ADK has no tool_input_guardrail, so the check runs in-tool,
    # backed by the wallet's on-chain cap (the layer nothing can bypass).
    resp = await x402_client.get(url, max_payment_usdc=max_payment_usdc)
    return X402PaymentResult(
        content=resp.content,
        amount_paid_usdc=Decimal(str(resp.amount_paid_usdc)),
        tx_hash=resp.tx_hash,
    )

@function_tool
async def search_agent_directory(query: str, max_price_per_call_usdc: Decimal) -> list[DiscoveryResult]:
    """Search Agent.market for x402-paywalled services."""
    results = await agent_market_client.search(query=query, max_price_per_call_usdc=max_price_per_call_usdc)
    return [
        DiscoveryResult(
            service_id=r.service_id,
            name=r.name,
            description=r.description,
            price_per_call_usdc=Decimal(str(r.price_per_call_usdc)),
            endpoint_url=r.endpoint_url,
        )
        for r in results
    ]

# The agent itself: a Google ADK Agent, the direct analog of the SDK's Agent.
research_agent = Agent(
    name="research-agent",
    description="Research the user's query by paying for data via x402.",
    instructions="""Research the query.
    1. Use search_agent_directory to find relevant paid services
    2. Use x402_paid_fetch to pull data ($0.50 max per call)
    3. Write up the findings
    Stay under $10 per session.""",
    model="deepseek-v4-flash",  # illustrative; any ADK-compatible model id works
    tools=[search_agent_directory, x402_paid_fetch],
)

# Driver: the user signs the Intent Mandate once at session start;
# the agent then runs on its own inside the mandate's scope until the session ends.
async def run_research_session(user_did, user_signer, query):
    intent = await create_research_intent_mandate(
        user_did=user_did,
        user_signer=user_signer,
        session_cap_usdc=Decimal("10.00"),
    )
    return await research_agent.run_async(
        query,
        context={"intent_mandate": intent, "wallet": agent_kit},
    )

逐行 translation。 左边是 Decision 2 的 OpenAI 加 Stripe code，右边是 Decision 5 的 Google ADK 加 Coinbase code。这张表就是 payoff：每一行都是同一个 concept 换了不同名称。

Decision 2（OpenAI + Stripe）Decision 5（Google ADK + Coinbase）同一概念，不同 library
from agents import Agent, function_toolfrom google.adk import Agent + from google.adk.tools import function_toolAgent runtime
@function_tool（OpenAI Agents SDK）@function_tool（Google ADK）Tool decorator
RunContextWrappercontext={...} kwarg to run_asyncPer-run state
stripe.Customer.modify(...) for capsSmartWalletProvider(spend_limits={...})Spend caps，chain-native
tool_input_guardrail decoratorin-tool check + wallet capsPre-execution validation
Runner.run(agent, ...)agent.run_async(...)Agent execution

保持相同的东西。 architecture：Layer 1 是 MCP 或 directory，Layer 2 是 mandate 加 wallet caps，Layer 3 无（machine-to-machine 会折叠 commerce），Layer 4 是 x402。primitives：Intent Mandate、EIP-3009 signatures、402 responses、payment-signature headers、on-chain spend caps。survive library swap 的是 framework。

两个真实 operational differences。

  1. Google ADK 没有 first-class tool input guardrail（截至 2026 年中）。 SDK 的 tool_input_guardrail 对 pre-execution checks 确实方便；ADK 的 tool decorator 暂时没有直接等价物。workaround 是 in-tool validation，加 wallet 的 on-chain caps 支撑。caps 仍会保护你；in-tool check 只是更快失败。选择 ADK，就是用 guardrail convenience 换另一种 multi-agent story。
  2. AP2 mandate signing 在这里更 native。 Google 构建了 AP2，所以 ADK ecosystem 更干净地集成 mandate-signing UI flows。如果你的 use case 强依赖 mandate-rigorous authorization（Decisions 3 和 4），ADK 加 AP2 是真实 fit，不只是替代。

Decision 5 的底线： 四层 architecture 不是伪装成 Stripe-and-OpenAI 故事。Discovery、Authorization、Commerce、Settlement，每层选一个协议，并按 use case 证明它：这一点能穿过任何 library swap。Decision 5 用 Google ADK 和 Coinbase wallet 构建了与 Decision 2 完全相同的 composition；只有 imports 改了。如果一个 framework 无法用不同 stack 表达，那只是穿着 costume 的 library tutorial。本课不是。

第 6 部分：Production concerns，系统上线后真正会杀你的东西​

第 1 到第 5 部分建立了 framework，并走过真实 decisions。第 6 部分覆盖决定你的 stack 能否扛住真实用户的东西。这些 failure 不会出现在 demo 中，却会在凌晨 2 点出现。

Reader track

如果你只是为理解而读，不是为了交付，可以略读第 6 部分。如果你真的要构建这些东西，这里才进入现实。这里四个概念，是「系统能工作」和「系统会耗尽 wallet」之间的差别。

概念 15：三层架构级 spend-limit enforcement​

一句话： 通过三个相互独立的位置强制限制，阻止 agent 超额花费；这样一个位置的 bug 会被另外两个抓住。

agent-commerce system 最严重的失败方式很简单：agent 花得比允许的更多。一个 stuck agent loop 能在几秒内耗尽 wallet。每个协议都有自己的 cap（ACP 的 SPT amount、MPP 的 session cap、x402 的 per-request max），但没有任何单一 cap 本身足够。production systems 会在三个独立层级执行 spend limits。

Level 1：wallet 和 payment-method limits。 这是真正保护你的 cap。agent 的 smart-contract wallet（用于 x402）或 Stripe customer account（用于 ACP 和 MPP）携带 infrastructure level 设置的 spend caps。chain 或 Stripe 会强制执行它们，不管 agent code 做了什么。当 agent loop 完全失控时，只有这一层仍然有效。

对使用 smart-contract wallet 的 x402：

note

下面的 SmartContractWallet.deploy(...) 调用是 illustrative。它展示 Level 1 caps 放在哪里，不是真实 pip package。实践中，你会在真实 smart-contract wallet 上设置这些 caps（例如通过 Coinbase AgentKit 的 SmartWalletProvider）。三层纪律才是真正课程重点。

from decimal import Decimal

# Set ONCE when the wallet is deployed. The agent cannot change this.
wallet_spend_limits = {
    "max_per_transaction_usdc": Decimal("10.00"),    # cap per single transfer
    "max_per_day_usdc": Decimal("100.00"),           # rolling 24-hour cap
    "max_per_merchant_usdc": Decimal("50.00"),       # cap to any single recipient
}
agent_wallet = SmartContractWallet.deploy(
    owner=user_did,
    spend_limits=wallet_spend_limits,
    chain="eip155:8453",  # Base
)

对使用 Stripe 的 ACP 或 MPP，cap 位于 Stripe customer 上：

stripe.Customer.modify(...) 是真实 Stripe API call。它会针对你的 Stripe account 运行。

# Set once via the Stripe Dashboard or API. The caps live in Stripe's infrastructure.
stripe.Customer.modify(
    user_session.stripe_customer_id,
    metadata={
        "max_per_session_usd": "500",
        "max_per_day_usd": "2000",
    },
)
# When an SPT or MPP session is minted above these, Stripe rejects it at the API level.

Level 2：SDK tool guardrails。 OpenAI Agents SDK 的 tool_input_guardrail 会在每个 payment tool 执行前运行，并可以拒绝调用。概念 5 已见过它：这是 SDK-native 的方式，在付款发生前阻止付款，也是能及时运行的 guardrail family。这里给出完整版本，因为这是整门课的 canonical guardrail block。下面代码真实可运行。

import json
from decimal import Decimal
from agents import Agent, function_tool, RunContextWrapper
from agents.tool_guardrails import (
    tool_input_guardrail,
    tool_output_guardrail,
    ToolInputGuardrailData,
    ToolOutputGuardrailData,
    ToolGuardrailFunctionOutput,
)

# Tool INPUT guardrail: pre-payment check. Runs BEFORE the tool executes.
@tool_input_guardrail
def enforce_per_run_spend_cap(data: ToolInputGuardrailData) -> ToolGuardrailFunctionOutput:
    """Reject any payment tool call where the run's total spend would exceed the user's cap.
    Runs before the tool executes, the only guardrail family that can stop a payment in time."""
    args = json.loads(data.context.tool_arguments or "{}")          # raw JSON args string -> dict
    requested = Decimal(str(args.get("max_total_usd") or args.get("max_payment_usdc") or 0))
    ctx = data.context.context                                       # the run context (a dict)
    cap = Decimal(str(ctx["user_session"].per_run_spend_cap_usd))
    spent = Decimal(str(ctx.get("run_spend_usd", 0)))
    if spent + requested > cap:
        return ToolGuardrailFunctionOutput.reject_content(
            f"Refused: would spend ${spent + requested}, run cap is ${cap}"
        )
    return ToolGuardrailFunctionOutput.allow()

# Tool OUTPUT guardrail: post-payment check. Runs AFTER the tool executes.
# Useful for verifying receipts (paid more than expected, wrong amount, etc.).
@tool_output_guardrail
def verify_receipt_integrity(data: ToolOutputGuardrailData) -> ToolGuardrailFunctionOutput:
    output = data.output or {}
    if isinstance(output, dict) and "amount_paid_usdc" in output:
        # Cross-check the receipt against what we asked for.
        pass
    return ToolGuardrailFunctionOutput.allow()

# Attach BOTH guardrails to every payment-authorizing tool.
@function_tool(
    tool_input_guardrails=[enforce_per_run_spend_cap],
    tool_output_guardrails=[verify_receipt_integrity],
)
async def x402_fetch(
    ctx: RunContextWrapper,
    url: str,
    max_payment_usdc: Decimal,
) -> "X402PaymentResult":
    ...

# An agent-level output_guardrail is fine for final-reply safety
# (like redacting PII in the agent's answer), but it does NOT prevent payments.
agent = Agent(
    name="ShoppingAgent",
    tools=[x402_fetch],
    # output_guardrails=[response_safety_guardrail],  # different job, not payment safety
)
使用正确的 guardrail family

SDK 有三种 guardrail families。Input guardrails 运行在用户给 agent 的第一条消息上。Output guardrails 运行在 agent 的最终回复上。Tool guardrails（tool_input_guardrail 和 tool_output_guardrail）在每个 custom tool call 上运行。对于 payment safety，你具体需要 tool_input_guardrail：只有它会在 payment tool 运行前触发并能阻止它。用 output_guardrail 控制 spend，是 agent-commerce code 中最常见的错误。等它触发时，钱已经走了。

Level 3：application 和 business-logic limits。 你自己的代码执行 user-specific rules：per-user daily caps、per-category caps、allowed merchants。这是 business rules 所在的位置。「这个用户每天可以在任意 merchant 花 $500，但在 unverified merchants 上每天最多 $50。」这条规则属于这里，不属于某个 protocol。下面是普通 Python，真实可用。

from decimal import Decimal

class UserSession:
    def can_spend(self, amount_usd: Decimal, merchant_id: str) -> bool:
        # Per-day cap
        if self.today_spend_usd + amount_usd > self.daily_cap_usd:
            return False
        # Per-merchant cap
        merchant_cap = self._merchant_cap_for(merchant_id)
        if self.merchant_spend_usd[merchant_id] + amount_usd > merchant_cap:
            return False
        # Per-category cap (for example, "office supplies" vs "personal")
        category = self._category_for(merchant_id)
        if self.category_spend_usd[category] + amount_usd > self.category_caps[category]:
            return False
        return True

每一层都位于不同基础设施。Level 1 在 chain 或 Stripe 中。Level 2 在 agent SDK 中。Level 3 在你的 application code 中。一个层的 bug 会被另外两层抓住。跳过 Level 1，一个 agent-loop bug 就能耗尽整个 wallet。跳过 Level 2，你会失去 mid-flight abort run 的能力。跳过 Level 3，你无法执行 per-user 或 per-category policy。

陷阱是只信任 protocol caps。ACP 的 SPT cap、MPP 的 session cap 和 x402 的 per-request max 都是 protocol-level limits。它们能阻止特定 protocol abuse，但不会跨 protocols 聚合。一个只用 $50 cap 的 ACP SPTs 的团队，无法阻止 agent 连续 mint 100 个 SPT，总额 $5,000。上面三层存在的原因，正是 protocol caps 不会 aggregate。

概念 16：Agent identity hygiene：keys、wallets 和 audit logs​

一句话： agent 的 signing key 是 authorized spending 与 fraud 之间唯一的分隔线，所以要保护 key、按 agent 分离、轮换，并把每次 spend 记录到 durable storage。

agent commerce 带来一种 autonomous systems 特有的失败：agent 的 cryptographic identity 是真实 spending 与 fraud 之间的全部边界。如果 signing key 泄露，wallet（或 Stripe customer，或 AP2 mandate signer）就可能在你 rotate key 前被耗尽或冒充。identity hygiene 是防止这种失败的一组习惯，共有四个。

1. Per-agent wallet separation。 每个 agent，或每类 agent，都有自己的 wallet 或 payment handle。绝不要让不同工作的 agents 共享 signing keys。如果 shopping agent 和 procurement agent 共用一个 wallet，任意一个被 compromise 都会拖垮两个。separate wallets 几乎没有成本（一次性 deployment cost），安全收益却很真实。

SmartContractWallet.deploy(...) 与概念 15 一样是 illustrative。重要的是 pattern：每个 agent class 一个 wallet，绝不共享。

# Wrong: one wallet shared across agents
shared_wallet = SmartContractWallet.deploy(...)
shopping_agent.wallet = shared_wallet
procurement_agent.wallet = shared_wallet
research_agent.wallet = shared_wallet  # one compromise drains all three

# Right: a separate wallet per agent class
shopping_agent.wallet = SmartContractWallet.deploy(
    spend_limits={"max_per_day_usdc": 100},
)
procurement_agent.wallet = SmartContractWallet.deploy(
    spend_limits={"max_per_day_usdc": 1000, "allowed_recipients": [...]},
)
research_agent.wallet = SmartContractWallet.deploy(
    spend_limits={"max_per_day_usdc": 50, "max_per_call_usdc": 0.50},
)

2. Key rotation，按 schedule，也按需。 以 90 天为 baseline 轮换 signing keys（与 Stripe 对 API keys 的建议一致）。当 agent operator 离开团队、deployment 触碰 signing surface，或有任何异常迹象时，立即轮换。轮换这个习惯比具体天数更重要。

note

下面的 pattern 是真实的。client 名称 azure_key_vault 是 illustrative；请使用你的 provider 的 vault SDK。关键是 key 存在 vault 中，并在使用时读取当前 version。

# Read the current key version from the vault. The version changes when the key rotates.
def get_signing_key(agent_class: str) -> SigningKey:
    return azure_key_vault.get_latest_version(
        secret_name=f"agent-wallet-signing-key-{agent_class}",
    )

# Old transactions, signed with the previous version, stay valid until they expire.
# New transactions use the current version.

3. 能熬过 crash 的 audit logs。 每个 authorization decision 都要记录到与 agent runtime 分离的 durable storage：每个 SPT minted、每个 mandate signed、每个 x402 signature、每个 MPP session opened。如果 agent crash，audit log 必须仍然存在。Neon Postgres 加 Inngest 的 step memoization 可以做到；也可以直接写 object storage（S3 或等价物）来获得最大 durability。

note

这里教学的是 audit-log pattern。client 名称 neon_client 是 illustrative；使用你自己的 database client。规则是在 payment 发生前记录 decision。

# Every payment-authorizing action logs to durable storage BEFORE the action completes.
@function_tool
async def acp_create_cart_and_checkout(
    ctx: RunContextWrapper,
    merchant_id: str,
    items: list["CartItem"],
    max_total_usd: Decimal,
) -> "CheckoutResult":
    audit_id = str(uuid4())

    # Log the authorization decision FIRST, before any payment happens.
    await neon_client.audit_log.insert({
        "audit_id": audit_id,
        "agent_class": ctx.context["agent_class"],
        "user_did": ctx.context["user_session"].did,
        "action": "acp_create_cart_and_checkout",
        "merchant_id": merchant_id,
        "max_total_usd": max_total_usd,
        "timestamp": datetime.utcnow().isoformat(),
        "status": "initiated",
    })

    try:
        result = await _actually_complete_checkout(merchant_id, items, max_total_usd)
        await neon_client.audit_log.update(audit_id, {
            "status": "completed",
            "actual_total_usd": result.total_charged_usd,
            "order_id": result.order_id,
        })
        return result
    except Exception as e:
        await neon_client.audit_log.update(audit_id, {"status": "failed", "error": str(e)})
        raise

4. 跨整笔 transaction 的 distributed traces。 audit log 告诉你什么成功了。traces 告诉你发生了什么，包括失败、重试或卡住的 calls。在 agent commerce 中，完整 trace 常常是 debug 失败 transaction 的唯一办法，因为一个 user request 可能 fan out 到一个 SDK run、5–10 个 tool calls、2 或 3 个 protocol HTTP requests、一个稍后到达的 Stripe webhook，以及一个数小时后恢复的 Inngest function。如果没有一个 trace ID 把这些连起来，post-mortem 不可能完成。下面的 OpenTelemetry 代码使用真实稳定的 OTel API。

from opentelemetry import trace
from opentelemetry.trace import Status, StatusCode

tracer = trace.get_tracer("agent-commerce")

@function_tool
async def acp_create_cart_and_checkout(
    ctx: RunContextWrapper,
    merchant_id: str,
    items: list["CartItem"],
    max_total_usd: Decimal,
) -> "CheckoutResult":
    # The span name is the protocol action; attributes capture what you filter by
    # in your observability tool (Datadog, Honeycomb, Grafana, and so on).
    with tracer.start_as_current_span(
        "acp.checkout",
        attributes={
            "agent.class": ctx.context["agent_class"],
            "user.did": ctx.context["user_session"].did,
            "acp.merchant_id": merchant_id,
            "acp.max_total_usd": float(max_total_usd),
            "acp.item_count": len(items),
        },
    ) as span:
        try:
            result = await _actually_complete_checkout(merchant_id, items, max_total_usd)
            span.set_attribute("acp.order_id", result.order_id)
            span.set_attribute("acp.actual_total_usd", float(result.total_charged_usd))
            span.set_status(Status(StatusCode.OK))
            return result
        except Exception as e:
            span.set_status(Status(StatusCode.ERROR, str(e)))
            span.record_exception(e)
            raise

trace context 会通过正确 instrumentation 在 httpx、openai-agents 和 Inngest 中自动流动。当 20 分钟后 Stripe webhook 为这个 order 的 dispute 到达时，webhook handler 会通过 order metadata 中携带的 trace_id 加入同一条 trace。一个 trace ID 覆盖整笔 transaction 的生命，从第一次 request 到 dispute resolution。

Audit logs 和 traces 回答不同问题，两者都需要。Audit logs 回答业务问题（「这个用户周二花了多少钱？」）。Traces 回答 debugging 问题（「order abc123 的 checkout 为什么失败？」）。audit log 只记录 successful path；它不会捕捉到一个 call 成功前重试 5 次，也不会捕捉 protocol call 在 timeout 前卡了 30 秒。这些 failure shapes 只会出现在 traces 中。因为有 audit logs 就跳过 traces，是误解了两个工具的用途。

最常见的 identity mistake 是把 wallet address 当成 agent identity。address 是公开的：任何人都能向它转账，任何人都能验证某笔 transaction 来自它。private signing key 才是 identity，你要像保护 identity 一样保护它。把 signing keys 放在 environment variables（更糟是 source code）中的团队，等于把 agent identity 交给了任何能访问这些 secrets 的人。keys 放进 vault，绝不放 env vars。

概念 17：四个协议中的 dispute 和 refund mechanics​

一句话： 每个协议处理 disputes 和 refunds 的方式不同，而 use case 需要的 dispute model 常常是决定你组合哪些协议的最强因素。

四个协议分别用自己的方式处理 disputes 和 refunds，而这种差异往往决定一个 use case 的协议选择。需要 chargeback protection 的 use case 不能跑在纯 x402 上。seller 没有 customer-service setup 的 use case 不能使用 ACP。下面是每个协议如何处理 dispute，让你能把协议匹配到实际需要的 dispute model。

ACP：通过 card network 处理 disputes。 因为 merchant 在 ACP 中仍是 merchant of record，每条标准 card-network dispute path 都能工作。buyer 的银行发起 chargeback；Stripe（或其他 processor）处理 merchant defense；merchant 遵守既有 refund policy。这是 ACP 最大的 practical advantage。每个 retail buyer 对 returns 的预期都能自然成立。

对 agent 发起的 ACP refund：

与本课其他 ACP client calls 一样，acp_client.refunds.create(...) 是 illustrative。result model 和 tool wiring 是真实的。

@function_tool
async def acp_refund(
    order_id: str,
    reason: str,
    amount_usd: Decimal | None = None,
) -> "RefundResult":
    """Start a refund through ACP. The merchant's standard refund policy applies."""
    raw = await acp_client.refunds.create(
        order_id=order_id,
        reason=reason,
        amount_usd=amount_usd,  # None means full refund
    )
    return RefundResult(
        refund_id=raw.refund_id,
        order_id=order_id,
        status=raw.status,
        amount_refunded_usd=raw.amount_refunded_usd,
    )

AP2：通过 audit trail 处理 disputes。 AP2 对 dispute 的贡献是 mandate chain（先 Intent，再 Cart，再 Payment）。dispute 出现时，这条 chain 是用户实际授权内容的证据，并且具有法律效力。它不替代 underlying rail 的 dispute path：如果 AP2 mandate 授权了一笔通过 Stripe 的 card payment，Stripe 的 dispute process 仍然适用。AP2 增加的是用户同意内容的 proof。

dispute flow：

1. User claims: "I never authorized this purchase."
2. Merchant retrieves the signed AP2 Cart Mandate from the transaction record.
3. Merchant presents the Cart Mandate (with the user's signature) to the
   payment processor as part of the dispute defense.
4. The card network or processor checks the signature against the user's
   registered public key. If it is valid, the dispute is resolved for the merchant.

x402：没有 formal dispute mechanism。 纯 x402 payments 按设计 non-refundable。payment 在链上 1–2 秒 settlement；没有 chargeback。这是 x402 最大的 practical limit。它适合 $0.001 API call（dispute 成本会高于付款本身），但不适合任何 buyer 合理需要 refund 的场景。

软化 x402 no-refund 属性有三种方式：

  • Escrow。 对更高价值的 x402 payments，使用 smart-contract escrow 持有资金，直到 buyer 发出 acceptance signal。ERC-8004 包含 multi-agent transactions 的 escrow primitives。
  • 与 AP2 和另一种 rail 组合。 如果你需要 x402 的速度，同时也需要 dispute support，AP2 plus x402 composition 会给你 mandate chain 作为 evidence，同时由 x402 保持快速 settlement。settlement 本身仍不可逆；mandate 只是证明双方同意了什么。
  • Seller guarantees。 对 paid APIs，seller terms 通常包含 off-chain 执行的 refund rules（如果服务失败，seller 自愿把 USDC 退回）。这对 reputable sellers 可行，对 anonymous sellers 会崩掉。

MPP：通过 Stripe 处理 disputes。 在 card rails 上 settlement 的 MPP sessions 继承 Stripe 的标准 dispute machinery，与 ACP 相同。通过 Tempo stablecoin 或 Lightning settlement 的 MPP sessions，则走 Stripe 的 seller-side dispute resolution。无论 rail 是什么，Stripe 都会让 merchant 对 outcome 负责。

你需要的 dispute model 常常比 cost 或 latency 更能驱动 composition。consumer-shopping platform 需要 chargebacks，所以 ACP 合适。纯 machine-to-machine API marketplace 完全不需要 disputes，所以 x402 合适。enterprise procurement platform 需要 audit-grade evidence，所以 AP2 加任意 settlement rail 都合适。

概念 18：FastAPI 和 Inngest webhook plumbing：闭合 request/response loop​

一句话： 有些 payment events 会按自己的时间表到达（disputes、mandate signatures、seller-side payment requests），所以你需要薄 FastAPI handler 接住它们，再用 Inngest event 把工作带入 durable workflow。

第 1 到第 5 部分以及概念 15 到 17 都把 agent 当作 buyer：它发送 request，protocol 回复，SDK 针对结果推理。但 agent commerce 是双向的。Stripe 会发送 charge.dispute.created webhooks。AP2 mandate signing 在你的 server 外发生，在用户设备上完成，并稍后 post back。x402 sellers 需要 server-side middleware 返回 402 Payment Required 并检查 X-PAYMENT headers。这些都无法放进单个 Runner.run() call 中。它们需要 FastAPI handlers 作为 HTTP boundary，并用 Inngest events 作为桥接，回到 durable workflows。

这个 concept 会走 3 个生产中需要的 patterns。没有它们，系统就会留下 async events 掉进去的裂缝。

Pattern 1：Stripe webhook 流入 suspended Inngest function。 当用户对 ACP order 提交 chargeback 时，Stripe 会向你的 endpoint 发送 charge.dispute.created。它可能在下单后 5 分钟到达，也可能 60 天后到达。下单 workflow 早已退出，但 agent 仍需要反应：通知 merchant、写 audit，可能还要构建 defense。FastAPI handler 把 webhook 变成 Inngest event，Inngest function 接住它并运行 agent 处理 dispute。

stripe.Webhook.construct_event(...) 和 stripe.error.SignatureVerificationError 是真实 Stripe APIs。Inngest wiring 是真实的。注意 firing API：events 要用 send(events=[inngest.Event(...)]) 发出，不是裸 dict。

from fastapi import FastAPI, Request, HTTPException
from pydantic import BaseModel
from decimal import Decimal
import stripe, inngest

app = FastAPI()
inngest_client = inngest.Inngest(app_id="agent-commerce", is_production=False)

class StripeDisputeEventPayload(BaseModel):
    order_id: str
    dispute_id: str
    amount_usd: Decimal
    reason: str
    raw_event_id: str

@app.post("/webhooks/stripe")
async def stripe_webhook(request: Request):
    # 1. Verify the Stripe signature. This security gate is required.
    signature = request.headers.get("Stripe-Signature")
    payload = await request.body()
    try:
        event = stripe.Webhook.construct_event(
            payload=payload, sig_header=signature, secret=settings.stripe_webhook_secret,
        )
    except stripe.error.SignatureVerificationError:
        raise HTTPException(status_code=400, detail="Invalid signature")

    # 2. Route by event type. This handler does NO business logic.
    #    It only fires Inngest events so the durable workflow does the work.
    if event.type == "charge.dispute.created":
        await inngest_client.send(events=[
            inngest.Event(
                name="stripe/dispute.created",
                data=StripeDisputeEventPayload(
                    order_id=event.data.object.metadata.get("order_id"),
                    dispute_id=event.data.object.id,
                    amount_usd=Decimal(event.data.object.amount) / 100,
                    reason=event.data.object.reason,
                    raw_event_id=event.id,
                ).model_dump(),
                id=event.id,            # idempotency seed
            ),
        ])

    # 3. ACK Stripe right away. The real work runs in Inngest, durably.
    return {"received": True, "event_id": event.id}

# The Inngest function that handles the dispute: fully durable and retryable.
@inngest_client.create_function(
    fn_id="handle-stripe-dispute",
    trigger=inngest.TriggerEvent(event="stripe/dispute.created"),
    # Idempotency by raw_event_id makes sure Stripe retries do not process twice.
    idempotency="event.data.raw_event_id",
)
async def handle_stripe_dispute(ctx: inngest.Context) -> dict:
    payload = StripeDisputeEventPayload(**ctx.event.data)

    # Log to audit immediately.
    await ctx.step.run("audit-dispute-received", log_dispute_to_neon, payload)

    # Run an agent to assemble the dispute defense.
    defense_agent = Agent(
        name="DisputeDefenseAgent",
        instructions="Assemble dispute defense materials: order receipt, AP2 mandate if any, "
                     "delivery confirmation, customer communication history.",
        tools=[fetch_order_details, fetch_mandate_chain, fetch_delivery_proof, submit_dispute_response],
    )
    defense = await ctx.step.run(
        "build-and-submit-defense",
        Runner.run, defense_agent, f"Build defense for dispute {payload.dispute_id}",
    )
    return {"status": "completed", "output": {"defense_submitted": defense.final_output.model_dump()}}

FastAPI handler 保持 thin（verify，然后 fire event）。Inngest function 是 durable 的（idempotency、retries、step memoization）。agent reasoning 发生在 Inngest function 中，绝不发生在 webhook handler 中。这个拆分很重要，因为 Stripe 期望约 5 秒内收到 2xx response，而一次 agent run 可能需要 30 秒或更久。

Pattern 2：AP2 mandate-signing callback 恢复 suspended step.wait_for_event。 概念 9 展示了 AP2 signing tool 请求 signature 并等待。在生产中，signing 发生在你的 server 外：用户打开手机 app，看到 mandate，点 Approve，然后 signed mandate post back。agent 的 Inngest workflow 正 suspended 在 step.wait_for_event；FastAPI callback 触发唤醒它的 event。

note

signing callback 是你自己的 FastAPI route。Inngest correlation 使用 if_exp=（一个 CEL expression），等待的 payload 位于 async. prefix 下。wait_for_event timeout 时返回 None。ap2_verify_signature 和 persistence client 是 illustrative；workflow shape 是真实的。

from datetime import timedelta
from pydantic import BaseModel

class MandateSignedPayload(BaseModel):
    mandate_id: str
    user_did: str
    signature: str  # the user's signature over the mandate hash
    signed_at: str

@app.post("/callbacks/ap2/mandate-signed")
async def mandate_signed_callback(payload: MandateSignedPayload, request: Request):
    # 1. Verify the signature against this user's registered public key.
    is_valid = await ap2_verify_signature(
        mandate_id=payload.mandate_id,
        user_did=payload.user_did,
        signature=payload.signature,
    )
    if not is_valid:
        raise HTTPException(status_code=400, detail="Invalid mandate signature")

    # 2. Persist the signed mandate. Mandates have a 7-year retention requirement.
    await neon_client.mandates.insert({
        "mandate_id": payload.mandate_id,
        "user_did": payload.user_did,
        "signature": payload.signature,
        "signed_at": payload.signed_at,
        "status": "signed",
    })

    # 3. Fire the Inngest event that resumes the agent's workflow.
    await inngest_client.send(events=[
        inngest.Event(name="ap2/mandate.signed", data=payload.model_dump(), id=payload.mandate_id),
    ])
    return {"received": True, "event_id": payload.mandate_id}

# The Inngest function that was waiting. if_exp correlates the wait to this mandate.
@inngest_client.create_function(
    fn_id="agent-procurement-workflow",
    trigger=inngest.TriggerEvent(event="procurement/task.created"),
)
async def procurement_workflow(ctx: inngest.Context) -> dict:
    # ... agent creates the Intent Mandate, fires "ap2/mandate.signing.requested" ...

    # Suspend until the user signs. Zero compute is used during the wait.
    signed = await ctx.step.wait_for_event(
        "wait-for-intent-mandate-signature",
        event="ap2/mandate.signed",
        if_exp=f"async.data.mandate_id == '{ctx.event.data['mandate_id']}'",
        timeout=timedelta(hours=24),   # users can take real time
    )

    if signed is None:                 # timeout returns None
        return {"status": "abandoned", "reason": "user did not sign within 24h"}

    # Resume with the signed mandate. The agent continues from exactly where it left off.
    cont = await ctx.step.run("continue-procurement", continue_with_signed_mandate, signed)
    return {"status": "completed", "output": {"procurement_continuation": cont}}

Inngest 的 step.wait_for_event 加 if_exp 正是为此构建的。FastAPI handler 是从 HTTP 到 Inngest event bus 的单向桥。几个小时前 suspend 的 workflow 会带 signed payload 恢复，agent 从停下的地方继续。

Pattern 3：x402 seller-side middleware（当你暴露 paid API，而不只是消费 paid API）。 在 multi-agent marketplace（Decision 4）中，你的 agent 有时是 buyer，有时是 seller，让其他 agents 为你的 research、analysis 或 code 付费。seller side 需要 FastAPI middleware：返回 402 Payment Required、检查 X-PAYMENT headers，并且只有在 facilitator 验证 payment 后才服务资源。

note

下面的 seller-side X402Middleware 是 illustrative；没有 x402_server PyPI package。真实 server-side x402 通过 x402 package 的 server 和 facilitator helpers 加 framework integration 实现。buyer-and-seller 对称这个想法是真实的，FastAPI route 也是真实的。

from fastapi import FastAPI, Request, Response
from decimal import Decimal
# Illustrative seller-side imports (see note above).
from x402_server import X402Middleware, PaymentRequirement

# Configure the middleware once at app startup.
app = FastAPI()
app.add_middleware(
    X402Middleware,
    payment_requirements_by_route={
        "/api/research": PaymentRequirement(
            scheme="exact",
            network="eip155:8453",  # Base
            asset=USDC_BASE_CONTRACT,
            recipient=settings.merchant_wallet_address,
            max_amount_usdc=Decimal("0.50"),
            expiry_seconds=300,
        ),
        "/api/code-review": PaymentRequirement(
            scheme="exact",
            network="eip155:8453",
            asset=USDC_BASE_CONTRACT,
            recipient=settings.merchant_wallet_address,
            max_amount_usdc=Decimal("2.00"),
            expiry_seconds=300,
        ),
    },
    facilitator_url="https://facilitator.cloudflare.com/x402",
)

# Your business logic. The middleware enforces payment before this runs.
@app.post("/api/research")
async def research_endpoint(request: Request):
    # By the time we reach here, the X-PAYMENT header has been verified and settled.
    # request.state.x402_proof carries the on-chain transaction hash for audit.
    query = (await request.json())["query"]

    research_agent = Agent(
        name="ResearchAgent",
        instructions="Conduct deep research on the query and return a structured report.",
        tools=[search_web, fetch_papers, summarize],
    )
    result = await Runner.run(research_agent, query)

    return Response(
        content=result.final_output,
        headers={"X-PAYMENT-PROOF": request.state.x402_proof.tx_hash},
    )

x402 是对称的。概念 10 中的 buyer-side code 和这里的 seller-side middleware，是同一个协议的两半。multi-agent marketplace 两者都运行：它的 agents 从外部 services 购买，也向外部 agents 出售。

生产中反复出现三类失败：

  1. Webhook handlers inline 执行业务逻辑。 在 webhook response 中运行 agent 的 FastAPI handler 会超过 Stripe 的 5 秒 timeout。Stripe retry，agent 运行两次，然后你向用户收费两次。handler 保持 thin；Inngest function 是 durable 的。
  2. 忘记 webhook idempotency。 Stripe 会用同一个 event.id retry failed deliveries。Inngest function 上没有 idempotency key，每次 retry 都会创建 duplicate。使用 "event.data.raw_event_id" 作为 idempotency key。
  3. callbacks 不检查 signature。 AP2 mandate-signed callbacks 必须对 registered public key 验证用户 signature。否则任何 caller 都能伪造 mandate-signed events。检查失败要返回 400，而不是只记录 warning。

第 7 部分：结语：本课真正教的是什么​

概念 19：layered composition 的纪律​

一句话： 本课的一切都归结为一个工作：读懂 use case，把它拆成四层，并在每层选择正确协议。

本课有 19 个 Concepts 和 5 个 Decisions。它们都是为一个 claim 搭建 scaffolding：2026 年的 agent commerce 不是单一协议，而是 layered architecture；你的工作是为眼前的 use case 在每层选择正确协议。其他一切都由此展开。

同一个形状会出现在三个尺度上。

在 protocol scale，四个 headline protocols 在不同层解决不同问题：ACP 在 Layer 3，AP2 在 Layer 2，x402 和 MPP 在 Layer 4。把它们当成同一层 rivals，是最常见的 architecture mistake。它们只在层重叠处竞争（settlement 上 x402 against MPP；某些 flows 的 authorization 上 AP2 against ACP's SPT）。其他地方都可以 compose。

在 system scale，production system 每层有一个 protocol，并通过 OpenAI Agents SDK 这个 universal client 接起来。SDK 的 @function_tool、RunContextWrapper 和 tool_input_guardrail 清楚映射到每层的 concerns。SDK 不是 protocol。它是让你干净组合 protocols 的 orchestrator。

在 discipline scale，你的工作是读一个 use case，把它拆成四层，为每层选择正确协议，并按 use case 的真实 constraints 证明每个选择：transaction value、latency budget、dispute model、audit requirements。工作不是「选一个喜欢的协议」。工作是「对这个 use case，每层要求什么？」

第 5 部分的五个 Decisions 已经把真实 use cases 放进这个纪律中走了一遍。Decision 1（consumer shopping）落在 ACP 加 Stripe card rails，因为 use case 需要 chargeback protection。Decision 2（API-paying agent）落在 x402 only，因为 machine-to-machine flows 中 Layers 2 和 4 会合并。Decision 3（enterprise procurement）走到最复杂的 composition（AP2 plus ACP plus MPP），因为 audit 和 recurrence needs 迫使它如此。Decision 4（multi-agent marketplace）落在 AP2 plus ERC-8004 plus x402，因为 bilateral-trust need 没有其他方式满足。Decision 5 用完全不同 stack（Google ADK 加 Coinbase wallet 加 AP2 加 x402）重建 Decision 2，并得到同一个四层形状，证明 architecture 不是 Stripe 和 OpenAI 外面的一层 wrapper。

你该选择的 composition 由 use case 决定，不由品味决定。一个团队因为「stablecoins are the future」而选择 x402，却在构建 consumer shopping experience，就选错了 composition。一个团队因为「Stripe is enterprise-grade」而选择 ACP，却在为每次 $0.001 的 API access 付款，也选错了 composition。让 use case 驱动选择。

Cheat sheet：一页 framework​

打印出来。贴在墙上。review 任何 agent-commerce design 时使用。

四层（记住这个 stack）​

Layer 1: DISCOVERY      ->  "What's available to buy?"
Layer 2: AUTHORIZATION  ->  "Am I allowed to spend this?"
Layer 3: COMMERCE       ->  "What's the full purchase lifecycle?"
Layer 4: SETTLEMENT     ->  "Where does the money actually move?"

每层协议（2026 年 top picks）​

LayerTop picksPick by
DiscoveryMCP、A2A、agent directories、AI shopping surfacesagent 需要的 services 在哪里
AuthorizationAP2 mandates、ACP SPT、TAP、ERC-8004trust model：audit-rigorous、Stripe-native、identity-only 或 multi-agent
CommerceACP、UCP、direct API（none）use case 是否需要 commerce lifecycle
Settlementx402、MPP、card rails、bank/Lightning经济性：transaction value 驱动 rail

四种 canonical compositions​

Use caseStack
Consumer shoppingAI surface + ACP SPT + ACP + Stripe cards
API-paying agentMCP/directory + EIP-3009 + (none) + x402
Enterprise procurementA2A/MCP + AP2 + ACP/UCP + MPP/cards
Multi-agent marketplaceA2A + AP2 + ERC-8004 + (none) + x402

三层 spend-limit enforcement（必需）​

Level 1: Wallet / payment-method limits  (smart-contract caps OR Stripe customer caps)
Level 2: SDK tool guardrails             (tool_input_guardrail on each payment tool)
Level 3: Application business logic      (per-user, per-category, per-merchant policies)

跳过任何一层，离 total drain 只差一个 bug。最常见错误是把 agent-level output_guardrail 用作 Level 2。它在 agent 最终回复上触发，太晚，无法阻止 payment。请改用 tool_input_guardrail。

经济阈值（记住）​

Transaction value
├── < $5         → x402 or MPP stablecoin (card fees exceed the transaction)
├── $5 - $1,000  → ACP + card rails (chargeback protection worth the 2.9%)
└── > $1,000     → AP2 + composed stack (audit + dispute defense + multi-rail)

dispute model（往往是最强约束）​

Use case needs chargeback protection?
├── Yes → ACP + card rails (or MPP card mode)
└── No  → x402 acceptable (faster, cheaper, no refunds)

Use case needs audit evidence?
├── Yes → AP2 at Layer 2 (mandates as legally-admissible evidence)
└── No  → SPT or EIP-3009 sufficient

生产上线前 checklist​

  • Wallet/payment-method spend limits configured at Level 1
  • SDK tool_input_guardrail attached to every payment-authorizing tool (Level 2)
  • Application enforces per-user, per-category, and per-merchant caps (Level 3)
  • Signing keys in a key vault (Azure Key Vault or equivalent), never in env vars
  • Per-agent wallet separation (no shared keys across agent classes)
  • Key rotation schedule defined (90-day baseline)
  • Audit logs to durable storage independent of the agent runtime
  • OpenTelemetry traces span SDK, httpx, Inngest, and FastAPI; one trace ID per transaction
  • Pydantic models at every boundary (tool returns, protocol payloads, FastAPI bodies, Inngest events)
  • Decimal for money everywhere (never float)
  • Stripe webhook handler verifies the signature and fires an Inngest event (thin handler, durable function)
  • AP2 mandate-signed callback verifies the signature and resumes step.wait_for_event
  • If exposing paid APIs: x402 seller-side middleware configured per route
  • Inngest idempotency key set on every webhook-triggered function
  • Dispute and refund mechanism understood and documented per protocol
  • Human-in-the-loop confirmation gate for the first 30 days of production
  • Cart-accuracy metric measured before relaxing the confirmation gate

Quick reference：压缩决策树​

有新 use case 时，按这棵树走。

1. What's the agent buying?
   ├── Retail goods for a user → Decision 1 pattern (ACP + cards)
   ├── API access for itself → Decision 2 pattern (x402-only)
   ├── Supplier goods/services for an org → Decision 3 pattern (AP2 + composed)
   └── Work from another agent → Decision 4 pattern (AP2 + ERC-8004 + x402)

2. What's the transaction value?
   ├── < $5 → settlement = x402 or MPP stablecoin
   ├── $5-$1,000 → settlement = card rails via ACP/MPP
   └── > $1,000 → settlement = MPP sessions OR bank rails

3. What's the dispute model?
   ├── Need chargeback protection → must include card rails somewhere
   ├── Need audit evidence → must include AP2 at Layer 2
   └── Neither → x402 / direct rail sufficient

4. What's the latency budget?
   ├── < 1 sec → only stablecoin rails work
   ├── 1-5 sec → x402, MPP, or pre-signed AP2 mandates
   └── > 5 sec → full ACP checkout works

5. Compose: pick one protocol from each layer, justified against the constraints above.

Design-review template：review 任何 agent-commerce architecture 时要问的问题​

review 同事设计（或隔一天 review 自己的 draft）时，按顺序走这些问题。每个问题都映射到本课某一节；如果答案不满足问题，就回到对应章节。

Architecture questions​

  1. 这个 use case 服务什么？ 如果答案是「多个」，primary one 是什么？试图服务全部四种 canonical use cases 的设计，是概念 12 中的 anti-pattern；标记它。
  2. 明确走四层。 Discovery？Authorization？Commerce？Settlement？每层哪个 protocol？把 four-protocol composition 明确说出来。
  3. 按 use case 证明每层选择。 为什么这一层是这个协议？换成另一个会改变什么？概念 13 的 across-vs-within test 适用于这里。
  4. 某层是否有 protocol overlap？ Layer 4 有两个协议竞争？Layer 2 有两个？如果有，这就是需要 use case 证明的 complexity。

Economic questions​

  1. transaction value distribution 是什么？ sub-dollar？$10–$100？$1,000-plus？概念 14 的 economic threshold 应该驱动 settlement choice。
  2. latency budget 是什么？ sub-second？5–30 秒？latency budget 应该约束 settlement 和 mandate choices。
  3. expected volume 下每笔成本是多少？ 加上 protocol fees、processor fees 和 per-transaction infrastructure cost。用它检查 composition 是否经济可行。

Operational questions​

  1. 三个层级都有 spend-limit enforcement 吗？ Wallet/payment-method、SDK、application business logic。概念 15：跳过任何一层都会暴露风险。
  2. Identity hygiene？ per-agent wallet separation、key rotation、audit logs 到 durable storage、每笔 transaction 一个 trace ID 的 distributed traces。概念 16 的四个习惯。
  3. Dispute and refund mechanics？ composition 是否处理 use case 实际会遇到的 disputes？概念 17：这常常是最强约束。
  4. Operational envelope？ Inngest（或等价 durable execution）在哪里处理 long-running flows、retries 和 idempotency？这在 Production Worker 速成课 中覆盖。
  5. Human-in-the-loop gates？ flow 中用户或 operator 在哪里确认？生产前 30 天默认使用 confirmation gate。

Webhook and async-callback questions（概念 18）​

  1. Stripe webhook handler 是否存在且 thin？ 验证 signature，发 Inngest event，5 秒内 ACK。business logic 位于 Inngest function，而不是 handler。
  2. AP2 mandate-signing callback 是否存在并验证 user signatures？ 没有它，agent 的 step.wait_for_event 永远不会恢复。没有 signature verification，任何人都能伪造 mandate-signed events。
  3. 如果暴露 paid APIs：是否按 route 配置 x402 seller-side middleware？ Multi-agent marketplaces 同时运行 buyer-side 和 seller-side code；seller side 需要 middleware。
  4. 每个 webhook-triggered function 是否有 Inngest idempotency keys？ Stripe 用同一个 event.id retry；没有 idempotency 会重复 charge 或 refund。
  5. Contract layer：每个边界都有 Pydantic models 吗？ Tool returns、protocol payloads、FastAPI request and response bodies、Inngest event payloads 全部 typed。money 用 Decimal，绝不使用 float。

Failure-mode questions​

  1. 生产中最可能的 failure mode 是什么？ 每个 canonical decision 都有一个：cart mismatch（Decision 1）、runaway spend（Decision 2）、Intent Mandate scope mismatch（Decision 3）、gamed reputation（Decision 4）。你的是什么？
  2. mitigation 是什么？ 第 5 部分每个 failure mode 都有 explicit mitigation。设计是否包含它，还是只依赖「protocol 会处理」？

Track-readiness questions（用于 production deployment）​

  1. 团队处于哪条 learning track？ Reader、Beginner、Intermediate、Advanced。要诚实。生产部署需要 Advanced-track depth，不是 Reader-track familiarity。
  2. 如果 agent 行为异常，rollback plan 是什么？ wallet kill switch？SPT revocation？在 SDK level disable agent？上线前就定义好，不要等 incident 之后再想。

References​

四个协议的 primary sources。优先看它们，不要先看 secondary commentary；specs 演化速度比关于它们的文章更快。

ACP（Agentic Commerce Protocol）​

  • Specification repository：github.com/agentic-commerce-protocol/agentic-commerce-protocol（Apache 2.0）
  • Developer site：agenticcommerce.dev
  • Co-maintainers：OpenAI 和 Stripe
  • Stripe integration docs：stripe.com/docs/agentic-commerce
  • Latest spec version cited：2026-04-17

AP2（Agent Payments Protocol）​

  • Specification repository：github.com/google-agentic-commerce/AP2（Apache 2.0）
  • Documentation site：ap2-protocol.org
  • Coalition members：Google 加 60-plus partners（Salesforce、ServiceNow、Adobe、Shopee、Etsy、Adyen、American Express、JCB、UnionPay International、PayPal、Mastercard、Coinbase 等）
  • a2a-x402 extension repository：github.com/google-a2a/a2a-x402
  • Latest version cited：v0.2.0（2026 年 4 月）

x402​

  • Specification repository：github.com/coinbase/x402（Apache 2.0）
  • Documentation site：x402.gitbook.io（也包括 x402.org）
  • 由 Coinbase 创建，现在由 Linux Foundation 的 x402 Foundation steward（2026 年 4 月），成员包括 Cloudflare、Stripe、AWS、Google 等。
  • Cloudflare OpenAI Agents SDK integration：developers.cloudflare.com/agents/x402
  • Adoption directory：agent.market
  • x402 Foundation members include：Adyen、AWS、American Express、Base、Circle、Cloudflare、Coinbase、Google、Mastercard、Microsoft、Polygon Labs、Shopify、Solana Foundation、Stripe、Visa

MPP（Machine Payments Protocol）​

  • Specification site：mpp.dev
  • Co-developers：Stripe 和 Tempo（Stripe 的 L1 blockchain，与 Paradigm 共同 incubated）
  • Stripe announcement：stripe.com/blog/machine-payments-protocol
  • Launch date：2026 年 3 月 18 日（mainnet）
  • Partners at launch：Stripe、Visa、Lightspark、Anthropic、OpenAI、Shopify，以及 100-plus services

Adjacent protocols​

  • A2A（Agent2Agent，Google）：github.com/google-a2a/A2A，AP2 扩展的协议
  • MCP（Model Context Protocol，Anthropic）：modelcontextprotocol.io，agent 用来发现 services 的 tool 和 context layer
  • UCP（Universal Commerce Protocol，Google）：作为 Google shopping surfaces 上 ACP 的 peer 发布
  • TAP（Trusted Agent Protocol，Visa 和 Cloudflare）：2025 年 10 月 14 日发布的 identity verification protocol
  • ERC-8004：用于 multi-agent transactions 的 on-chain trust standard

OpenAI Agents SDK​

  • Python SDK：pip install openai-agents（latest release 2026 年 5 月 19 日）
  • Documentation：openai.github.io/openai-agents-python
  • JavaScript/TypeScript SDK：github.com/openai/openai-agents-js

Operational and contract-layer tools（在概念 16 和 18 中大量使用）​

  • Inngest：inngest.com，本课中运行 agent workflows 的 durable execution platform
  • FastAPI：fastapi.tiangolo.com，webhook endpoints 和 seller-side x402 middleware 所在的 async HTTP layer
  • Pydantic：pydantic.dev，每个边界上的 type contract（tool returns、protocol payloads、FastAPI bodies、Inngest events）
  • OpenTelemetry：opentelemetry.io，为 httpx、openai-agents、FastAPI 和 Inngest 做 auto-instrumentation 的 distributed tracing；一个 trace ID 跨整笔 transaction
  • httpx：python-httpx.org，x402-client、ACP client 和 AP2 a2a-x402 extension 底层的 async HTTP client
  • Azure Key Vault：生产中存放 signing keys 的地方，按概念 16 的纪律轮换
  • Stripe webhooks documentation：stripe.com/docs/webhooks，signature verification、event types 和 retry semantics

现在你已经拥有完整 framework：四层、四个协议、组合规则、五个 worked decisions、决定系统能否存活的 production concerns，以及一页 reference。协议会继续变化。四层纪律不会。按层构建，即使协议名称改变，你仍然会是对的。