"""Lists a LAX offer on the registry, paying the 402 through a caller supplied function.

CLI: LAX_PAYER_KEY=0x... python list_offer.py   (reads lax-offer.json; the key is never printed)
"""
from __future__ import annotations

import base64
import json
import os
import sys
from pathlib import Path
from typing import Callable

import httpx

DEFAULT_REGISTRY = "https://lax.kognai.ai/registry"


class ListOfferError(Exception):
    def __init__(self, message: str, status: int):
        super().__init__(message)
        self.status = status


def read_payment_required(res: httpx.Response) -> dict:
    h = res.headers.get("payment-required")
    if h:
        try:
            v = json.loads(base64.b64decode(h).decode())
            if isinstance(v.get("accepts"), list):
                return v
        except Exception:
            pass
    try:
        body = res.json()
    except ValueError:
        body = {}
    if isinstance(body, dict) and isinstance(body.get("accepts"), list):
        return body
    raise ListOfferError("the registry answered 402 without payment terms", 402)


def list_offer(offer: dict, pay: Callable[[dict], str], registry_url: str = DEFAULT_REGISTRY,
               client: httpx.Client | None = None) -> dict:
    url = registry_url.rstrip("/") + "/offers"
    c = client or httpx.Client(timeout=60)
    headers = {"content-type": "application/json", "accept": "application/json"}
    body = json.dumps(offer)
    res = c.post(url, content=body, headers=headers)
    if res.status_code == 402:
        terms = read_payment_required(res)
        value = pay(terms)
        if not isinstance(value, str) or not value:
            raise ListOfferError("pay() must return the payment header value", 402)
        name = "PAYMENT-SIGNATURE" if int(terms.get("x402Version", 1)) >= 2 else "X-PAYMENT"
        res = c.post(url, content=body, headers={**headers, name: value})
    try:
        out = res.json()
    except ValueError:
        out = {}
    if not 200 <= res.status_code < 300:
        raise ListOfferError(out.get("error") if isinstance(out.get("error"), str) else f"the registry answered HTTP {res.status_code}", res.status_code)
    if not isinstance(out.get("id"), str) or not out["id"]:
        raise ListOfferError("the registry answer has no listing id", res.status_code)
    return {"id": out["id"]}


def main() -> int:  # pragma: no cover
    from eth_account import Account
    from evm_pay import exact_payment_header
    key = os.environ.get("LAX_PAYER_KEY")
    if not key:
        sys.stderr.write("Set LAX_PAYER_KEY in the environment.\n")
        return 1
    account = Account.from_key(key)
    offer = json.loads((Path(__file__).parent / "lax-offer.json").read_text())
    cap = int(os.environ.get("MAX_FEE_ATOMIC", "10000"))
    out = list_offer(offer, lambda t: exact_payment_header(t, account, max_atomic=cap))
    print(f"listed: {out['id']}")
    return 0


if __name__ == "__main__":  # pragma: no cover
    sys.exit(main())

agentgate-source: code mirror on Cloudflare Pages. All repos