"""Small x402 client for egg-verify (https://egg-verify.onrender.com). egg-verify sells three pay-per-call research services over x402 v2, paid in USDC on Base (eip155:8453): POST /verify $0.05 claim + URL -> SUPPORTED / REFUTED / UNCLEAR with quotes POST /scout $0.50 brief + up to 3 URLs -> extractive summary and quotes POST /deepdive $2.00 brief + up to 8 URLs -> cross-source synthesis Flow: POST -> HTTP 402 with a PAYMENT-REQUIRED header -> the x402 SDK signs an EIP-3009 USDC authorization for exactly that amount -> retry with PAYMENT-SIGNATURE -> HTTP 200 + PAYMENT-RESPONSE (settlement tx hash). egg-verify settles only after a successful response. The private key comes from AGENT_PRIVATE_KEY (or is passed in) and is only used locally to sign. Every payment is capped by max_usdc_per_call, which defaults to $0.05 so only /verify can be paid unless you raise it on purpose. """ from __future__ import annotations import base64 import json import os from dataclasses import dataclass, field from decimal import Decimal from typing import Any import requests BASE_URL = "https://egg-verify.onrender.com" BASE_MAINNET = "eip155:8453" USDC_DECIMALS = 6 DEFAULT_MAX_USDC_PER_CALL = "0.05" USER_AGENT = "langchain-egg-verify/0.1.0 (+https://scrapemole.pages.dev)" PRICES_USDC = {"verify": "0.05", "scout": "0.50", "deepdive": "2.00"} class EggVerifyError(RuntimeError): """Raised when egg-verify returns an error or a payment is refused.""" @dataclass class PaidResult: """JSON body of a paid call plus its x402 settlement proof.""" url: str status: int body: dict[str, Any] settlement: dict[str, Any] = field(default_factory=dict) @property def transaction(self) -> str | None: return self.settlement.get("transaction") @property def basescan_url(self) -> str | None: tx = self.transaction return f"https://basescan.org/tx/{tx}" if tx else None def _usdc_to_atomic(usdc: str | float | Decimal) -> int: return int(Decimal(str(usdc)) * (10**USDC_DECIMALS)) class EggVerifyClient: """Pays egg-verify endpoints with x402 from a local EVM key. Args: private_key: hex key of a dedicated wallet holding a little USDC on Base. Defaults to AGENT_PRIVATE_KEY. Never use a main wallet. max_usdc_per_call: hard cap per payment; a higher quote is refused before anything is signed. """ def __init__( self, private_key: str | None = None, max_usdc_per_call: str | float = DEFAULT_MAX_USDC_PER_CALL, base_url: str = BASE_URL, timeout: float = 90.0, ) -> None: self.base_url = base_url.rstrip("/") self.timeout = timeout self.max_atomic = _usdc_to_atomic(max_usdc_per_call) self._private_key = private_key or os.environ.get("AGENT_PRIVATE_KEY") self._paying_session: requests.Session | None = None self._free = requests.Session() self._free.headers["User-Agent"] = USER_AGENT # -- free ----------------------------------------------------------- def service_card(self) -> dict[str, Any]: """Free: the service card with endpoints, prices and payTo.""" r = self._free.get(self.base_url + "/", timeout=self.timeout) r.raise_for_status() return r.json() def health(self) -> bool: r = self._free.get(self.base_url + "/health", timeout=self.timeout) return r.status_code == 200 def quote(self, endpoint: str, body: dict[str, Any] | None = None) -> dict[str, Any]: """Free: read the unpaid 402 for an endpoint. Returns the Base USDC payment option (amount in atomic units, payTo, asset). Nothing is signed.""" r = self._free.post( f"{self.base_url}/{endpoint.lstrip('/')}", json=body or {}, timeout=self.timeout ) if r.status_code != 402: raise EggVerifyError(f"expected HTTP 402 quote, got {r.status_code}") header = r.headers.get("payment-required") if not header: raise EggVerifyError("402 without a PAYMENT-REQUIRED header") req = json.loads(base64.b64decode(header)) base = [a for a in req.get("accepts", []) if a.get("network") == BASE_MAINNET] if not base: raise EggVerifyError("no USDC-on-Base payment option offered") return min(base, key=lambda a: int(a["amount"])) # -- paid ----------------------------------------------------------- def _session(self) -> requests.Session: if self._paying_session is None: if not self._private_key: raise EggVerifyError( "Set AGENT_PRIVATE_KEY to a dedicated Base wallet holding a little " "USDC (or pass private_key=...)." ) from eth_account import Account from x402 import x402ClientSync from x402.client_base import max_amount from x402.http.clients import wrapRequestsWithPayment from x402.mechanisms.evm.exact.register import register_exact_evm_client x402 = x402ClientSync() register_exact_evm_client( x402, Account.from_key(self._private_key), networks=BASE_MAINNET, policies=[max_amount(self.max_atomic)], ) x402.set_spend_controls( {"max_amount_per_payment": f"${Decimal(self.max_atomic) / 10**USDC_DECIMALS}"} ) session = wrapRequestsWithPayment(requests.Session(), x402) session.headers["User-Agent"] = USER_AGENT self._paying_session = session return self._paying_session def post_paid(self, endpoint: str, body: dict[str, Any]) -> PaidResult: """POST a paid endpoint, paying the 402 with x402 (capped).""" price = _usdc_to_atomic(PRICES_USDC.get(endpoint.strip("/"), "0")) if price > self.max_atomic: raise EggVerifyError( f"/{endpoint} costs ${PRICES_USDC[endpoint.strip('/')]}, above the " f"per-call cap; raise max_usdc_per_call to allow it" ) url = f"{self.base_url}/{endpoint.lstrip('/')}" r = self._session().post(url, json=body, timeout=self.timeout) settlement: dict[str, Any] = {} header = r.headers.get("payment-response") if header: from x402.http.utils import decode_payment_response_header settled = decode_payment_response_header(header) settlement = ( settled.model_dump(by_alias=True, exclude_none=True) if hasattr(settled, "model_dump") else dict(settled) ) try: data = r.json() except ValueError: data = {"raw": r.text[:2000]} if r.status_code != 200: raise EggVerifyError(f"HTTP {r.status_code} from {url}: {str(data)[:500]}") return PaidResult(url=url, status=r.status_code, body=data, settlement=settlement) def verify(self, claim: str, url: str) -> PaidResult: """$0.05: check a factual claim against one public URL.""" return self.post_paid("verify", {"claim": claim[:500], "url": url}) def scout(self, brief: str, urls: list[str]) -> PaidResult: """$0.50: extractive summary of up to 3 URLs against a brief.""" return self.post_paid("scout", {"brief": brief[:500], "urls": urls[:3]}) def deepdive(self, brief: str, urls: list[str]) -> PaidResult: """$2.00: cross-source synthesis of up to 8 URLs.""" return self.post_paid("deepdive", {"brief": brief[:500], "urls": urls[:8]})