"""Hunch read-only client. The operator must enforce results before signing.""" import json import urllib.error import urllib.parse import urllib.request class _NoRedirect(urllib.request.HTTPRedirectHandler): def redirect_request(self, req, fp, code, msg, headers, newurl): return None class HunchError(RuntimeError): pass class HunchClient: def __init__(self, base_url="https://hunchagents.xyz", profile="conservative", timeout=22): url = urllib.parse.urlparse(base_url) if url.scheme != "https" and not (url.scheme == "http" and url.hostname in ("localhost", "127.0.0.1")): raise ValueError("Use HTTPS or a loopback development server.") if profile not in ("conservative", "standard"): raise ValueError("Unknown policy profile.") self.base_url, self.profile, self.timeout = base_url.rstrip("/"), profile, timeout def _post(self, path, payload): request = urllib.request.Request(self.base_url + path, json.dumps(payload).encode(), {"Content-Type": "application/json", "Accept": "application/json", "User-Agent": "HunchPython/0.2"}) try: with urllib.request.build_opener(_NoRedirect).open(request, timeout=self.timeout) as response: return json.load(response) except (urllib.error.URLError, ValueError, TimeoutError) as error: raise HunchError("Hunch could not complete the check. Stop and request review.") from error def inspect_token(self, address): return self._post("/api/rh/intelligence", {"address": address}) def get_decision_context(self, address, tokens="1", counter_asset=None, transaction=None): payload = {"address": address, "tokens": tokens} if counter_asset is not None: payload["counterAsset"] = counter_asset if transaction is not None: payload["transaction"] = transaction return self._post("/api/rh/context", payload) def inspect_instrument(self, address, tokens="1"): return self._post("/api/rh/instrument", {"address": address, "tokens": tokens}) def inspect_text(self, text): return self._post("/api/text", {"text": text}) def assess_action(self, address, amount_usd, slippage_bps, text=None): payload = {"address": address, "amountUsd": amount_usd, "slippageBps": slippage_bps, "profile": self.profile} if text is not None: payload["text"] = text return self._post("/api/rh/assess", payload) def simulate_transaction(self, transaction): return self._post("/api/rh/simulate", {"transaction": transaction}) def require_within_profile(self, **action): report = self.assess_action(**action) if report.get("decision") != "within_profile": raise HunchError("Action requires review or is blocked: " + "; ".join(report.get("reasons", []))) return report