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