@mindtdilly / Omnipay / commits / 05a2cb6

Add local-only x402 agentic commerce sim for Omnipay docs.

Ports the OpenAI cookbook controlled agentic commerce local path into examples/ plus a Mintlify guide, without AWS/AgentCore, live funds, or wallet secrets—so ProcureNet docs can teach 402→approval→proof→audit safely.

mindtdilly committed Sep 23, 2026 at 09:34 UTC 05a2cb64f1fb4551124595756ebc3af3b135fbe9
23 files changed +2049 -1
docs.json
+2 -1
@@ -36,7 +36,8 @@
36 "guides/integrate-checkout",
37 "guides/field-evaluation-payments",
38 "guides/webhooks",
39 - "guides/escrow-flows"
39 + "guides/escrow-flows",
40 + "guides/agentic-commerce-x402"
41 ]
42 },
43 {
examples/agentic-commerce-x402/.gitignore new
+9
@@ -0,0 +1,9 @@
1 +.venv/
2 +__pycache__/
3 +*.py[cod]
4 +.pytest_cache/
5 +.ruff_cache/
6 +dist/
7 +*.egg-info/
8 +.uv/
9 +uv.lock
examples/agentic-commerce-x402/LICENSE-NOTE.md new
+6
@@ -0,0 +1,6 @@
1 +# License note
2 +
3 +- Omnipay / ProcureNet documentation repository: MIT (see repository root `LICENSE`).
4 +- Adapted simulation modules originate from the OpenAI Cookbook partner example
5 + under that repository's Apache-2.0 / MIT terms. Attribution and SOURCE.md
6 + must be preserved with redistributions of the adapted code.
examples/agentic-commerce-x402/README.md new
+72
@@ -0,0 +1,72 @@
1 +# Agentic commerce x402 (local simulation)
2 +
3 +Local-only demonstration of an HTTP **402 Payment Required** → application
4 +approval → synthetic payment proof → receipt / audit loop for ProcureNet /
5 +Omnipay docs.
6 +
7 +> **Safety:** This example never moves real funds. There are no live payment
8 +> gates, no AWS credentials, no Bedrock calls, and no wallet private keys.
9 +> `value_transferred=false` always. Live AgentCore Payments is stubbed to
10 +> `SKIPPED` / `NOT_READY`.
11 +
12 +Adapted from the OpenAI Cookbook example *Controlled agentic commerce with
13 +AgentCore Payments*. See [SOURCE.md](./SOURCE.md) for attribution.
14 +
15 +## Four responsibilities
16 +
17 +1. **Agent** — proposes a paid resource fetch (scripted; no live model).
18 +2. **Application policy** — merchant allowlist, purpose, budgets, human approval.
19 +3. **Payment session (simulated)** — `LocalPaymentProcessor` mints synthetic proofs.
20 +4. **Paid API** — `SyntheticMerchant` serves a fictional supplier-risk report.
21 +
22 +## Setup
23 +
24 +```bash
25 +cd examples/agentic-commerce-x402
26 +uv sync
27 +uv run pytest
28 +```
29 +
30 +Or with pip:
31 +
32 +```bash
33 +python -m venv .venv && source .venv/bin/activate
34 +pip install -e ".[dev]" # or: pip install -e . && pip install pytest
35 +pytest
36 +```
37 +
38 +## Quick demo
39 +
40 +```python
41 +from datetime import UTC, datetime, timedelta
42 +from decimal import Decimal
43 +
44 +from omnipay_x402.demo import build_demo
45 +from omnipay_x402.merchant import RESOURCE_URL
46 +from omnipay_x402.models import ApprovalGrant, PurchaseRequest
47 +
48 +now = datetime.now(UTC)
49 +app, merchant, payments = build_demo(now)
50 +purchase = PurchaseRequest(
51 + request_id="request-001",
52 + resource_url=RESOURCE_URL,
53 + purpose="supplier_due_diligence",
54 + idempotency_key="purchase-001",
55 +)
56 +grant = ApprovalGrant(
57 + approval_id="approval-001",
58 + request_id=purchase.request_id,
59 + resource_url=purchase.resource_url,
60 + purpose=purchase.purpose,
61 + maximum_amount=Decimal("0.25"),
62 + approved_by="synthetic-reviewer",
63 + approved_at=now,
64 + expires_at=now + timedelta(minutes=10),
65 +)
66 +result = app.purchase(purchase, approval=grant, now=now)
67 +print(result.status, result.receipt.receipt_id, payments.charge_count)
68 +```
69 +
70 +## Mapping to ProcureNet docs
71 +
72 +See the Mintlify guide: [`guides/agentic-commerce-x402.mdx`](../../guides/agentic-commerce-x402.mdx).
examples/agentic-commerce-x402/SOURCE.md new
+33
@@ -0,0 +1,33 @@
1 +# Upstream source
2 +
3 +This example adapts the OpenAI Cookbook partner example:
4 +
5 +**Controlled agentic commerce with AgentCore Payments**
6 +
7 +- Upstream directory:
8 + `https://github.com/openai/openai-cookbook/tree/main/examples/partners/AWS/controlled_agentic_commerce_with_agentcore_payments`
9 +- Raw base used for module download:
10 + `https://raw.githubusercontent.com/openai/openai-cookbook/main/examples/partners/AWS/controlled_agentic_commerce_with_agentcore_payments/`
11 +- Notebook: `controlled_agentic_commerce.ipynb`
12 +- Authors (upstream): Deepak Jain and Sid Rampally
13 +- License: OpenAI Cookbook content is generally available under the Apache
14 + License 2.0 / MIT terms applicable to that repository. Retain attribution
15 + when redistributing adapted modules.
16 +- Snapshot retrieved: 2026-09-23 (exact upstream commit SHA not pinned here;
17 + re-check the cookbook tree before relying on API shapes outside this local
18 + sim).
19 +
20 +## What was adapted
21 +
22 +| Upstream | Omnipay local sim |
23 +|---|---|
24 +| `src/agentic_commerce/*` local modules | `src/omnipay_x402/` |
25 +| OpenAI Agents SDK + scripted `Model` | Tiny scripted `run_supplier_research` (no SDK) |
26 +| AgentCore Payments / Bedrock / AWS e2e | Omitted; `aws_stub.py` returns `SKIPPED` / `NOT_READY` |
27 +| Live testnet USDC | Never enabled (`value_transferred=false`) |
28 +
29 +## Intentional omissions
30 +
31 +- `agentcore_*.py` modules (payments, session, e2e, infrastructure, runtime, agent, application, learning_model)
32 +- Live wallet connectors, private keys, mnemonics, Coinbase CDP credentials
33 +- Requirements on `openai-agents`, `openai[bedrock]`, `botocore`, `bedrock-agentcore`
examples/agentic-commerce-x402/notebooks/local_x402_sim.ipynb new
+67
@@ -0,0 +1,67 @@
1 +{
2 + "nbformat": 4,
3 + "nbformat_minor": 5,
4 + "metadata": {
5 + "kernelspec": {
6 + "display_name": "Python 3",
7 + "language": "python",
8 + "name": "python3"
9 + },
10 + "language_info": {
11 + "name": "python",
12 + "pygments_lexer": "ipython3"
13 + }
14 + },
15 + "cells": [
16 + {
17 + "cell_type": "markdown",
18 + "metadata": {},
19 + "source": [
20 + "# Local x402 simulation (Omnipay)\n",
21 + "\n",
22 + "Local-only. No AWS, no live funds, no Bedrock. Synthetic proofs only.\n",
23 + "`value_transferred=false`."
24 + ]
25 + },
26 + {
27 + "cell_type": "code",
28 + "execution_count": null,
29 + "metadata": {},
30 + "outputs": [],
31 + "source": [
32 + "from datetime import UTC, datetime, timedelta\n",
33 + "from decimal import Decimal\n",
34 + "\n",
35 + "from omnipay_x402 import VALUE_TRANSFERRED, live_agentcore_status\n",
36 + "from omnipay_x402.demo import build_demo\n",
37 + "from omnipay_x402.merchant import RESOURCE_URL\n",
38 + "from omnipay_x402.models import ApprovalGrant, PurchaseRequest\n",
39 + "\n",
40 + "assert VALUE_TRANSFERRED is False\n",
41 + "print(live_agentcore_status())\n",
42 + "\n",
43 + "now = datetime.now(UTC)\n",
44 + "app, merchant, payments = build_demo(now)\n",
45 + "purchase = PurchaseRequest(\n",
46 + " request_id=\"request-001\",\n",
47 + " resource_url=RESOURCE_URL,\n",
48 + " purpose=\"supplier_due_diligence\",\n",
49 + " idempotency_key=\"purchase-001\",\n",
50 + ")\n",
51 + "grant = ApprovalGrant(\n",
52 + " approval_id=\"approval-001\",\n",
53 + " request_id=purchase.request_id,\n",
54 + " resource_url=purchase.resource_url,\n",
55 + " purpose=purchase.purpose,\n",
56 + " maximum_amount=Decimal(\"0.25\"),\n",
57 + " approved_by=\"synthetic-reviewer\",\n",
58 + " approved_at=now,\n",
59 + " expires_at=now + timedelta(minutes=10),\n",
60 + ")\n",
61 + "result = app.purchase(purchase, approval=grant, now=now)\n",
62 + "print(result.status, result.receipt.receipt_id)\n",
63 + "print([e.event_type for e in result.audit_events])"
64 + ]
65 + }
66 + ]
67 +}
examples/agentic-commerce-x402/pyproject.toml new
+29
@@ -0,0 +1,29 @@
1 +[project]
2 +name = "omnipay-x402"
3 +version = "0.1.0"
4 +description = "Local-only x402 agentic commerce simulation for Omnipay / ProcureNet docs"
5 +requires-python = ">=3.11,<3.14"
6 +dependencies = [
7 + "httpx>=0.28.0,<1",
8 + "pydantic>=2.13.0,<3",
9 +]
10 +license = { text = "MIT AND Apache-2.0" }
11 +readme = "README.md"
12 +
13 +[build-system]
14 +requires = ["hatchling"]
15 +build-backend = "hatchling.build"
16 +
17 +[dependency-groups]
18 +dev = [
19 + "pytest>=8.4.0,<9",
20 +]
21 +
22 +[tool.uv]
23 +package = true
24 +
25 +[tool.hatch.build.targets.wheel]
26 +packages = ["src/omnipay_x402"]
27 +
28 +[tool.pytest.ini_options]
29 +testpaths = ["tests"]
examples/agentic-commerce-x402/src/omnipay_x402/__init__.py new
+36
@@ -0,0 +1,36 @@
1 +"""Local-only x402 agentic commerce simulation for Omnipay / ProcureNet docs.
2 +
3 +Adapted from the OpenAI Cookbook example
4 +``controlled_agentic_commerce_with_agentcore_payments``. This package never
5 +moves real funds, never calls AWS/AgentCore, and uses only synthetic proofs.
6 +"""
7 +
8 +from .agent import SupplierResearchOutput, run_supplier_research
9 +from .application import CommerceApplication
10 +from .aws_stub import live_agentcore_status
11 +from .merchant import SyntheticMerchant
12 +from .models import (
13 + ApprovalGrant,
14 + CommercePolicy,
15 + PurchaseRequest,
16 + PurchaseResult,
17 +)
18 +from .payments import LocalPaymentProcessor
19 +from .policy import PolicyEngine
20 +
21 +VALUE_TRANSFERRED = False
22 +
23 +__all__ = [
24 + "ApprovalGrant",
25 + "CommerceApplication",
26 + "CommercePolicy",
27 + "LocalPaymentProcessor",
28 + "PolicyEngine",
29 + "PurchaseRequest",
30 + "PurchaseResult",
31 + "SupplierResearchOutput",
32 + "SyntheticMerchant",
33 + "VALUE_TRANSFERRED",
34 + "live_agentcore_status",
35 + "run_supplier_research",
36 +]
examples/agentic-commerce-x402/src/omnipay_x402/agent.py new
+185
@@ -0,0 +1,185 @@
1 +"""Scripted one-tool supplier research loop (no live model, no Agents SDK)."""
2 +
3 +from __future__ import annotations
4 +
5 +import json
6 +from dataclasses import dataclass, field
7 +from datetime import datetime
8 +from typing import Literal
9 +
10 +from pydantic import BaseModel, ConfigDict
11 +
12 +from .application import CommerceApplication
13 +from .errors import AgentResultInvalid
14 +from .models import ApprovalGrant, PurchaseResult
15 +from .tool import X402FetchTool, build_x402_fetch_tool
16 +
17 +DEFAULT_RESOURCE_URL = (
18 + "https://merchant.invalid/reports/SYNTH-SUPPLIER-RISK-001"
19 +)
20 +DEFAULT_PURPOSE = "supplier_due_diligence"
21 +
22 +
23 +class SupplierResearchOutput(BaseModel):
24 + """Typed, non-authoritative summary proposed by the scripted agent."""
25 +
26 + model_config = ConfigDict(frozen=True)
27 +
28 + status: Literal["completed"]
29 + report_id: Literal["SYNTH-SUPPLIER-RISK-001"]
30 + supplier: Literal["Northstar Components"]
31 + signals: tuple[str, ...]
32 + disclaimer: str
33 + receipt_id: str
34 + amount: Literal["0.25"]
35 + currency: Literal["USDC"]
36 + requires_human_approval: bool
37 +
38 +
39 +@dataclass
40 +class PurchaseResultRecorder:
41 + """Application-owned, per-run record of completed tool purchases."""
42 +
43 + results: list[PurchaseResult] = field(default_factory=list)
44 +
45 + def record(self, result: PurchaseResult) -> None:
46 + self.results.append(result)
47 +
48 +
49 +@dataclass(frozen=True)
50 +class SupplierResearchRun:
51 + """Agent proposal paired with the application evidence that validated it."""
52 +
53 + output: SupplierResearchOutput
54 + purchase: PurchaseResult
55 +
56 +
57 +@dataclass
58 +class ScriptedSupplierAgent:
59 + """Tiny stand-in for an Agents SDK Agent with one bound economic tool."""
60 +
61 + name: str
62 + tools: list[X402FetchTool]
63 + output_type: type[SupplierResearchOutput]
64 + instructions: str
65 +
66 +
67 +def build_supplier_research_agent(
68 + application: CommerceApplication,
69 + *,
70 + request_id: str,
71 + idempotency_key: str,
72 + approval: ApprovalGrant | None,
73 + recorder: PurchaseResultRecorder,
74 + now: datetime | None = None,
75 +) -> ScriptedSupplierAgent:
76 + """Create a fresh agent whose only economic tool is application-bound."""
77 +
78 + tool = build_x402_fetch_tool(
79 + application,
80 + request_id=request_id,
81 + idempotency_key=idempotency_key,
82 + approval=approval,
83 + on_purchase=recorder.record,
84 + now=now,
85 + )
86 + return ScriptedSupplierAgent(
87 + name="Synthetic supplier research agent",
88 + tools=[tool],
89 + output_type=SupplierResearchOutput,
90 + instructions=(
91 + "Use x402_fetch exactly once for the requested paid resource. "
92 + "The application—not you—owns merchant policy, budgets, human "
93 + "approval, payment execution, receipts, and audit state. Copy "
94 + "only facts returned by the tool. Never claim success after a "
95 + "denial, and never invent a report or receipt."
96 + ),
97 + )
98 +
99 +
100 +def validate_supplier_research_output(
101 + output: SupplierResearchOutput,
102 + recorder: PurchaseResultRecorder,
103 +) -> PurchaseResult:
104 + """Fail closed unless one tool purchase supports every returned field."""
105 +
106 + if len(recorder.results) != 1:
107 + raise AgentResultInvalid(
108 + "purchase_count_invalid",
109 + "A valid agent result requires exactly one completed purchase.",
110 + )
111 +
112 + purchase = recorder.results[0]
113 + expected = {
114 + "status": purchase.status,
115 + "report_id": purchase.report.report_id,
116 + "supplier": purchase.report.supplier,
117 + "signals": purchase.report.signals,
118 + "disclaimer": purchase.report.disclaimer,
119 + "receipt_id": purchase.receipt.receipt_id,
120 + "amount": str(purchase.receipt.amount),
121 + "currency": purchase.receipt.currency,
122 + "requires_human_approval": (purchase.authorization.requires_human_approval),
123 + }
124 + actual = output.model_dump()
125 + mismatches = sorted(
126 + field_name
127 + for field_name, expected_value in expected.items()
128 + if actual[field_name] != expected_value
129 + )
130 + if mismatches:
131 + raise AgentResultInvalid(
132 + "agent_output_mismatch",
133 + "Agent output did not match application evidence for: "
134 + + ", ".join(mismatches),
135 + )
136 + return purchase
137 +
138 +
139 +def run_supplier_research(
140 + application: CommerceApplication,
141 + *,
142 + request_id: str,
143 + idempotency_key: str,
144 + approval: ApprovalGrant | None,
145 + resource_url: str = DEFAULT_RESOURCE_URL,
146 + purpose: str = DEFAULT_PURPOSE,
147 + now: datetime | None = None,
148 +) -> SupplierResearchRun:
149 + """Run a deterministic one-tool loop and validate against app evidence.
150 +
151 + This replaces the OpenAI Agents SDK + scripted Model path from the
152 + upstream cookbook so Omnipay learners need only pydantic + httpx + pytest.
153 + """
154 +
155 + recorder = PurchaseResultRecorder()
156 + agent = build_supplier_research_agent(
157 + application,
158 + request_id=request_id,
159 + idempotency_key=idempotency_key,
160 + approval=approval,
161 + recorder=recorder,
162 + now=now,
163 + )
164 + tool = agent.tools[0]
165 + raw = tool(resource_url, purpose)
166 + evidence = json.loads(raw)
167 + if evidence.get("status") != "completed":
168 + raise AgentResultInvalid(
169 + "tool_denied",
170 + "The scripted agent received a denial instead of purchase evidence.",
171 + )
172 + report = evidence["report"]
173 + output = SupplierResearchOutput(
174 + status=evidence["status"],
175 + report_id=report["report_id"],
176 + supplier=report["supplier"],
177 + signals=tuple(report["signals"]),
178 + disclaimer=report["disclaimer"],
179 + receipt_id=evidence["receipt_id"],
180 + amount=str(evidence["amount"]),
181 + currency=evidence["currency"],
182 + requires_human_approval=evidence["requires_human_approval"],
183 + )
184 + purchase = validate_supplier_research_output(output, recorder)
185 + return SupplierResearchRun(output=output, purchase=purchase)
examples/agentic-commerce-x402/src/omnipay_x402/application.py new
+215
@@ -0,0 +1,215 @@
1 +"""Application controller for the deterministic x402 purchase sequence."""
2 +
3 +from __future__ import annotations
4 +
5 +from datetime import UTC, datetime
6 +
7 +import httpx
8 +from pydantic import ValidationError
9 +
10 +from .audit import AuditTrail
11 +from .codec import decode_json
12 +from .errors import MerchantRejectedPayment, PolicyDenied, ProtocolError
13 +from .merchant import (
14 + PAYMENT_REQUIRED_HEADER,
15 + PAYMENT_RESPONSE_HEADER,
16 + PAYMENT_SIGNATURE_HEADER,
17 +)
18 +from .models import (
19 + ApprovalGrant,
20 + AuditEventType,
21 + PaymentRequired,
22 + PurchaseRequest,
23 + PurchaseResult,
24 + SettlementResponse,
25 + SupplierRiskReport,
26 +)
27 +from .payments import LocalPaymentProcessor
28 +from .policy import PolicyEngine
29 +
30 +
31 +class CommerceApplication:
32 + """Orchestrate HTTP, authorization, payment, receipt, and audit state."""
33 +
34 + def __init__(
35 + self,
36 + *,
37 + client: httpx.Client,
38 + policy: PolicyEngine,
39 + payments: LocalPaymentProcessor,
40 + audit: AuditTrail | None = None,
41 + ) -> None:
42 + self.client = client
43 + self.policy = policy
44 + self.payments = payments
45 + self.audit = audit or AuditTrail()
46 +
47 + def purchase(
48 + self,
49 + request: PurchaseRequest,
50 + *,
51 + approval: ApprovalGrant | None = None,
52 + now: datetime | None = None,
53 + ) -> PurchaseResult:
54 + now = now or datetime.now(UTC)
55 + self.audit.append(
56 + request_id=request.request_id,
57 + event_type=AuditEventType.RESOURCE_REQUESTED,
58 + detail={
59 + "resource_url": str(request.resource_url),
60 + "purpose": request.purpose,
61 + },
62 + )
63 + preflight = self.policy.preflight(request, now=now)
64 + if not preflight.allowed:
65 + self.audit.append(
66 + request_id=request.request_id,
67 + event_type=AuditEventType.REQUEST_DENIED,
68 + detail={"code": preflight.code, "stage": "preflight"},
69 + )
70 + raise PolicyDenied(preflight.code, preflight.reason)
71 +
72 + initial = self.client.get(str(request.resource_url))
73 + if initial.status_code != 402:
74 + raise ProtocolError(
75 + "payment_challenge_expected",
76 + "The paid resource did not return HTTP 402.",
77 + )
78 +
79 + required = self._payment_required(initial)
80 + requirement = required.accepts[0]
81 + self.audit.append(
82 + request_id=request.request_id,
83 + event_type=AuditEventType.PAYMENT_REQUIRED,
84 + detail={
85 + "merchant_domain": requirement.merchant_domain,
86 + "amount": str(requirement.decimal_amount),
87 + "currency": requirement.currency,
88 + "network": requirement.network,
89 + },
90 + )
91 +
92 + decision = self.policy.authorize(
93 + request,
94 + required,
95 + approval=approval,
96 + now=now,
97 + )
98 + self.audit.append(
99 + request_id=request.request_id,
100 + event_type=AuditEventType.AUTHORIZATION_CHECKED,
101 + detail={
102 + "allowed": decision.allowed,
103 + "code": decision.code,
104 + "requires_human_approval": (decision.requires_human_approval),
105 + },
106 + )
107 + if not decision.allowed:
108 + self.audit.append(
109 + request_id=request.request_id,
110 + event_type=AuditEventType.REQUEST_DENIED,
111 + detail={"code": decision.code},
112 + )
113 + raise PolicyDenied(decision.code, decision.reason)
114 +
115 + self.audit.append(
116 + request_id=request.request_id,
117 + event_type=AuditEventType.PAYMENT_ATTEMPTED,
118 + detail={"idempotency_key": request.idempotency_key},
119 + )
120 + authorization = self.payments.authorize(request, requirement)
121 + self.policy.record(authorization.receipt)
122 + self.audit.append(
123 + request_id=request.request_id,
124 + event_type=(
125 + AuditEventType.PROOF_REUSED
126 + if authorization.receipt.reused
127 + else AuditEventType.PROOF_CREATED
128 + ),
129 + detail={"receipt_id": authorization.receipt.receipt_id},
130 + )
131 +
132 + self.audit.append(
133 + request_id=request.request_id,
134 + event_type=AuditEventType.MERCHANT_RETRY,
135 + detail={"payment_header": PAYMENT_SIGNATURE_HEADER},
136 + )
137 + paid = self.client.get(
138 + str(request.resource_url),
139 + headers={
140 + PAYMENT_SIGNATURE_HEADER: authorization.proof_header,
141 + },
142 + )
143 + if paid.status_code != 200:
144 + raise MerchantRejectedPayment(
145 + "merchant_rejected_payment",
146 + "The merchant did not accept the synthetic payment proof.",
147 + )
148 +
149 + settlement = self._settlement(paid)
150 + if (
151 + not settlement.success
152 + or settlement.transaction != authorization.receipt.transaction
153 + ):
154 + raise ProtocolError(
155 + "settlement_receipt_mismatch",
156 + "The settlement response does not match the local receipt.",
157 + )
158 + try:
159 + report = SupplierRiskReport.model_validate(paid.json())
160 + except (ValueError, ValidationError) as exc:
161 + raise ProtocolError(
162 + "invalid_paid_resource",
163 + "The paid resource did not match the typed report contract.",
164 + ) from exc
165 +
166 + self.audit.append(
167 + request_id=request.request_id,
168 + event_type=AuditEventType.CONTENT_RETURNED,
169 + detail={
170 + "report_id": report.report_id,
171 + "receipt_id": authorization.receipt.receipt_id,
172 + },
173 + )
174 + return PurchaseResult(
175 + status="completed",
176 + request_id=request.request_id,
177 + authorization=decision,
178 + receipt=authorization.receipt,
179 + report=report,
180 + audit_events=self.audit.for_request(request.request_id),
181 + )
182 +
183 + @staticmethod
184 + def _payment_required(response: httpx.Response) -> PaymentRequired:
185 + header = response.headers.get(PAYMENT_REQUIRED_HEADER)
186 + if header is None:
187 + raise ProtocolError(
188 + "payment_required_header_missing",
189 + "HTTP 402 did not include PAYMENT-REQUIRED.",
190 + )
191 + payload = decode_json(header, header_name=PAYMENT_REQUIRED_HEADER)
192 + try:
193 + return PaymentRequired.model_validate(payload)
194 + except ValidationError as exc:
195 + raise ProtocolError(
196 + "invalid_payment_requirement",
197 + "PAYMENT-REQUIRED does not match the expected contract.",
198 + ) from exc
199 +
200 + @staticmethod
201 + def _settlement(response: httpx.Response) -> SettlementResponse:
202 + header = response.headers.get(PAYMENT_RESPONSE_HEADER)
203 + if header is None:
204 + raise ProtocolError(
205 + "payment_response_header_missing",
206 + "Paid response did not include PAYMENT-RESPONSE.",
207 + )
208 + payload = decode_json(header, header_name=PAYMENT_RESPONSE_HEADER)
209 + try:
210 + return SettlementResponse.model_validate(payload)
211 + except ValidationError as exc:
212 + raise ProtocolError(
213 + "invalid_settlement_response",
214 + "PAYMENT-RESPONSE does not match the expected contract.",
215 + ) from exc
examples/agentic-commerce-x402/src/omnipay_x402/audit.py new
+47
@@ -0,0 +1,47 @@
1 +"""Append-only in-memory audit trail for the deterministic local path."""
2 +
3 +from __future__ import annotations
4 +
5 +from collections.abc import Callable
6 +from datetime import UTC, datetime
7 +from typing import Any
8 +
9 +from .models import AuditEvent, AuditEventType
10 +
11 +Clock = Callable[[], datetime]
12 +
13 +
14 +def utc_now() -> datetime:
15 + return datetime.now(UTC)
16 +
17 +
18 +class AuditTrail:
19 + """Record ordered events without exposing mutable internal state."""
20 +
21 + def __init__(self, *, clock: Clock = utc_now) -> None:
22 + self._clock = clock
23 + self._events: list[AuditEvent] = []
24 +
25 + def append(
26 + self,
27 + *,
28 + request_id: str,
29 + event_type: AuditEventType,
30 + detail: dict[str, Any] | None = None,
31 + ) -> AuditEvent:
32 + event = AuditEvent(
33 + sequence=len(self._events) + 1,
34 + occurred_at=self._clock(),
35 + request_id=request_id,
36 + event_type=event_type,
37 + detail=detail or {},
38 + )
39 + self._events.append(event)
40 + return event
41 +
42 + def for_request(self, request_id: str) -> tuple[AuditEvent, ...]:
43 + return tuple(event for event in self._events if event.request_id == request_id)
44 +
45 + @property
46 + def events(self) -> tuple[AuditEvent, ...]:
47 + return tuple(self._events)
examples/agentic-commerce-x402/src/omnipay_x402/aws_stub.py new
+43
@@ -0,0 +1,43 @@
1 +"""Permanent stubs for live AgentCore / AWS payment paths.
2 +
3 +Live gates are intentionally omitted from this Omnipay adaptation. Callers
4 +receive SKIPPED / NOT_READY without any network I/O or credential use.
5 +"""
6 +
7 +from __future__ import annotations
8 +
9 +from typing import Any, Literal
10 +
11 +
12 +def live_agentcore_status() -> dict[str, Any]:
13 + """Report that the connected AgentCore Payments path is out of scope."""
14 +
15 + return {
16 + "status": "SKIPPED",
17 + "reason": "NOT_READY",
18 + "provider": "agentcore_payments",
19 + "value_transferred": False,
20 + "network_calls": False,
21 + "message": (
22 + "Live Amazon Bedrock AgentCore Payments is intentionally omitted "
23 + "from the Omnipay local simulation. Use LocalPaymentProcessor only."
24 + ),
25 + }
26 +
27 +
28 +def create_live_payment_session(*_args: Any, **_kwargs: Any) -> dict[str, Any]:
29 + """Refuse any attempt to open a live payment session."""
30 +
31 + return live_agentcore_status()
32 +
33 +
34 +def authorize_live_payment(*_args: Any, **_kwargs: Any) -> dict[str, Literal[False] | str]:
35 + """Refuse any attempt to authorize a live payment."""
36 +
37 + status = live_agentcore_status()
38 + return {
39 + "ok": False,
40 + "status": status["status"],
41 + "reason": status["reason"],
42 + "value_transferred": False,
43 + }
examples/agentic-commerce-x402/src/omnipay_x402/codec.py new
+38
@@ -0,0 +1,38 @@
1 +"""Base64 JSON helpers used by the protocol-shaped local transport."""
2 +
3 +from __future__ import annotations
4 +
5 +import base64
6 +import json
7 +from typing import Any
8 +
9 +from pydantic import BaseModel
10 +
11 +from .errors import ProtocolError
12 +
13 +
14 +def encode_model(model: BaseModel) -> str:
15 + """Serialize a model to standard Base64-encoded JSON."""
16 +
17 + payload = model.model_dump_json(by_alias=True, exclude_none=True)
18 + return base64.b64encode(payload.encode("utf-8")).decode("ascii")
19 +
20 +
21 +def decode_json(value: str, *, header_name: str) -> dict[str, Any]:
22 + """Decode a Base64 JSON header without leaking its raw value in errors."""
23 +
24 + try:
25 + decoded = base64.b64decode(value, validate=True)
26 + payload = json.loads(decoded)
27 + except (ValueError, UnicodeDecodeError, json.JSONDecodeError) as exc:
28 + raise ProtocolError(
29 + "malformed_protocol_header",
30 + f"{header_name} is not valid Base64-encoded JSON.",
31 + ) from exc
32 +
33 + if not isinstance(payload, dict):
34 + raise ProtocolError(
35 + "malformed_protocol_header",
36 + f"{header_name} must decode to a JSON object.",
37 + )
38 + return payload
examples/agentic-commerce-x402/src/omnipay_x402/demo.py new
+35
@@ -0,0 +1,35 @@
1 +"""Build the deterministic demo used by the notebook and tests."""
2 +
3 +from __future__ import annotations
4 +
5 +from datetime import datetime, timedelta
6 +from decimal import Decimal
7 +
8 +from .application import CommerceApplication
9 +from .merchant import SyntheticMerchant
10 +from .models import CommercePolicy
11 +from .payments import LocalPaymentProcessor
12 +from .policy import PolicyEngine
13 +
14 +
15 +def build_demo(
16 + now: datetime,
17 +) -> tuple[CommerceApplication, SyntheticMerchant, LocalPaymentProcessor]:
18 + payments = LocalPaymentProcessor()
19 + merchant = SyntheticMerchant(payments, now=now)
20 + policy = PolicyEngine(
21 + CommercePolicy(
22 + allowed_merchants=frozenset({"merchant.invalid"}),
23 + allowed_purposes=frozenset({"supplier_due_diligence"}),
24 + per_request_limit=Decimal("0.50"),
25 + per_run_limit=Decimal("1.00"),
26 + approval_threshold=Decimal("0.10"),
27 + session_expires_at=now + timedelta(minutes=15),
28 + )
29 + )
30 + application = CommerceApplication(
31 + client=merchant.client(),
32 + policy=policy,
33 + payments=payments,
34 + )
35 + return application, merchant, payments
examples/agentic-commerce-x402/src/omnipay_x402/errors.py new
+44
@@ -0,0 +1,44 @@
1 +"""Safe, typed errors for the synthetic commerce flow."""
2 +
3 +from __future__ import annotations
4 +
5 +from collections.abc import Mapping
6 +
7 +
8 +class CommerceError(RuntimeError):
9 + """Base error with a stable, non-secret error code."""
10 +
11 + def __init__(
12 + self,
13 + code: str,
14 + message: str,
15 + *,
16 + diagnostics: Mapping[str, object] | None = None,
17 + ) -> None:
18 + super().__init__(message)
19 + self.code = code
20 + self.diagnostics = dict(diagnostics or {})
21 +
22 +
23 +class ProtocolError(CommerceError):
24 + """The merchant response did not satisfy the expected x402 shape."""
25 +
26 +
27 +class PolicyDenied(CommerceError):
28 + """Application-owned policy denied the proposed purchase."""
29 +
30 +
31 +class IdempotencyConflict(CommerceError):
32 + """An idempotency key was reused for a different purchase."""
33 +
34 +
35 +class MerchantRejectedPayment(CommerceError):
36 + """The synthetic merchant rejected the supplied payment proof."""
37 +
38 +
39 +class LivePaymentDisabled(CommerceError):
40 + """A networked live payment was attempted; Omnipay sim refuses it."""
41 +
42 +
43 +class AgentResultInvalid(CommerceError):
44 + """The agent result did not match application-observed purchase evidence."""
examples/agentic-commerce-x402/src/omnipay_x402/learning_model.py new
+12
@@ -0,0 +1,12 @@
1 +"""Scripted learning loop without OpenAI Agents SDK or network inference.
2 +
3 +Upstream cookbook used ``ScriptedCommerceLearningModel`` implementing the
4 +Agents SDK ``Model`` interface. Omnipay keeps the same teaching intent via
5 +``run_supplier_research`` in ``agent.py``.
6 +"""
7 +
8 +from __future__ import annotations
9 +
10 +from .agent import run_supplier_research
11 +
12 +__all__ = ["run_supplier_research"]
examples/agentic-commerce-x402/src/omnipay_x402/merchant.py new
+127
@@ -0,0 +1,127 @@
1 +"""In-memory synthetic merchant that exposes a protocol-shaped x402 flow."""
2 +
3 +from __future__ import annotations
4 +
5 +from datetime import UTC, datetime, timedelta
6 +from decimal import Decimal
7 +
8 +import httpx
9 +
10 +from .codec import encode_model
11 +from .errors import CommerceError
12 +from .models import (
13 + PaymentRequired,
14 + PaymentRequirement,
15 + ResourceInfo,
16 + SupplierRiskReport,
17 +)
18 +from .payments import LocalPaymentProcessor
19 +
20 +PAYMENT_REQUIRED_HEADER = "PAYMENT-REQUIRED"
21 +PAYMENT_SIGNATURE_HEADER = "PAYMENT-SIGNATURE"
22 +PAYMENT_RESPONSE_HEADER = "PAYMENT-RESPONSE"
23 +
24 +RESOURCE_URL = "https://merchant.invalid/reports/SYNTH-SUPPLIER-RISK-001"
25 +
26 +
27 +class SyntheticMerchant:
28 + """Serve one fictional paid report through an in-memory HTTP transport."""
29 +
30 + def __init__(
31 + self,
32 + payment_processor: LocalPaymentProcessor,
33 + *,
34 + price: Decimal = Decimal("0.25"),
35 + now: datetime | None = None,
36 + malformed_challenge: bool = False,
37 + ) -> None:
38 + self.payment_processor = payment_processor
39 + self.request_count = 0
40 + self.fulfilled_count = 0
41 + self.invalid_proof_count = 0
42 + self.malformed_challenge = malformed_challenge
43 + issued_at = now or datetime.now(UTC)
44 + expires_at = issued_at + timedelta(minutes=5)
45 + amount = str(int(price * Decimal(1_000_000)))
46 +
47 + self.requirement = PaymentRequirement(
48 + amount=amount,
49 + asset="synthetic-usdc-base-sepolia",
50 + payTo="synthetic-merchant-wallet",
51 + maxTimeoutSeconds=300,
52 + extra={
53 + "challengeId": "challenge-synth-report-001",
54 + "currency": "USDC",
55 + "decimals": 6,
56 + "merchantDomain": "merchant.invalid",
57 + "issuedAt": issued_at.isoformat(),
58 + "expiresAt": expires_at.isoformat(),
59 + "simulation": True,
60 + },
61 + )
62 + self.payment_required = PaymentRequired(
63 + resource=ResourceInfo(
64 + url=RESOURCE_URL,
65 + description="Synthetic supplier-risk report",
66 + mimeType="application/json",
67 + ),
68 + accepts=(self.requirement,),
69 + error="Payment is required for this synthetic report.",
70 + )
71 + self.report = SupplierRiskReport(
72 + report_id="SYNTH-SUPPLIER-RISK-001",
73 + supplier="Northstar Components",
74 + generated_from="synthetic data",
75 + signals=(
76 + "Delivery concentration increased in the last synthetic quarter.",
77 + "Two fictional certifications require manual verification.",
78 + ),
79 + disclaimer=(
80 + "Training-only synthetic report; not purchasing, legal, "
81 + "compliance, or risk advice."
82 + ),
83 + )
84 +
85 + def client(self) -> httpx.Client:
86 + return httpx.Client(transport=httpx.MockTransport(self._handle))
87 +
88 + def _handle(self, request: httpx.Request) -> httpx.Response:
89 + self.request_count += 1
90 + if request.method != "GET" or str(request.url) != RESOURCE_URL:
91 + return httpx.Response(
92 + 404,
93 + json={"error": "synthetic_resource_not_found"},
94 + )
95 +
96 + proof_header = request.headers.get(PAYMENT_SIGNATURE_HEADER)
97 + if proof_header is None:
98 + challenge = (
99 + "not-base64-json"
100 + if self.malformed_challenge
101 + else encode_model(self.payment_required)
102 + )
103 + return httpx.Response(
104 + 402,
105 + headers={PAYMENT_REQUIRED_HEADER: challenge},
106 + json={"error": "payment_required"},
107 + )
108 +
109 + try:
110 + settlement = self.payment_processor.verify(
111 + proof_header,
112 + self.requirement,
113 + )
114 + except CommerceError as exc:
115 + self.invalid_proof_count += 1
116 + return httpx.Response(
117 + 402,
118 + headers={PAYMENT_REQUIRED_HEADER: encode_model(self.payment_required)},
119 + json={"error": exc.code},
120 + )
121 +
122 + self.fulfilled_count += 1
123 + return httpx.Response(
124 + 200,
125 + headers={PAYMENT_RESPONSE_HEADER: encode_model(settlement)},
126 + json=self.report.model_dump(mode="json"),
127 + )
examples/agentic-commerce-x402/src/omnipay_x402/models.py new
+230
@@ -0,0 +1,230 @@
1 +"""Typed contracts for authorization, x402 transport, receipts, and audit."""
2 +
3 +from __future__ import annotations
4 +
5 +from datetime import datetime
6 +from decimal import Decimal
7 +from enum import StrEnum
8 +from typing import Any, Literal
9 +
10 +from pydantic import (
11 + AwareDatetime,
12 + BaseModel,
13 + ConfigDict,
14 + Field,
15 + HttpUrl,
16 + computed_field,
17 + field_validator,
18 +)
19 +from pydantic_core import PydanticCustomError
20 +
21 +
22 +class FrozenModel(BaseModel):
23 + """Immutable model used for application and protocol contracts."""
24 +
25 + model_config = ConfigDict(frozen=True, populate_by_name=True)
26 +
27 +
28 +class AuditEventType(StrEnum):
29 + RESOURCE_REQUESTED = "resource_requested"
30 + PAYMENT_REQUIRED = "payment_required"
31 + AUTHORIZATION_CHECKED = "authorization_checked"
32 + PAYMENT_ATTEMPTED = "payment_attempted"
33 + PROOF_CREATED = "proof_created"
34 + PROOF_REUSED = "proof_reused"
35 + MERCHANT_RETRY = "merchant_retry"
36 + CONTENT_RETURNED = "content_returned"
37 + REQUEST_DENIED = "request_denied"
38 +
39 +
40 +class AuditEvent(FrozenModel):
41 + sequence: int = Field(ge=1)
42 + occurred_at: datetime
43 + request_id: str
44 + event_type: AuditEventType
45 + detail: dict[str, Any] = Field(default_factory=dict)
46 +
47 +
48 +class CommercePolicy(FrozenModel):
49 + allowed_merchants: frozenset[str]
50 + allowed_purposes: frozenset[str]
51 + per_request_limit: Decimal = Field(gt=0)
52 + per_run_limit: Decimal = Field(gt=0)
53 + approval_threshold: Decimal = Field(ge=0)
54 + currency: Literal["USDC"] = "USDC"
55 + network: Literal["eip155:84532"] = "eip155:84532"
56 + session_expires_at: datetime
57 +
58 +
59 +class PurchaseRequest(FrozenModel):
60 + request_id: str = Field(min_length=1)
61 + resource_url: HttpUrl
62 + purpose: str = Field(min_length=1)
63 + idempotency_key: str = Field(min_length=8)
64 +
65 +
66 +class ApprovalGrant(FrozenModel):
67 + approval_id: str = Field(min_length=1)
68 + request_id: str = Field(min_length=1)
69 + resource_url: HttpUrl
70 + purpose: str = Field(min_length=1)
71 + maximum_amount: Decimal = Field(gt=0)
72 + currency: Literal["USDC"] = "USDC"
73 + approved_by: str = Field(min_length=1)
74 + approved_at: datetime
75 + expires_at: datetime
76 +
77 +
78 +class ResourceInfo(FrozenModel):
79 + url: HttpUrl
80 + description: str
81 + mime_type: str = Field(alias="mimeType")
82 +
83 +
84 +class PaymentRequirementExtra(FrozenModel):
85 + """Validated x402 metadata used by local policy computed fields."""
86 +
87 + model_config = ConfigDict(frozen=True, populate_by_name=True, extra="forbid")
88 +
89 + decimals: int = Field(strict=True, ge=0)
90 + currency: Literal["USDC"]
91 + merchant_domain: str = Field(
92 + alias="merchantDomain",
93 + strict=True,
94 + min_length=1,
95 + )
96 + challenge_id: str = Field(
97 + alias="challengeId",
98 + strict=True,
99 + min_length=1,
100 + )
101 + issued_at: AwareDatetime | None = Field(default=None, alias="issuedAt")
102 + expires_at: AwareDatetime = Field(alias="expiresAt")
103 + simulation: bool | None = Field(default=None, strict=True)
104 +
105 + @field_validator("issued_at", "expires_at", mode="before")
106 + @classmethod
107 + def require_iso_timestamp_text(cls, value: object) -> object:
108 + if value is None:
109 + return value
110 + if not isinstance(value, str):
111 + raise PydanticCustomError(
112 + "x402_timestamp_type",
113 + "x402 timestamps must be ISO 8601 strings.",
114 + )
115 + return value
116 +
117 +
118 +class PaymentRequirement(FrozenModel):
119 + """Protocol-shaped x402 V2 exact-payment requirement.
120 +
121 + The local proof is deliberately synthetic. These fields mirror the
122 + current x402 V2 transport shape but do not claim chain conformance.
123 + """
124 +
125 + scheme: Literal["exact"] = "exact"
126 + network: Literal["eip155:84532"] = "eip155:84532"
127 + amount: str = Field(pattern=r"^[0-9]+$")
128 + asset: str
129 + pay_to: str = Field(alias="payTo")
130 + max_timeout_seconds: int = Field(alias="maxTimeoutSeconds", gt=0)
131 + extra: PaymentRequirementExtra
132 +
133 + @computed_field
134 + @property
135 + def decimal_amount(self) -> Decimal:
136 + return Decimal(self.amount) / (Decimal(10) ** self.extra.decimals)
137 +
138 + @computed_field
139 + @property
140 + def currency(self) -> Literal["USDC"]:
141 + return self.extra.currency
142 +
143 + @computed_field
144 + @property
145 + def merchant_domain(self) -> str:
146 + return self.extra.merchant_domain
147 +
148 + @computed_field
149 + @property
150 + def challenge_id(self) -> str:
151 + return self.extra.challenge_id
152 +
153 + @computed_field
154 + @property
155 + def expires_at(self) -> datetime:
156 + return self.extra.expires_at
157 +
158 +
159 +class PaymentRequired(FrozenModel):
160 + x402_version: Literal[2] = Field(alias="x402Version", default=2)
161 + resource: ResourceInfo
162 + accepts: tuple[PaymentRequirement, ...] = Field(min_length=1)
163 + error: str | None = None
164 +
165 +
166 +class PaymentPayload(FrozenModel):
167 + x402_version: Literal[2] = Field(alias="x402Version", default=2)
168 + accepted: PaymentRequirement
169 + payload: dict[str, str]
170 +
171 +
172 +class SettlementResponse(FrozenModel):
173 + success: bool
174 + transaction: str
175 + network: str
176 + payer: str
177 +
178 +
179 +class AuthorizationDecision(FrozenModel):
180 + allowed: bool
181 + code: str
182 + reason: str
183 + requires_human_approval: bool
184 + approved_amount: Decimal | None = None
185 +
186 +
187 +class PaymentReceipt(FrozenModel):
188 + receipt_id: str
189 + request_id: str
190 + idempotency_key: str
191 + merchant_domain: str
192 + resource_url: HttpUrl
193 + amount: Decimal
194 + currency: Literal["USDC"]
195 + network: str
196 + transaction: str
197 + reused: bool = False
198 +
199 +
200 +class PaymentAuthorization(FrozenModel):
201 + proof_header: str
202 + receipt: PaymentReceipt
203 +
204 +
205 +class SupplierRiskReport(FrozenModel):
206 + report_id: Literal["SYNTH-SUPPLIER-RISK-001"]
207 + supplier: Literal["Northstar Components"]
208 + generated_from: Literal["synthetic data"]
209 + signals: tuple[str, ...]
210 + disclaimer: str
211 +
212 +
213 +class PurchaseResult(FrozenModel):
214 + status: Literal["completed"]
215 + request_id: str
216 + authorization: AuthorizationDecision
217 + receipt: PaymentReceipt
218 + report: SupplierRiskReport
219 + audit_events: tuple[AuditEvent, ...]
220 +
221 +
222 +class AgentPurchaseEvidence(FrozenModel):
223 + """Minimum purchase evidence returned to the model by the function tool."""
224 +
225 + status: Literal["completed"]
226 + report: SupplierRiskReport
227 + receipt_id: str
228 + amount: Decimal
229 + currency: Literal["USDC"]
230 + requires_human_approval: bool
examples/agentic-commerce-x402/src/omnipay_x402/payments.py new
+158
@@ -0,0 +1,158 @@
1 +"""Deterministic local proof creation; no wallet, chain, or value transfer."""
2 +
3 +from __future__ import annotations
4 +
5 +import hashlib
6 +import json
7 +from dataclasses import dataclass
8 +
9 +from pydantic import ValidationError
10 +
11 +from .codec import decode_json, encode_model
12 +from .errors import IdempotencyConflict, ProtocolError
13 +from .models import (
14 + PaymentAuthorization,
15 + PaymentPayload,
16 + PaymentReceipt,
17 + PaymentRequirement,
18 + PurchaseRequest,
19 + SettlementResponse,
20 +)
21 +
22 +
23 +@dataclass(frozen=True)
24 +class _StoredAuthorization:
25 + fingerprint: str
26 + proof_header: str
27 + receipt: PaymentReceipt
28 +
29 +
30 +class LocalPaymentProcessor:
31 + """A fake local payment adapter with idempotent proof generation."""
32 +
33 + payer = "synthetic-test-payer"
34 +
35 + def __init__(self) -> None:
36 + self._authorizations: dict[str, _StoredAuthorization] = {}
37 +
38 + @property
39 + def charge_count(self) -> int:
40 + return len(self._authorizations)
41 +
42 + def authorize(
43 + self,
44 + request: PurchaseRequest,
45 + requirement: PaymentRequirement,
46 + ) -> PaymentAuthorization:
47 + fingerprint = self._fingerprint(request, requirement)
48 + stored = self._authorizations.get(request.idempotency_key)
49 + if stored is not None:
50 + if stored.fingerprint != fingerprint:
51 + raise IdempotencyConflict(
52 + "idempotency_conflict",
53 + "The idempotency key is already bound to another purchase.",
54 + )
55 + return PaymentAuthorization(
56 + proof_header=stored.proof_header,
57 + receipt=stored.receipt.model_copy(update={"reused": True}),
58 + )
59 +
60 + digest = hashlib.sha256(
61 + f"{request.idempotency_key}:{fingerprint}".encode()
62 + ).hexdigest()
63 + receipt = PaymentReceipt(
64 + receipt_id=f"receipt-{digest[:16]}",
65 + request_id=request.request_id,
66 + idempotency_key=request.idempotency_key,
67 + merchant_domain=requirement.merchant_domain,
68 + resource_url=request.resource_url,
69 + amount=requirement.decimal_amount,
70 + currency=requirement.currency,
71 + network=requirement.network,
72 + transaction=f"synthetic-{digest[16:40]}",
73 + )
74 + payload = PaymentPayload(
75 + accepted=requirement,
76 + payload={
77 + "proof": f"synthetic-proof-{digest[40:]}",
78 + "receiptId": receipt.receipt_id,
79 + },
80 + )
81 + proof_header = encode_model(payload)
82 + self._authorizations[request.idempotency_key] = _StoredAuthorization(
83 + fingerprint=fingerprint,
84 + proof_header=proof_header,
85 + receipt=receipt,
86 + )
87 + return PaymentAuthorization(
88 + proof_header=proof_header,
89 + receipt=receipt,
90 + )
91 +
92 + def verify(
93 + self,
94 + proof_header: str,
95 + requirement: PaymentRequirement,
96 + ) -> SettlementResponse:
97 + payload_dict = decode_json(
98 + proof_header,
99 + header_name="PAYMENT-SIGNATURE",
100 + )
101 + try:
102 + payload = PaymentPayload.model_validate(payload_dict)
103 + except ValidationError as exc:
104 + raise ProtocolError(
105 + "invalid_payment_payload",
106 + "PAYMENT-SIGNATURE does not match the expected payload.",
107 + ) from exc
108 +
109 + if payload.accepted != requirement:
110 + raise ProtocolError(
111 + "payment_requirement_mismatch",
112 + "The proof does not match the merchant requirement.",
113 + )
114 +
115 + receipt_id = payload.payload.get("receiptId")
116 + proof = payload.payload.get("proof")
117 + stored = next(
118 + (
119 + candidate
120 + for candidate in self._authorizations.values()
121 + if candidate.receipt.receipt_id == receipt_id
122 + ),
123 + None,
124 + )
125 + if (
126 + stored is None
127 + or stored.proof_header != proof_header
128 + or not proof
129 + or not proof.startswith("synthetic-proof-")
130 + ):
131 + raise ProtocolError(
132 + "unrecognized_synthetic_proof",
133 + "The local payment proof is not recognized.",
134 + )
135 +
136 + return SettlementResponse(
137 + success=True,
138 + transaction=stored.receipt.transaction,
139 + network=stored.receipt.network,
140 + payer=self.payer,
141 + )
142 +
143 + @staticmethod
144 + def _fingerprint(
145 + request: PurchaseRequest,
146 + requirement: PaymentRequirement,
147 + ) -> str:
148 + stable = {
149 + "resource_url": str(request.resource_url),
150 + "purpose": request.purpose,
151 + "challenge_id": requirement.challenge_id,
152 + "amount": requirement.amount,
153 + "asset": requirement.asset,
154 + "network": requirement.network,
155 + "pay_to": requirement.pay_to,
156 + }
157 + raw = json.dumps(stable, sort_keys=True, separators=(",", ":"))
158 + return hashlib.sha256(raw.encode()).hexdigest()
examples/agentic-commerce-x402/src/omnipay_x402/policy.py new
+237
@@ -0,0 +1,237 @@
1 +"""Deterministic application authorization for synthetic paid resources."""
2 +
3 +from __future__ import annotations
4 +
5 +from datetime import UTC, datetime
6 +from decimal import Decimal
7 +
8 +from .models import (
9 + ApprovalGrant,
10 + AuthorizationDecision,
11 + CommercePolicy,
12 + PaymentReceipt,
13 + PaymentRequired,
14 + PurchaseRequest,
15 +)
16 +
17 +
18 +class PolicyEngine:
19 + """Authorize purchases independently of model reasoning."""
20 +
21 + def __init__(self, policy: CommercePolicy) -> None:
22 + self.policy = policy
23 + self._spent_by_idempotency: dict[str, Decimal] = {}
24 +
25 + @property
26 + def spent(self) -> Decimal:
27 + return sum(self._spent_by_idempotency.values(), start=Decimal(0))
28 +
29 + def preflight(
30 + self,
31 + request: PurchaseRequest,
32 + *,
33 + now: datetime | None = None,
34 + ) -> AuthorizationDecision:
35 + """Reject unsafe destinations and purposes before any HTTP request."""
36 +
37 + now = now or datetime.now(UTC)
38 + if request.resource_url.scheme != "https":
39 + return self._deny("https_required", "Paid resources must use HTTPS.")
40 + if request.resource_url.host not in self.policy.allowed_merchants:
41 + return self._deny(
42 + "merchant_not_allowed",
43 + "The merchant is not in the application allowlist.",
44 + )
45 + if request.purpose not in self.policy.allowed_purposes:
46 + return self._deny(
47 + "purpose_not_allowed",
48 + "The requested purchase purpose is not allowed.",
49 + )
50 + if now >= self.policy.session_expires_at:
51 + return self._deny(
52 + "session_expired",
53 + "The application payment session has expired.",
54 + )
55 + return AuthorizationDecision(
56 + allowed=True,
57 + code="preflight_allowed",
58 + reason="The request destination and purpose satisfy preflight policy.",
59 + requires_human_approval=False,
60 + )
61 +
62 + def authorize(
63 + self,
64 + request: PurchaseRequest,
65 + required: PaymentRequired,
66 + *,
67 + approval: ApprovalGrant | None,
68 + now: datetime | None = None,
69 + ) -> AuthorizationDecision:
70 + now = now or datetime.now(UTC)
71 + requirement = required.accepts[0]
72 + host = request.resource_url.host
73 + amount = requirement.decimal_amount
74 +
75 + preflight = self.preflight(request, now=now)
76 + if not preflight.allowed:
77 + return preflight
78 + if required.resource.url != request.resource_url:
79 + return self._deny(
80 + "resource_mismatch",
81 + "The payment challenge refers to a different resource.",
82 + )
83 + if requirement.merchant_domain != host:
84 + return self._deny(
85 + "merchant_mismatch",
86 + "The payment challenge merchant does not match the URL.",
87 + )
88 + return self.authorize_challenge(
89 + request,
90 + merchant_domain=requirement.merchant_domain,
91 + network=requirement.network,
92 + currency=requirement.currency,
93 + amount=amount,
94 + approval=approval,
95 + expires_at=requirement.expires_at,
96 + now=now,
97 + )
98 +
99 + def authorize_challenge(
100 + self,
101 + request: PurchaseRequest,
102 + *,
103 + merchant_domain: str,
104 + network: str,
105 + currency: str,
106 + amount: Decimal,
107 + approval: ApprovalGrant | None,
108 + expires_at: datetime | None = None,
109 + now: datetime | None = None,
110 + ) -> AuthorizationDecision:
111 + """Authorize normalized challenge facts from any x402 transport."""
112 +
113 + now = now or datetime.now(UTC)
114 + host = request.resource_url.host
115 + preflight = self.preflight(request, now=now)
116 + if not preflight.allowed:
117 + return preflight
118 + if merchant_domain != host:
119 + return self._deny(
120 + "merchant_mismatch",
121 + "The payment challenge merchant does not match the URL.",
122 + )
123 + if network != self.policy.network:
124 + return self._deny(
125 + "network_not_allowed",
126 + "The payment network is outside the configured policy.",
127 + )
128 + if currency != self.policy.currency:
129 + return self._deny(
130 + "currency_not_allowed",
131 + "The payment currency is outside the configured policy.",
132 + )
133 + if expires_at is not None and now >= expires_at:
134 + return self._deny(
135 + "challenge_expired",
136 + "The merchant payment challenge has expired.",
137 + )
138 + if amount > self.policy.per_request_limit:
139 + return self._deny(
140 + "request_limit_exceeded",
141 + "The amount exceeds the per-request spending limit.",
142 + )
143 + new_spend = (
144 + Decimal(0)
145 + if request.idempotency_key in self._spent_by_idempotency
146 + else amount
147 + )
148 + if self.spent + new_spend > self.policy.per_run_limit:
149 + return self._deny(
150 + "run_limit_exceeded",
151 + "The amount exceeds the remaining run budget.",
152 + )
153 +
154 + requires_approval = amount > self.policy.approval_threshold
155 + if requires_approval:
156 + approval_error = self._validate_approval(
157 + request,
158 + approval,
159 + amount=amount,
160 + now=now,
161 + )
162 + if approval_error is not None:
163 + return approval_error
164 +
165 + return AuthorizationDecision(
166 + allowed=True,
167 + code="authorized",
168 + reason="The request satisfies application-owned commerce policy.",
169 + requires_human_approval=requires_approval,
170 + approved_amount=amount,
171 + )
172 +
173 + def record_spend(self, idempotency_key: str, amount: Decimal) -> None:
174 + """Record one authorized amount without inventing receipt fields."""
175 +
176 + self._spent_by_idempotency.setdefault(idempotency_key, amount)
177 +
178 + def record(self, receipt: PaymentReceipt) -> None:
179 + self.record_spend(receipt.idempotency_key, receipt.amount)
180 +
181 + def _validate_approval(
182 + self,
183 + request: PurchaseRequest,
184 + approval: ApprovalGrant | None,
185 + *,
186 + amount: Decimal,
187 + now: datetime,
188 + ) -> AuthorizationDecision | None:
189 + if approval is None:
190 + return self._deny(
191 + "human_approval_required",
192 + "A human approval grant is required above the threshold.",
193 + requires_human_approval=True,
194 + )
195 + if now >= approval.expires_at:
196 + return self._deny(
197 + "approval_expired",
198 + "The human approval grant has expired.",
199 + requires_human_approval=True,
200 + )
201 + if (
202 + approval.request_id != request.request_id
203 + or approval.resource_url != request.resource_url
204 + or approval.purpose != request.purpose
205 + ):
206 + return self._deny(
207 + "approval_scope_mismatch",
208 + "The human approval grant is not bound to this request.",
209 + requires_human_approval=True,
210 + )
211 + if approval.currency != self.policy.currency:
212 + return self._deny(
213 + "approval_currency_mismatch",
214 + "The human approval grant uses a different currency.",
215 + requires_human_approval=True,
216 + )
217 + if amount > approval.maximum_amount:
218 + return self._deny(
219 + "approval_amount_exceeded",
220 + "The amount exceeds the human-approved maximum.",
221 + requires_human_approval=True,
222 + )
223 + return None
224 +
225 + @staticmethod
226 + def _deny(
227 + code: str,
228 + reason: str,
229 + *,
230 + requires_human_approval: bool = False,
231 + ) -> AuthorizationDecision:
232 + return AuthorizationDecision(
233 + allowed=False,
234 + code=code,
235 + reason=reason,
236 + requires_human_approval=requires_human_approval,
237 + )
examples/agentic-commerce-x402/src/omnipay_x402/tool.py new
+87
@@ -0,0 +1,87 @@
1 +"""Application-bound x402_fetch callable (no OpenAI Agents SDK required)."""
2 +
3 +from __future__ import annotations
4 +
5 +import json
6 +from collections.abc import Callable
7 +from dataclasses import dataclass
8 +from datetime import datetime
9 +
10 +from .application import CommerceApplication
11 +from .errors import CommerceError
12 +from .models import (
13 + AgentPurchaseEvidence,
14 + ApprovalGrant,
15 + PurchaseRequest,
16 + PurchaseResult,
17 +)
18 +
19 +
20 +@dataclass(frozen=True)
21 +class X402FetchTool:
22 + """Named tool wrapper that mimics an Agents SDK FunctionTool surface."""
23 +
24 + name: str
25 + description: str
26 + _fn: Callable[[str, str], str]
27 +
28 + def __call__(self, resource_url: str, purpose: str) -> str:
29 + return self._fn(resource_url, purpose)
30 +
31 +
32 +def build_x402_fetch_tool(
33 + application: CommerceApplication,
34 + *,
35 + request_id: str,
36 + idempotency_key: str,
37 + approval: ApprovalGrant | None = None,
38 + on_purchase: Callable[[PurchaseResult], None] | None = None,
39 + now: datetime | None = None,
40 +) -> X402FetchTool:
41 + """Build a tool whose authority is supplied by the application.
42 +
43 + The scripted agent can propose a URL and purpose. It cannot mint an
44 + approval, choose its own spending policy, or access proof/wallet material.
45 + """
46 +
47 + def x402_fetch(resource_url: str, purpose: str) -> str:
48 + try:
49 + result = application.purchase(
50 + PurchaseRequest(
51 + request_id=request_id,
52 + resource_url=resource_url,
53 + purpose=purpose,
54 + idempotency_key=idempotency_key,
55 + ),
56 + approval=approval,
57 + now=now,
58 + )
59 + if on_purchase is not None:
60 + on_purchase(result)
61 + evidence = AgentPurchaseEvidence(
62 + status=result.status,
63 + report=result.report,
64 + receipt_id=result.receipt.receipt_id,
65 + amount=result.receipt.amount,
66 + currency=result.receipt.currency,
67 + requires_human_approval=(result.authorization.requires_human_approval),
68 + )
69 + return evidence.model_dump_json()
70 + except CommerceError as exc:
71 + return json.dumps(
72 + {
73 + "status": "denied",
74 + "code": exc.code,
75 + "message": str(exc),
76 + },
77 + sort_keys=True,
78 + )
79 +
80 + return X402FetchTool(
81 + name="x402_fetch",
82 + description=(
83 + "Fetch one paid HTTPS resource through application-controlled "
84 + "merchant, purpose, spending, approval, and audit policy."
85 + ),
86 + _fn=x402_fetch,
87 + )
examples/agentic-commerce-x402/tests/test_local_flow.py new
+217
@@ -0,0 +1,217 @@
1 +"""Local x402 happy path, denial, and fabricated-receipt rejection tests."""
2 +
3 +from __future__ import annotations
4 +
5 +from datetime import UTC, datetime, timedelta
6 +from decimal import Decimal
7 +
8 +import pytest
9 +
10 +from omnipay_x402.agent import (
11 + PurchaseResultRecorder,
12 + SupplierResearchOutput,
13 + run_supplier_research,
14 + validate_supplier_research_output,
15 +)
16 +from omnipay_x402.application import CommerceApplication
17 +from omnipay_x402.aws_stub import live_agentcore_status
18 +from omnipay_x402.codec import encode_model
19 +from omnipay_x402.errors import AgentResultInvalid, PolicyDenied
20 +from omnipay_x402.merchant import (
21 + PAYMENT_REQUIRED_HEADER,
22 + PAYMENT_SIGNATURE_HEADER,
23 + RESOURCE_URL,
24 + SyntheticMerchant,
25 +)
26 +from omnipay_x402.models import (
27 + ApprovalGrant,
28 + AuditEventType,
29 + CommercePolicy,
30 + PurchaseRequest,
31 +)
32 +from omnipay_x402.payments import LocalPaymentProcessor
33 +from omnipay_x402.policy import PolicyEngine
34 +from omnipay_x402.tool import build_x402_fetch_tool
35 +from omnipay_x402 import VALUE_TRANSFERRED
36 +
37 +NOW = datetime(2026, 7, 30, 16, 0, tzinfo=UTC)
38 +
39 +
40 +def make_system(
41 + *,
42 + price: Decimal = Decimal("0.25"),
43 + request_limit: Decimal = Decimal("0.50"),
44 + run_limit: Decimal = Decimal("1.00"),
45 + threshold: Decimal = Decimal("0.10"),
46 + session_expires_at: datetime | None = None,
47 +) -> tuple[CommerceApplication, SyntheticMerchant, LocalPaymentProcessor]:
48 + payments = LocalPaymentProcessor()
49 + merchant = SyntheticMerchant(payments, price=price, now=NOW)
50 + policy = PolicyEngine(
51 + CommercePolicy(
52 + allowed_merchants=frozenset({"merchant.invalid"}),
53 + allowed_purposes=frozenset({"supplier_due_diligence"}),
54 + per_request_limit=request_limit,
55 + per_run_limit=run_limit,
56 + approval_threshold=threshold,
57 + session_expires_at=(session_expires_at or NOW + timedelta(minutes=15)),
58 + )
59 + )
60 + app = CommerceApplication(
61 + client=merchant.client(),
62 + policy=policy,
63 + payments=payments,
64 + )
65 + return app, merchant, payments
66 +
67 +
68 +def request(
69 + *,
70 + request_id: str = "request-001",
71 + resource_url: str = RESOURCE_URL,
72 + purpose: str = "supplier_due_diligence",
73 + idempotency_key: str = "purchase-001",
74 +) -> PurchaseRequest:
75 + return PurchaseRequest(
76 + request_id=request_id,
77 + resource_url=resource_url,
78 + purpose=purpose,
79 + idempotency_key=idempotency_key,
80 + )
81 +
82 +
83 +def approval(
84 + purchase: PurchaseRequest,
85 + *,
86 + maximum_amount: Decimal = Decimal("0.25"),
87 + expires_at: datetime | None = None,
88 +) -> ApprovalGrant:
89 + return ApprovalGrant(
90 + approval_id=f"approval-{purchase.request_id}",
91 + request_id=purchase.request_id,
92 + resource_url=purchase.resource_url,
93 + purpose=purchase.purpose,
94 + maximum_amount=maximum_amount,
95 + approved_by="synthetic-reviewer",
96 + approved_at=NOW,
97 + expires_at=expires_at or NOW + timedelta(minutes=10),
98 + )
99 +
100 +
101 +def test_value_transferred_is_always_false() -> None:
102 + assert VALUE_TRANSFERRED is False
103 + status = live_agentcore_status()
104 + assert status["status"] == "SKIPPED"
105 + assert status["reason"] == "NOT_READY"
106 + assert status["value_transferred"] is False
107 + assert status["network_calls"] is False
108 +
109 +
110 +def test_approved_purchase_completes_full_402_sequence() -> None:
111 + """Happy path: 402 → approve → synthetic proof → receipt/audit."""
112 +
113 + app, merchant, payments = make_system()
114 + purchase = request()
115 +
116 + result = app.purchase(purchase, approval=approval(purchase), now=NOW)
117 +
118 + assert result.status == "completed"
119 + assert result.receipt.amount == Decimal("0.25")
120 + assert result.authorization.requires_human_approval is True
121 + assert result.report.report_id == "SYNTH-SUPPLIER-RISK-001"
122 + assert merchant.request_count == 2
123 + assert merchant.fulfilled_count == 1
124 + assert payments.charge_count == 1
125 + assert [event.event_type for event in result.audit_events] == [
126 + AuditEventType.RESOURCE_REQUESTED,
127 + AuditEventType.PAYMENT_REQUIRED,
128 + AuditEventType.AUTHORIZATION_CHECKED,
129 + AuditEventType.PAYMENT_ATTEMPTED,
130 + AuditEventType.PROOF_CREATED,
131 + AuditEventType.MERCHANT_RETRY,
132 + AuditEventType.CONTENT_RETURNED,
133 + ]
134 +
135 +
136 +def test_missing_human_approval_is_denied_before_payment() -> None:
137 + """Denial without approval: policy blocks before synthetic payment."""
138 +
139 + app, merchant, payments = make_system()
140 +
141 + with pytest.raises(PolicyDenied) as exc_info:
142 + app.purchase(request(), now=NOW)
143 +
144 + assert exc_info.value.code == "human_approval_required"
145 + assert merchant.request_count == 1
146 + assert payments.charge_count == 0
147 +
148 +
149 +def test_application_validation_rejects_fabricated_receipt() -> None:
150 + """Fabricated receipt rejection: agent output must match app evidence."""
151 +
152 + app, _, _ = make_system()
153 + purchase = request()
154 + completed = app.purchase(purchase, approval=approval(purchase), now=NOW)
155 + fabricated = SupplierResearchOutput(
156 + status=completed.status,
157 + report_id=completed.report.report_id,
158 + supplier=completed.report.supplier,
159 + signals=completed.report.signals,
160 + disclaimer=completed.report.disclaimer,
161 + receipt_id="receipt-fabricated",
162 + amount=str(completed.receipt.amount),
163 + currency=completed.receipt.currency,
164 + requires_human_approval=completed.authorization.requires_human_approval,
165 + )
166 +
167 + with pytest.raises(AgentResultInvalid) as exc_info:
168 + validate_supplier_research_output(
169 + fabricated,
170 + PurchaseResultRecorder(results=[completed]),
171 + )
172 +
173 + assert exc_info.value.code == "agent_output_mismatch"
174 + assert "receipt_id" in str(exc_info.value)
175 +
176 +
177 +def test_scripted_agent_run_matches_application_evidence() -> None:
178 + app, merchant, payments = make_system()
179 + purchase = request(request_id="request-agent-001", idempotency_key="purchase-agent-001")
180 +
181 + result = run_supplier_research(
182 + app,
183 + request_id=purchase.request_id,
184 + idempotency_key=purchase.idempotency_key,
185 + approval=approval(purchase),
186 + now=NOW,
187 + )
188 +
189 + assert result.output.receipt_id == result.purchase.receipt.receipt_id
190 + assert merchant.request_count == 2
191 + assert payments.charge_count == 1
192 +
193 +
194 +def test_merchant_rejects_unrecognized_proof() -> None:
195 + _, merchant, payments = make_system()
196 + response = merchant.client().get(
197 + RESOURCE_URL,
198 + headers={PAYMENT_SIGNATURE_HEADER: encode_model(merchant.payment_required)},
199 + )
200 +
201 + assert response.status_code == 402
202 + assert PAYMENT_REQUIRED_HEADER in response.headers
203 + assert merchant.invalid_proof_count == 1
204 + assert payments.charge_count == 0
205 +
206 +
207 +def test_agents_tool_has_prebound_authority_and_expected_name() -> None:
208 + app, _, _ = make_system()
209 + purchase = request()
210 + tool = build_x402_fetch_tool(
211 + app,
212 + request_id=purchase.request_id,
213 + idempotency_key=purchase.idempotency_key,
214 + approval=approval(purchase),
215 + )
216 +
217 + assert tool.name == "x402_fetch"
guides/agentic-commerce-x402.mdx new
+120
@@ -0,0 +1,120 @@
1 +---
2 +title: "Agentic Commerce with x402 (Local Simulation)"
3 +sidebarTitle: "Agentic Commerce x402"
4 +description: "Map the OpenAI cookbook controlled agentic commerce flow onto ProcureNet sessions, USDC settlement, and AI agents — using a local-only x402 simulation with no live funds."
5 +---
6 +
7 +ProcureNet already orchestrates buyer sessions, USDC settlement, and background AI agents. This guide shows how an **agentic commerce** pattern fits those concepts when an agent needs to call a **paid API** that speaks the [x402](https://github.com/coinbase/x402) `HTTP 402 Payment Required` challenge.
8 +
9 +The runnable example lives at [`examples/agentic-commerce-x402/`](https://github.com/mindtdilly/Omnipay/tree/main/examples/agentic-commerce-x402). It is adapted from the OpenAI Cookbook partner example *Controlled agentic commerce with AgentCore Payments*.
10 +
11 +<Warning>
12 + **Local simulation only.** This example never moves real funds. There are no live payment gates, no AWS credentials, no Amazon Bedrock AgentCore calls, and no wallet private keys. Synthetic payment proofs only. Live AgentCore Payments is intentionally out of scope for this guide (`value_transferred=false`).
13 +</Warning>
14 +
15 +## What you will learn
16 +
17 +- How four authority boundaries cooperate: agent, application policy, payment session, paid API
18 +- How that maps to ProcureNet [session lifecycle](/concepts/session-lifecycle), [USDC settlement](/concepts/payment-settlement), [orchestration](/concepts/orchestration-modes), and [AI agents](/configuration/ai-agents)
19 +- How to run the local pytest suite without network payment side effects
20 +
21 +## Architecture in four responsibilities
22 +
23 +| Layer | Owns | Does not own |
24 +|---|---|---|
25 +| **Agent** | Tool selection and a typed proposal | Approval, budget, wallet access, settlement truth |
26 +| **Application policy** | Merchant allowlist, purpose, spending limits, human approval, idempotency, receipts, audit | Signing credentials |
27 +| **Payment session (simulated)** | Synthetic proof generation inside a bounded local adapter | Business authorization |
28 +| **Paid API (merchant)** | `402` challenge and fulfillment when a valid proof is presented | Deciding whether your app may spend |
29 +
30 +In the Omnipay package these map to:
31 +
32 +1. `omnipay_x402.agent` — scripted one-tool loop (no live model SDK required)
33 +2. `omnipay_x402.policy` + `CommerceApplication` — application-owned controls
34 +3. `LocalPaymentProcessor` — synthetic proofs; `aws_stub.live_agentcore_status()` returns `SKIPPED` / `NOT_READY`
35 +4. `SyntheticMerchant` — in-memory HTTP transport that returns `402` then `200`
36 +
37 +```mermaid
38 +sequenceDiagram
39 + participant Agent
40 + participant App as Application policy
41 + participant Pay as Local payment session
42 + participant API as Paid API
43 +
44 + Agent->>App: Propose x402_fetch(url, purpose)
45 + App->>API: GET resource
46 + API-->>App: 402 PAYMENT-REQUIRED
47 + App->>App: Authorize (merchant, budget, approval)
48 + App->>Pay: Authorize synthetic proof
49 + Pay-->>App: Proof + receipt
50 + App->>API: GET resource + PAYMENT-SIGNATURE
51 + API-->>App: 200 + report + PAYMENT-RESPONSE
52 + App-->>Agent: Typed evidence (no wallet material)
53 +```
54 +
55 +## How x402 works in the sim
56 +
57 +1. The merchant returns **HTTP 402** with a machine-readable `PAYMENT-REQUIRED` header.
58 +2. The application treats that challenge as a **proposal**, validates it against policy, and optionally requires a human `ApprovalGrant`.
59 +3. `LocalPaymentProcessor` creates a **synthetic** proof bound to an idempotency key.
60 +4. The application retries with `PAYMENT-SIGNATURE`. The merchant verifies the proof and returns the paid content plus `PAYMENT-RESPONSE`.
61 +5. An append-only **audit trail** records the sequence for later review.
62 +
63 +Nothing in this path touches a chain, a wallet, or an external payment provider.
64 +
65 +## Map onto ProcureNet concepts
66 +
67 +<CardGroup cols={2}>
68 + <Card title="Session lifecycle" icon="arrows-spin" href="/concepts/session-lifecycle">
69 + Treat the commerce `session_expires_at` and approval windows like ProcureNet
70 + session TTLs: expire closed, never silently extend spending authority.
71 + </Card>
72 + <Card title="Payment settlement" icon="coins" href="/concepts/payment-settlement">
73 + Production ProcureNet settles USDC via
74 + [`usdc-request`](/api/usdc-request),
75 + [`usdc-watch`](/api/usdc-watch), and
76 + [`usdc-manual-confirm`](/api/usdc-manual-confirm).
77 + The example only **shapes** the same currency/network vocabulary locally.
78 + </Card>
79 + <Card title="Orchestration modes" icon="diagram-project" href="/concepts/orchestration-modes">
80 + Application policy—not the model—owns merchant allowlists and budgets,
81 + analogous to how orchestration modes choose transport and prefill depth.
82 + </Card>
83 + <Card title="AI agents" icon="robot" href="/configuration/ai-agents">
84 + Like OpenClaw workers, the economic tool is application-bound. Agents propose;
85 + they do not mint approvals or hold settlement truth.
86 + </Card>
87 +</CardGroup>
88 +
89 +<Tip>
90 + When you later wire a real paid supplier API into ProcureNet, keep the same
91 + split: session + policy in your backend, settlement via the USDC APIs, and
92 + agents limited to proposing tool calls against pre-bound capabilities.
93 +</Tip>
94 +
95 +## Run the local example
96 +
97 +```bash
98 +cd examples/agentic-commerce-x402
99 +uv sync
100 +uv run pytest
101 +```
102 +
103 +Expected coverage includes:
104 +
105 +- **Happy path** — approved purchase completes the full `402 → proof → 200` sequence with audit events
106 +- **Denial without approval** — amounts above the threshold fail with `human_approval_required` before any synthetic charge
107 +- **Fabricated receipt rejection** — agent output that invents a `receipt_id` fails closed against application evidence
108 +
109 +Confirm the safety invariants:
110 +
111 +```python
112 +from omnipay_x402 import VALUE_TRANSFERRED, live_agentcore_status
113 +
114 +assert VALUE_TRANSFERRED is False
115 +assert live_agentcore_status()["status"] == "SKIPPED"
116 +```
117 +
118 +## Attribution
119 +
120 +Adapted from the OpenAI Cookbook example *Controlled agentic commerce with AgentCore Payments* (authors: Deepak Jain and Sid Rampally). See [`examples/agentic-commerce-x402/SOURCE.md`](https://github.com/mindtdilly/Omnipay/blob/main/examples/agentic-commerce-x402/SOURCE.md) for the upstream URL and intentional omissions (all live AgentCore / AWS modules).