""" HumanOps — verified human execution for AI agents, in Europe. GENERATED from https://quiescence.eu/humanops/api/v1/openapi.json on 2026-09-08. Do not edit by hand: regenerate it, or the next capability we add will be missing from your copy. Standard library only. Python 3.9+. from humanops import HumanOps, HumanOpsError ho = HumanOps(api_key="ho_test_...") task = ho.create_task(body={ "capability": "human.verify", "spec": {"claim": "This address is a working office", "subject": "Acme SL"}, "location": {"country_iso": "ES", "city_slug": "barcelona"}, "max_price_cents": 20000, }) print(ho.wait_for_result(task["ref"])["answer"]) """ from __future__ import annotations import json import time import urllib.error import urllib.parse import urllib.request import uuid __all__ = ["HumanOps", "HumanOpsError", "ERROR_CODES"] DEFAULT_BASE = "https://quiescence.eu/humanops/api/v1" #: Every code the API can return in error.code. Branch on these, never on the #: message — the message is written for a person and may be reworded. ERROR_CODES = ("unauthorized", "key_revoked", "agent_suspended", "forbidden_scope", "rate_limited", "quota_exceeded", "invalid_request", "schema_violation", "capability_unknown", "capability_refused", "coverage_unavailable", "sla_unavailable", "insufficient_credit", "policy_denied", "plan_required", "task_not_found", "task_not_cancellable", "bad_reason", "not_disputable", "already_disputed", "idempotency_conflict", "method_not_allowed", "server_error") class HumanOpsError(RuntimeError): """An API error. `code` is the stable machine string; `detail` says what to fix.""" def __init__(self, code: str, message: str, status: int, detail: dict | None = None): super().__init__(f"{code}: {message}") self.code = code self.message = message self.status = status self.detail = detail or {} class HumanOps: def __init__(self, api_key: str, base_url: str = DEFAULT_BASE, timeout: float = 30.0, client: str = "humanops-python"): self.api_key = api_key self.base_url = base_url.rstrip("/") self.timeout = timeout self.client = client #: Set from the X-RateLimit headers on every response, so a caller can pace #: itself instead of discovering the limit by hitting it. self.rate_limit_remaining: int | None = None self.rate_limit_reset: int | None = None @property def sandbox(self) -> bool: """Sandbox is a property of the KEY. There is no second base URL.""" return self.api_key.startswith("ho_test_") # -- transport ------------------------------------------------------------ def _request(self, verb: str, path: str, body: dict | None = None, query: dict | None = None, idempotent: bool = False) -> dict: url = self.base_url + path if query: clean = {k: v for k, v in query.items() if v is not None} if clean: url += "?" + urllib.parse.urlencode(clean) headers = { "Accept": "application/json", "X-HumanOps-Client": self.client, } if self.api_key: headers["Authorization"] = "Bearer " + self.api_key if idempotent: # REQUIRED on create. A retry after a timeout must not order twice, and # the caller who most needs the guard is the one who would forget it. headers.setdefault("Idempotency-Key", str(uuid.uuid4())) data = None if body is not None: data = json.dumps(body).encode() headers["Content-Type"] = "application/json" req = urllib.request.Request(url, data=data, headers=headers, method=verb) try: with urllib.request.urlopen(req, timeout=self.timeout) as res: payload = json.loads(res.read().decode() or "{}") self._note_limits(res.headers) except urllib.error.HTTPError as e: raw = e.read().decode() self._note_limits(e.headers) try: payload = json.loads(raw) except ValueError: raise HumanOpsError("server_error", raw[:200] or e.reason, e.code) from None err = payload.get("error", {}) raise HumanOpsError(err.get("code", "server_error"), err.get("message", "Request failed"), e.code, err.get("detail")) from None if not payload.get("ok", False): err = payload.get("error", {}) raise HumanOpsError(err.get("code", "server_error"), err.get("message", "Request failed"), 200, err.get("detail")) return payload.get("data", {}) def _note_limits(self, headers) -> None: try: self.rate_limit_remaining = int(headers.get("X-RateLimit-Remaining")) self.rate_limit_reset = int(headers.get("X-RateLimit-Reset")) except (TypeError, ValueError): pass # -- operations ----------------------------------------------------------- def list_capabilities(self) -> dict: """Every capability, with its JSON Schemas, prices and live geographies""" return self._request("GET", f"/capabilities") def get_capability(self, key) -> dict: """getCapability""" return self._request("GET", f"/capabilities/{key}") def get_coverage(self, **query) -> dict: """getCoverage""" return self._request("GET", f"/coverage", query=query) def quote_task(self, body: dict | None = None) -> dict: """An exact price, or a refusal with alternatives""" return self._request("POST", f"/quote", body=body) def list_tasks(self, **query) -> dict: """listTasks""" return self._request("GET", f"/tasks", query=query) def create_task(self, body: dict | None = None) -> dict: """Order a task""" return self._request("POST", f"/tasks", body=body, idempotent=True) def get_task(self, ref) -> dict: """getTask""" return self._request("GET", f"/tasks/{ref}") def get_task_result(self, ref) -> dict: """getTaskResult""" return self._request("GET", f"/tasks/{ref}/result") def list_task_evidence(self, ref) -> dict: """listTaskEvidence""" return self._request("GET", f"/tasks/{ref}/evidence") def cancel_task(self, ref) -> dict: """cancelTask""" return self._request("POST", f"/tasks/{ref}/cancel") def dispute_task(self, ref, body: dict | None = None) -> dict: """Say the delivered answer is wrong""" return self._request("POST", f"/tasks/{ref}/dispute", body=body) # -- the loop everyone writes by hand, once ------------------------------- def wait_for_result(self, ref: str, timeout: float = 3600.0, on_status=None) -> dict: """Poll until the answer is ready, obeying poll_after_s. poll_after_s is not a suggestion: before it elapses nothing can have changed, so an earlier read spends rate limit to learn what you knew. For a live task, prefer a webhook and do not call this at all. """ deadline = time.monotonic() + timeout last = None while True: snap = self.get_task_result(ref) if on_status and snap.get("status") != last: last = snap.get("status") on_status(last) if snap.get("ready"): return snap if time.monotonic() >= deadline: raise TimeoutError(f"{ref} was still {snap.get('status')} after {timeout}s") time.sleep(max(1, int(snap.get("poll_after_s") or 30)))