|
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 |