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