# -*- coding: utf-8 -*- """ KKOS Python SDK — zero dependencies, standard library only. from kkos_client import KKOS kk = KKOS(api_key="kk_live_xxxxxxxx") model = kk.run_text_session([ "I tend to decide fast and say what I think.", "When someone hesitates I usually push for a decision.", "I'd rather have a blunt answer than a polite one.", ]) for t in model["traits"]: print(t["name"], t["score"], t["confidence"]) Design notes ------------ * No third-party imports. `urllib` only, so the file can be dropped into any Python 3.8+ project without adding a requirement. * Auth is either an API key (`X-API-Key: kk_live_...`) or a user session token (`Authorization: Bearer ...`) obtained through /api/v1/auth/*. * Every method raises `KKOSError` carrying the HTTP status, the parsed error payload and the request URL, so callers can distinguish 401 from 429 from 5xx. API reference: https://kkmatch.com/sdk/ """ from __future__ import annotations import json import time import urllib.error import urllib.parse import urllib.request __version__ = "1.0.0" __all__ = ["KKOS", "KKOSError", "DEFAULT_BASE_URL"] DEFAULT_BASE_URL = "https://kkmatch.com" _API_PREFIX = "/api/v1" class KKOSError(Exception): """Raised for any non-2xx response, and for 2xx responses that are not JSON.""" def __init__(self, status, payload, url, method): self.status = status self.payload = payload self.url = url self.method = method detail = payload if isinstance(payload, dict): detail = payload.get("detail") or payload.get("error") or payload self.message = "[%s %s] %s -> %s" % (method, status, url, detail) super().__init__(self.message) @property def hint(self): """Remediation hint supplied by the API (mirrors `err.hint` in the Web SDK).""" return self.payload.get("hint", "") if isinstance(self.payload, dict) else "" class KKOS: """Minimal, dependency-free client for the KKOS / Human Mirror HTTP API.""" def __init__(self, api_key=None, token=None, base_url=DEFAULT_BASE_URL, timeout=90, user_agent=None): self.api_key = api_key self.token = token self.base_url = (base_url or DEFAULT_BASE_URL).rstrip("/") self.timeout = timeout self.user_agent = user_agent or ("kkos-python/%s" % __version__) # ------------------------------------------------------------------ utils def _headers(self, extra=None): h = {"Accept": "application/json", "User-Agent": self.user_agent} if self.api_key: h["X-API-Key"] = self.api_key if self.token: h["Authorization"] = "Bearer %s" % self.token if extra: h.update(extra) return h def _request(self, method, path, body=None, query=None, raw=False): """Core transport. Returns parsed JSON (dict/list) or raw bytes.""" url = self.base_url + path if query: clean = {k: v for k, v in query.items() if v not in (None, "")} if clean: url += "?" + urllib.parse.urlencode(clean) data = None headers = self._headers() if body is not None: data = json.dumps(body).encode("utf-8") headers["Content-Type"] = "application/json" req = urllib.request.Request(url, data=data, headers=headers, method=method) try: with urllib.request.urlopen(req, timeout=self.timeout) as resp: payload = resp.read() if raw: return payload ctype = resp.headers.get("Content-Type", "") if "application/json" not in ctype: # Guard: a base_url pointing at the site root instead of the API can # get a 200 + HTML page back from the web server's SPA fallback # (measured: 2,503,285 B). Without this the HTML is returned as data. raise KKOSError(resp.status, { "detail": "expected JSON but received %s (%d bytes)" % (ctype or "unknown content-type", len(payload)), "hint": "base_url probably does not point at the API. /health and " "/api/v1/* are served by the same origin as the site " "(e.g. https://kkmatch.com).", }, url, method) return json.loads(payload.decode("utf-8")) if payload else None except urllib.error.HTTPError as e: blob = e.read() try: parsed = json.loads(blob.decode("utf-8")) except Exception: parsed = blob.decode("utf-8", "replace")[:400] raise KKOSError(e.code, parsed, url, method) from None except urllib.error.URLError as e: raise KKOSError(0, {"detail": "network error: %s" % e.reason}, url, method) from None def _get(self, path, **q): return self._request("GET", path, query=q or None) def _post(self, path, body=None): return self._request("POST", path, body=body if body is not None else {}) # --------------------------------------------------------------- platform def health(self): """Liveness probe. `GET /health` — public.""" return self._get("/health") def plans(self): """Subscription plans. `GET /api/v1/plans` — public.""" return self._get(_API_PREFIX + "/plans") def me(self): """Current identity for the supplied credential. `GET /api/v1/auth/me`.""" return self._get(_API_PREFIX + "/auth/me") # ------------------------------------------------- mirror: human modelling def start(self, mode="text", consent=None, user_id="", **kwargs): """Open a modelling session. `POST /api/v1/mirror/start` mode: 'text' | 'voice' | 'voice_only' | 'video' consent keys: camera, audio, analysis, storage, research, share Pass `all_consent=True` to accept every flag (handy for tests / SDK users who have already collected consent in their own UI). """ all_consent = kwargs.pop("all_consent", False) if all_consent: c = dict(consent or {}) for k in ("camera", "audio", "analysis", "storage", "research", "share"): c.setdefault(k, True) consent = c body = {"mode": mode, "user_id": user_id, "consent": consent or {}} body.update(kwargs) return self._post(_API_PREFIX + "/mirror/start", body) def message(self, session_id, message, input_mode=None, **kwargs): """Send one user turn. `POST /api/v1/mirror/message`""" body = {"session_id": session_id, "message": message} if input_mode: body["input_mode"] = input_mode body.update(kwargs) return self._post(_API_PREFIX + "/mirror/message", body) def complete(self, session_id, transcript=None, **kwargs): """Finish the session and build the model. `POST /api/v1/mirror/complete` Returns the raw envelope: {"session_id": ..., "model_id": ..., "redirect": "/mirror-result.html?model=...", "model": {"model_id": ..., "overall_confidence": 0.6, "traits": [...11...], ...}} The model itself lives under the "model" key — use `complete_model()` to get it unwrapped. """ body = {"session_id": session_id} if transcript is not None: body["transcript"] = transcript body.update(kwargs) return self._post(_API_PREFIX + "/mirror/complete", body) def complete_model(self, session_id, transcript=None, **kwargs): """Same as `complete()` but returns the unwrapped model dict.""" resp = self.complete(session_id, transcript=transcript, **kwargs) if isinstance(resp, dict) and isinstance(resp.get("model"), dict): m = resp["model"] m.setdefault("session_id", resp.get("session_id")) return m return resp def traits(self, model): """Normalise the traits list out of either a model dict or an envelope.""" if isinstance(model, dict): if isinstance(model.get("traits"), list): return model["traits"] inner = model.get("model") if isinstance(inner, dict) and isinstance(inner.get("traits"), list): return inner["traits"] return [] def get_model(self, model_id): """Fetch a built model. `GET /api/v1/mirror/model/{model_id}`""" return self._get(_API_PREFIX + "/mirror/model/" + urllib.parse.quote(model_id)) def delete_model(self, model_id): """Delete a model and its derived data. `DELETE /api/v1/mirror/model/{model_id}`""" return self._request("DELETE", _API_PREFIX + "/mirror/model/" + urllib.parse.quote(model_id)) def regenerate_model(self, model_id): """Rebuild a model from the same session. `POST /api/v1/mirror/model/{model_id}/regenerate`""" return self._post(_API_PREFIX + "/mirror/model/" + urllib.parse.quote(model_id) + "/regenerate") def compare(self, model_a, model_b): """Compare two models. `POST /api/v1/mirror/compare`""" return self._post(_API_PREFIX + "/mirror/compare", {"model_a": model_a, "model_b": model_b}) # ------------------------------------------------------------------ voice def voices(self): """Available voices. `GET /api/v1/mirror/voices`""" return self._get(_API_PREFIX + "/mirror/voices") def tts(self, text, provider="auto", voice="", edge_voice="", rate=0, session_id=""): """Synthesise speech. `POST /api/v1/mirror/tts` — returns raw audio bytes. provider: 'auto' (Hume first, edge fallback) | 'hume' | 'edge'. For provider='edge' pass `edge_voice='en-US-AriaNeural'`. """ body = {"text": text, "provider": provider, "voice": voice, "edge_voice": edge_voice, "rate": rate, "session_id": session_id} return self._request("POST", _API_PREFIX + "/mirror/tts", body, raw=True) def tts_to_file(self, path, text, **kwargs): """Same as `tts()` but writes the audio to `path`. Returns the byte count.""" blob = self.tts(text, **kwargs) with open(path, "wb") as fh: fh.write(blob) return len(blob) def stt(self, audio_bytes, filename="audio.webm"): """Transcribe audio. `POST /api/v1/mirror/stt` — multipart upload. Kept dependency-free by hand-rolling the multipart body. """ boundary = "----kkos%d" % int(time.time() * 1000) head = ('--%s\r\nContent-Disposition: form-data; name="file"; filename="%s"\r\n' 'Content-Type: audio/webm\r\n\r\n' % (boundary, filename)).encode("utf-8") tail = ("\r\n--%s--\r\n" % boundary).encode("utf-8") payload = head + audio_bytes + tail headers = self._headers({"Content-Type": "multipart/form-data; boundary=%s" % boundary}) req = urllib.request.Request(self.base_url + _API_PREFIX + "/mirror/stt", data=payload, headers=headers, method="POST") try: with urllib.request.urlopen(req, timeout=self.timeout) as resp: raw = resp.read() try: return json.loads(raw.decode("utf-8")) except Exception: return {"text": raw.decode("utf-8", "replace")} except urllib.error.HTTPError as e: blob = e.read() try: parsed = json.loads(blob.decode("utf-8")) except Exception: parsed = blob.decode("utf-8", "replace")[:400] raise KKOSError(e.code, parsed, req.full_url, "POST") from None # ------------------------------------------------------------- convenience def run_text_session(self, lines, mode="text", consent=None, on_reply=None): """Start -> send each line -> complete. Returns the finished *model*. `on_reply(turn_index, reply_text)` is called as replies arrive, which is what the example script uses to stream progress to a terminal. """ sess = self.start(mode=mode, consent=consent, all_consent=(consent is None)) sid = sess.get("session_id") or sess.get("id") if not sid: raise KKOSError(0, {"detail": "no session_id in start response", "raw": sess}, self.base_url + "/mirror/start", "POST") replies = [] transcript = [] for i, line in enumerate(lines): transcript.append({"role": "user", "content": line}) r = self.message(sid, line, input_mode="text") text = r.get("reply") or r.get("message") or r.get("text") or "" if text: transcript.append({"role": "assistant", "content": text}) replies.append(text) if on_reply: on_reply(i, text) model = self.complete_model(sid, transcript=transcript) if isinstance(model, dict): model.setdefault("_session_id", sid) model.setdefault("_replies", replies) return model