这篇能做出什么
跑完这篇教程，你会得到一个能在 AP2 沙箱里端到端走通的购物智能体：给它一句自然语言需求（例如"帮我买一副降噪耳机，5 天内能到，到手价不超过预算"），它会自动完成下面这条链路——
1. 向多个商家源并发检索商品，把不同格式的报价归一化成同一种数据结构；
2. 用确定性代码比价（含运费、时效、库存），产出一份可解释的决策记录；
3. 把用户的预算与约束写成一张 Intent Mandate（意图授权），由用户私钥预签名；
4. 选定商品后生成 Cart Mandate（购物车授权），在人工确认后签名；
5. 向 Credentials Provider 换取 Payment Mandate（支付授权），拿到支付凭证；
6. 走沙箱支付通道，回收一张可验签的交易凭证，写进审计日志。
沙箱环境不产生真实扣款，所有支付凭证都是测试凭证。整条链路的验收清单：
- [ ] 一句需求能检索到 ≥2 个商家的报价
- [ ] 比价结果有明确排序理由，且人工可以复算
- [ ] Intent Mandate 能通过本地验签
- [ ] 超出意图约束时，流程会停下来等人确认，而不是硬闯
- [ ] 能拿到交易凭证，且验签通过
- [ ] 全流程有一份可回放的 JSONL 审计日志
前置条件清单
| 类别 | 需要什么 | 说明 |
|---|---|---|
| 运行时 | Python 3、Docker 与 Docker Compose | 具体版本以官方文档当前版本为准 |
| 工具 | git、curl、jq | jq 用来读沙箱返回的 JSON，排查效率高很多 |
| 凭据 | AP2 沙箱的测试账号与测试凭证 | 由沙箱签发，不要填真实卡号 |
| 网络 | 能访问沙箱入口与示例商家服务 | 公司网络注意放行 |
| 知识 | 看得懂 JSON、改得动 Python 脚本、知道非对称签名是什么 | 不需要密码学背景 |
建议的目录结构：
```text
ap2-shopping-agent/
├─ .env # 沙箱地址、密钥路径、商家源
├─ search.py # 检索与归一化
├─ decide.py # 比价与决策
├─ mandates.py # 三张 Mandate 的构造与验签
├─ checkout.py # 支付与凭证回收
├─ audit.jsonl # 审计日志
└─ keys/ # 本地测试密钥，务必写进 .gitignore
```
先认角色：谁在跟谁说话
AP2 的场景里有这么几个角色，理清楚它们，后面写代码就不会乱：
- User（用户）：出钱的人，也是唯一能给授权的人。
- Shopping Agent（购物智能体）：跑在你的机器上，代表用户检索、比价、发起下单。
- Merchant Agent（商家智能体）：商家侧的对话入口，负责报价、生成购物车。
- Credentials Provider（CP，凭证提供方）：负责签发支付凭证，类似发卡行或钱包的角色。
- Payment Processor（支付处理方）：真正执行扣款并把交易结果回传。
三张 Mandate 是这套协议的核心，本质上是"授权书"：
- Intent Mandate（意图授权）：用户预先签好的约束条件——买什么品类、找哪些商家、预算上限、收货时效、有效期。它解决的是"人不在场时，智能体凭什么可以自己动"。
- Cart Mandate（购物车授权）：针对一次具体下单的确认。人不在场时，它需要引用 Intent Mandate 并落在其约束范围内；人在场时，可以直接由用户签。
- Payment Mandate（支付授权）：把购物车和支付方式绑定起来，交给 CP 去换支付凭证。它里面会引用购物车的内容哈希，防止被调包。
顺带说明一个常见混淆：AP2 常和 A2A、MCP 一起出现。A2A 管"智能体之间怎么通信"，MCP 管"智能体能调用哪些工具和数据"，AP2 管"凭什么可以付这笔钱"。三者分工不同，不要互相替代。
步骤 1：把沙箱跑起来
从官方仓库获取示例代码并启动。仓库地址、镜像 tag、内部端口一律以官方文档与仓库 README 当前内容为准，下面用占位符表示。
```bash
git clone <AP2 官方 samples 仓库地址>
cd <仓库目录>
cp .env.example .env
```
打开 .env，按 README 的说明填入沙箱入口地址与测试凭证，然后启动：
```bash
docker compose up -d
docker compose ps # 确认各服务状态是 running
```
健康检查：
```bash
curl -sS "$AP2_SANDBOX_BASE_URL/health" | jq .
```
返回里出现成功标识，说明沙箱已就绪。把环境变量导出到当前 shell：
```bash
set -a; source .env; set +a
```
Python 侧的依赖：
```bash
python -m venv .venv
source .venv/bin/activate # Windows: .venv\Scripts\activate
pip install httpx pydantic cryptography fastapi uvicorn
```
步骤 2：检索层——把各家的报价归一化
不同商家返回的字段名、金额格式、运费规则都不一样。先定义统一的报价对象，把脏活都关在这一层。
```python
search.py
import asyncio, os
import httpx
from dataclasses import dataclass
from decimal import Decimal
@dataclass(frozen=True)
class Offer:
merchant_id: str
sku: str
title: str
unit_price: Decimal
currency: str
shipping: Decimal
eta_days: int
stock: int
url: str
@property
def landed_cost(self) -> Decimal:
"""到手价 = 单价 + 运费。后续所有比较都用这个字段。"""
return self.unit_price + self.shipping
SOURCES = [
{"id": "merchant-a", "base_url": os.environ["MERCHANT_A_URL"]},
{"id": "merchant-b", "base_url": os.environ["MERCHANT_B_URL"]},
]
def parse_offer(merchant_id: str, raw: dict) -> Offer:
return Offer(
merchant_id=merchant_id,
sku=raw["sku"],
title=raw["title"],
unit_price=Decimal(str(raw["price"]["amount"])),
currency=raw["price"]["currency"],
shipping=Decimal(str(raw.get("shipping", {}).get("amount", "0"))),
eta_days=int(raw.get("eta_days", 99)),
stock=int(raw.get("stock", 0)),
url=raw["url"],
)
async def search_one(client, source, query):
try:
resp = await client.post(
f"{source['base_url']}/search",
json={"query": query, "limit": 20},
timeout=8.0,
)
resp.raise_for_status()
return [parse_offer(source["id"], item) for item in resp.json()["items"]]
except (httpx.HTTPError, KeyError, ValueError) as exc:
print(f"[warn] {source['id']} 检索失败：{exc}")
return []
async def search_all(query: str) -> list[Offer]:
async with httpx.AsyncClient() as client:
groups = await asyncio.gather(*(search_one(client, s, query) for s in SOURCES))
return [offer for group in groups for offer in group]
```
两个关键点：金额一律用 Decimal，绝不用 float；单个商家检索失败只告警不中断，否则一家挂了整条链路就瘫了。
步骤 3：决策层——让代码算钱，让模型说话
比价这种有唯一正确答案的事，交给确定性代码；大模型只负责把结论翻译成人话，以及处理"用户需求模糊"这类问题。顺序反过来会埋雷：模型算错一次价格，后面所有授权都是错的。
```python
decide.py
from dataclasses import dataclass
from decimal import Decimal
from search import Offer
@dataclass
class Decision:
chosen: Offer
rejected: list[tuple[Offer, str]]
reason: str
def rank(offers: list[Offer], constraints: dict) -> Decision:
max_cost = Decimal(str(constraints["max_landed_cost"]))
currency = constraints["currency"]
max_eta = constraints["max_eta_days"]
ok, rejected = [], []
for o in offers:
if o.currency != currency:
rejected.append((o, f"币种不符：{o.currency}"))
elif o.stock <= 0:
rejected.append((o, "无库存"))
elif o.landed_cost > max_cost:
rejected.append((o, f"到手价 {o.landed_cost} 超预算 {max_cost}"))
else:
ok.append(o)
if not ok:
raise RuntimeError("没有满足约束的报价，转人工处理")
排序键：先看时效是否达标，再看到手价，最后看时效快慢
ok.sort(key=lambda o: (o.eta_days > max_eta, o.landed_cost, o.eta_days))
best = ok[0]
return Decision(
chosen=best,
rejected=rejected,
reason=f"在预算内且时效达标，选择 {best.merchant_id}，到手价 {best.landed_cost} {best.currency}",
)
```
把每次决策写进审计日志，格式用 JSONL，一行一条，方便回放：
```python
audit.py
import json, time
def append_audit(path: str, **event):
event["ts"] = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
with open(path, "a", encoding="utf-8") as f:
f.write(json.dumps(event, ensure_ascii=False, default=str) + "\n")
```
调用时记录候选数量、被拒原因、最终选择。三个月后你回看这些日志，才知道当时那个"看起来奇怪"的下单是怎么发生的。
步骤 4：Intent Mandate——把预算变成可签名的授权
Intent Mandate 是用户提前签好的"作业边界"。下面是一份结构示意，实际字段名与 schema 以官方文档为准。
```json
{
"type": "intent_mandate",
"mandate_id": "<uuid>",
"issuer": "user:<用户标识>",
"shopping_agent": "agent:<智能体标识>",
"constraints": {
"categories": ["headphones"],
"merchants_allowed": ["merchant-a", "merchant-b"],
"currency": "USD",
"max_landed_cost": "199.00",
"max_eta_days": 5,
"max_transactions": 1,
"expires_at": "2030-01-01T00:00:00Z"
},
"signature": { "alg": "ES256", "kid": "<key-id>", "value": "<base64url>" }
}
```
拿到沙箱返回的 Mandate 后，先在本地验签，确认它确实由用户私钥签发、且内容没被改动：
```python
mandates.py
from cryptography.exceptions import InvalidSignature
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.asymmetric import ec
def verify_es256(public_key, signing_input: bytes, signature: bytes) -> bool:
"""signing_input 是被签名载荷的规范化字节串，签名算法与沙箱配置保持一致。"""
try:
public_key.verify(signature, signing_input, ec.ECDSA(hashes.SHA256()))
return True
except InvalidSignature:
return False
```
还要做一道业务校验：当前时间是否在 expires_at 之前、品类是否匹配、金额是否超限。协议验签只证明"这封授权书没被改"，不证明"这次下单符合授权书"。这两件事必须分开做。
步骤 5：Cart Mandate 与那个人工确认点
选定商品后，构造 Cart Mandate，内容至少包含：商家标识、商品与数量、到手价、币种、预计时效、以及引用的 Intent Mandate ID。
这是整条链路里第一个必须留人工确认的环节。 确认界面应该把信息摊开，而不是只问一句"确认吗"：
```text
即将下单：
商家 merchant-a
商品 Noise-Cancelling Headphones X1 × 1
到手价 178.00 USD（单价 168.00 + 运费 10.00）
预计送达 3 个工作日
预算上限 199.00 USD（剩余 21.00）
授权来源 intent_mandate <id>（剩余可用次数 1）
[y] 确认并签名 [n] 取消 [e] 换一个候选
```
如果候选商品超出了 Intent Mandate 的约束（超预算、商家不允许、时效不达标），流程必须停下来重新征求用户授权，而不是自动放宽条件。做法是让 rank() 直接抛异常，由外层提示用户重新签发 Intent Mandate。
步骤 6：Payment Mandate 与交易凭证签发
购物车被用户签名后，把 Cart Mandate 交给 Credentials Provider 换取支付凭证。请求体的字段名以官方文档为准，形状大致是：
```bash
curl -sS -X POST "$AP2_SANDBOX_BASE_URL/credentials/issue" \
-H "Content-Type: application/json" \
-H "Authorization: Bearer $AP2_SANDBOX_TOKEN" \
-d @payment_request.json | jq .
```
payment_request.json 里要带上购物车内容哈希和 Cart Mandate 的引用。两个硬性检查：
1. 哈希一致：Payment Mandate 里引用的购物车哈希，必须和实际提交给商家的购物车逐字节对得上。对不上就中止——这说明中间有人改过商品或金额。
2. 幂等：用 Cart Mandate ID 作为幂等键（例如 hashlib.sha256(cart_mandate_id.encode()).hexdigest()）。网络超时重试时，处理器靠它识别出"这是同一笔"，避免重复下单。
支付完成后回收交易凭证，用它自带的公钥或 CP 提供的公钥验签，验签通过再把结果写进审计日志：
```python
append_audit("audit.jsonl",
event="payment_settled",
cart_mandate_id=cart_id,
payment_mandate_id=pay_id,
credential_id=cred_id,
verified=verify_es256(cp_public_key, signing_input, signature))
```
必须留人工确认的环节
| 环节 | 为什么必须留人 |
|---|---|
| Intent Mandate 首次签发与额度变更 | 这是全部自动化的授权源头，扩大额度等于扩大风险面 |
| Cart Mandate 最终确认 | 金额、商家、时效三要素具象化之后的确认，是唯一能拦住"理解偏差"的关卡 |
| 首次合作的新商家 | 没有历史记录，退货与售后成本未知 |
| 超出意图约束的任何情况 | 超预算、超时效、跨币种、跨境税费变化 |
| 订阅、自动续费、周期性扣款 | 一次确认可能变成长期扣款 |
| 取消、退货、退款发起 | 影响资金与账户状态，且往往不可逆 |
| 提升单笔或累计限额 | 风险敞口放大，需要重新评估 |
反过来，检索、比价、排序、生成候选这些只读且可回滚的动作，可以放心交给智能体自己跑。
常见坑与排错
金额用浮点数。0.1 + 0.2 那套问题在钱上会变成真事故。统一 Decimal，或者干脆用最小货币单位的整数（分）。
只验签不验约束。签名有效只说明"授权书没被改"，不代表"这次下单在授权范围内"。两道校验缺一不可。
忘记校验跨 Mandate 绑定。Cart Mandate 要引用 Intent Mandate，Payment Mandate 要引用 Cart Mandate 的内容哈希。少了这层绑定，攻击者可以把一份合法授权挪用到另一笔交易上。
时间戳与时钟偏移导致授权被判过期。沙箱机器与本地机器时区不一致时很容易踩到，统一用 UTC，并在校验时留一点容差。
重试造成重复下单。任何写操作都要带幂等键，客户端超时重试前先查询订单状态。
把私钥提交进了 git。.gitignore 里先写好 keys/ 和 .env，再开始写代码。
沙箱数据被重置。重置后旧的 mandate_id 会失效，报错时先确认沙箱是否被重启过。
排查顺序建议固定下来：先看 docker compose ps 与容器日志，再看审计日志里最后一个成功事件，最后才怀疑代码逻辑。这条顺序能省掉大量时间。
下一步建议
1. 加回归集：攒 20 条真实需求，每次改完比价逻辑跑一遍，看选择结果是否发生非预期漂移。
2. 把工具 MCP 化：检索、下单、查询订单状态封装成 MCP 工具，智能体换框架时这部分不用重写。
3. 做审批流：把人工确认从命令行搬到 IM 或审批系统，支持超时自动取消。
4. 接入告警：连续失败、金额异常、验签失败三类事件直接推送到值班渠道。
5. 看官方文档的扩展路径：AP2 覆盖多种支付轨道，具体支持范围与接入方式以官方文档当前版本为准。
把这条链路跑通一次，你对"智能体能不能自己花钱"这个问题的判断，会比读十篇文章都扎实。