“””
ucls_client.py — Unified client script that talks to every UCLS service
NaaS, AaaS, RegaaS, IDaaS, ValaaS, PRaaS, GaaS, OaaS, TaaS, RaaS,
TLaaS, ProvAaaS, VaaS, InteropaaS, ACaaS, DaaS, AnaaS, VizaaS.
Set:
export UCLS_BASE_URL=”https://ucls.example/api”
export UCLS_API_KEY=”your-token”
Install:
pip install requests
Run demo (touches every service once):
python ucls_client.py
“””
from future import annotations
import os, time, uuid, json, typing as t
import requests
DEFAULT_TIMEOUT = 20 # seconds
MAX_RETRIES = 4
RETRY_BACKOFF_BASE = 0.4 # seconds
class UCLSHTTPError(RuntimeError):
pass
class BaseAPI:
def init(self, base_url: str, api_key: str, session: requests.Session | None = None):
self.base_url = base_url.rstrip(“/”)
self.api_key = api_key
self.s = session or requests.Session()
def _headers(self, extra: dict[str, str] | None = None) -> dict[str, str]:
h = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json",
"Accept": "application/json",
"X-Idempotency-Key": str(uuid.uuid4()),
}
if extra:
h.update(extra)
return h
def _request(
self,
method: str,
path: str,
*,
params: dict | None = None,
json_body: dict | None = None,
timeout: int = DEFAULT_TIMEOUT,
) -> dict:
url = f"{self.base_url}{path}"
last_exc: Exception | None = None
for attempt in range(1, MAX_RETRIES + 1):
try:
resp = self.s.request(
method,
url,
params=params,
json=json_body,
headers=self._headers(),
timeout=timeout,
)
if 200 <= resp.status_code < 300:
if resp.content and "application/json" in resp.headers.get("Content-Type",""):
return resp.json()
return {}
# Retry on 429/5xx
if resp.status_code in (429, 500, 502, 503, 504):
delay = RETRY_BACKOFF_BASE * (2 ** (attempt - 1))
time.sleep(delay)
continue
# Hard error
raise UCLSHTTPError(f"{resp.status_code} {resp.text}")
except (requests.Timeout, requests.ConnectionError) as e:
last_exc = e
delay = RETRY_BACKOFF_BASE * (2 ** (attempt - 1))
time.sleep(delay)
continue
raise UCLSHTTPError(f"Failed after {MAX_RETRIES} attempts for {method} {path}. Last error: {last_exc}")———- Service Clients ———-
class NaaS(BaseAPI):
def normalize_name(self, value: str, code_hint: str | None = None):
return self._request(“POST”, “/ucls/names”, json_body={“value”: value, “code_hint”: code_hint})
def parse_binomial(self, value: str, code_hint: str | None = None):
return self._request("POST", "/ucls/names/parse", json_body={"value": value, "code_hint": code_hint})
def variants(self, value: str):
return self._request("POST", "/ucls/names/variants", json_body={"value": value})class AaaS(BaseAPI):
def bind_anchor(self, token_id: str, anchor: dict):
return self._request(“POST”, “/ucls/anchors”, json_body={“token_id”: token_id, “anchor”: anchor})
def specimen_match(self, token: str, collection_query: dict):
return self._request("POST", "/ucls/anchors/specimen-match", json_body={"token": token, "query": collection_query})
def sequence_exemplar(self, token: str, sequences: list[dict]):
return self._request("POST", "/ucls/anchors/sequence-exemplar", json_body={"token": token, "sequences": sequences})class RegaaS(BaseAPI):
def register(self, token_id: str, registry: str, payload: dict):
return self._request(“POST”, “/ucls/registrations”, json_body={“token_id”: token_id, “registry”: registry, “payload”: payload})
def resolve(self, registry: str, key: str):
return self._request("GET", "/ucls/registrations/resolve", params={"registry": registry, "key": key})
def reconcile(self, records: list[dict]):
return self._request("POST", "/ucls/registrations/reconcile", json_body={"records": records})class IDaaS(BaseAPI):
def mint(self, entity: dict, namespace: str):
return self._request(“POST”, “/ucls/ids/mint”, json_body={“entity”: entity, “namespace”: namespace})
def resolve(self, pid: str):
return self._request("GET", "/ucls/ids/resolve", params={"id": pid})
def synonyms(self, pairs: list[list[str]]):
return self._request("POST", "/ucls/ids/synonyms", json_body={"pairs": pairs})class ValaaS(BaseAPI):
def validate(self, token_id: str, code: str, evidence: dict):
return self._request(“POST”, “/ucls/validations”, json_body={“token_id”: token_id, “code”: code, “evidence”: evidence})
def shacl(self, graph: dict, shapes: dict):
return self._request("POST", "/ucls/validations/shacl", json_body={"graph": graph, "shapes": shapes})
def cross_code(self, token_id: str, codes: list[str]):
return self._request("POST", "/ucls/validations/cross-code", json_body={"token_id": token_id, "codes": codes})class PRaaS(BaseAPI):
def decide_priority(self, conflicts: list[dict], code: str):
return self._request(“POST”, “/ucls/priority/decide”, json_body={“conflicts”: conflicts, “code”: code})
def conservation_score(self, proposal: dict, usage_stats: dict):
return self._request("POST", "/ucls/priority/conservation-score", json_body={"proposal": proposal, "usage": usage_stats})class GaaS(BaseAPI):
def submit_opinion(self, proposal: dict, evidence: dict):
return self._request(“POST”, “/ucls/governance/opinions”, json_body={“proposal”: proposal, “evidence”: evidence})
def ballot(self, opinion_id: str, ballots: list[dict]):
return self._request("POST", "/ucls/governance/ballots", json_body={"opinion_id": opinion_id, "ballots": ballots})class OaaS(BaseAPI):
def ontology_diff(self, old: dict, current: dict, proposed: dict):
return self._request(“POST”, “/ucls/ontology/diff”, json_body={“old”: old, “current”: current, “proposed”: proposed})
def reasoner_check(self, ontology_graph: dict):
return self._request("POST", "/ucls/ontology/reasoner/check", json_body={"graph": ontology_graph})
def map_suggest(self, source: dict, target: dict):
return self._request("POST", "/ucls/ontology/map", json_body={"source": source, "target": target})class TaaS(BaseAPI):
def reconcile_trees(self, trees: list[dict], evidence: dict | None = None):
return self._request(“POST”, “/ucls/taxonomy/trees/reconcile”, json_body={“trees”: trees, “evidence”: evidence or {}})
def normalize_ranks(self, code_ranks: dict):
return self._request("POST", "/ucls/taxonomy/ranks/normalize", json_body={"code_ranks": code_ranks})
def holotype_search(self, token: str, filters: dict | None = None):
return self._request("GET", "/ucls/taxonomy/holotype/search", params={"token": token, **(filters or {})})class RaaS(BaseAPI):
def record_linkage(self, records: list[dict]):
return self._request(“POST”, “/ucls/reconcile/records”, json_body={“records”: records})
def synonym_collapse(self, graph: dict):
return self._request("POST", "/ucls/reconcile/synonyms/collapse", json_body={"graph": graph})
def crosswalk(self, mappings: list[dict]):
return self._request("POST", "/ucls/reconcile/crosswalk", json_body={"mappings": mappings})class TLaaS(BaseAPI):
def transliterate(self, token: str, lang_from: str, lang_to: str):
return self._request(“POST”, “/ucls/translate/transliterate”, json_body={“token”: token, “from”: lang_from, “to”: lang_to})
def label_rank(self, labels: list[str], signals: dict | None = None):
return self._request("POST", "/ucls/translate/label-rank", json_body={"labels": labels, "signals": signals or {}})
def ipa(self, token: str, locale: str):
return self._request("POST", "/ucls/translate/ipa", json_body={"token": token, "locale": locale})class ProvAaaS(BaseAPI):
def chain(self, events: list[dict], signatures: list[dict]):
return self._request(“POST”, “/ucls/provenance/chain”, json_body={“events”: events, “signatures”: signatures})
def snapshot(self, state_graph: dict):
return self._request("POST", "/ucls/provenance/snapshot", json_body={"state": state_graph})class VaaS(BaseAPI):
def bump(self, diff_report: dict):
return self._request(“POST”, “/ucls/version/bump”, json_body={“diff”: diff_report})
def delta(self, old: dict, new: dict):
return self._request("POST", "/ucls/version/delta", json_body={"old": old, "new": new})
def rollback_plan(self, versions: list[str], deps: dict | None = None):
return self._request("POST", "/ucls/version/rollback/plan", json_body={"versions": versions, "deps": deps or {}})class InteropaaS(BaseAPI):
def map(self, external: dict, schema_map: dict):
return self._request(“POST”, “/ucls/interop/map”, json_body={“payload”: external, “map”: schema_map})
def rate_plan(self, policy: dict):
return self._request("POST", "/ucls/interop/rate-limit/plan", json_body={"policy": policy})
def detect_schema_version(self, payload: dict):
return self._request("POST", "/ucls/interop/detect-schema-version", json_body={"payload": payload})
# Logarithm-related interop:
def log_base_convert(self, value: float, from_base: float, to_base: float):
return self._request("POST", "/ucls/interop/log-base/convert", json_body={"value": value, "from_base": from_base, "to_base": to_base})
def alias_normalize(self, symbol: str):
return self._request("POST", "/ucls/interop/alias/normalize", json_body={"symbol": symbol})
def schema_flag_log(self, schema: dict):
return self._request("POST", "/ucls/interop/schema/flag-log", json_body={"schema": schema})class ACaaS(BaseAPI):
def evaluate(self, subject: dict, action: str, resource: dict, attrs: dict | None = None):
return self._request(“POST”, “/ucls/access/evaluate”, json_body={
“subject”: subject, “action”: action, “resource”: resource, “attrs”: attrs or {}
})
def issue_credential(self, claims: dict, policy: dict):
return self._request("POST", "/ucls/access/credential/issue", json_body={"claims": claims, "policy": policy})
def policy_lint(self, policy_diff: dict):
return self._request("POST", "/ucls/access/policy/lint", json_body={"diff": policy_diff})class DaaS(BaseAPI):
def index(self, records: list[dict]):
return self._request(“POST”, “/ucls/graph/index”, json_body={“records”: records})
def search(self, q: str, context: dict | None = None, k: int = 10):
return self._request("GET", "/ucls/graph/search", params={"q": q, "k": k, "context": json.dumps(context or {})})
def paths(self, graph: dict, source: str, target: str):
return self._request("POST", "/ucls/graph/paths", json_body={"graph": graph, "source": source, "target": target})
def optimize(self, query_plan: dict):
return self._request("POST", "/ucls/query/optimize", json_body={"plan": query_plan})class AnaaS(BaseAPI):
def entropy(self, probs: list[float], base: float = 2.0):
return self._request(“POST”, “/ucls/analytics/entropy”, json_body={“probs”: probs, “base”: base})
def anomaly(self, metrics: list[dict]):
return self._request("POST", "/ucls/analytics/anomaly", json_body={"metrics": metrics})
def forecast(self, series: list[float], horizon: int = 12):
return self._request("POST", "/ucls/analytics/forecast", json_body={"series": series, "horizon": horizon})
# Logarithm domain calculators:
def zipf(self, ranks: list[float], freqs: list[float]):
return self._request("POST", "/ucls/analytics/zipf", json_body={"ranks": ranks, "freqs": freqs})
def power_law(self, xs: list[float], xmin: float):
return self._request("POST", "/ucls/analytics/power-law", json_body={"x": xs, "xmin": xmin})
def pH(self, h_conc_molar: float):
return self._request("POST", "/ucls/analytics/pH", json_body={"H_conc": h_conc_molar})
def decibel(self, numerator: float, denominator: float, mode: str = "power"):
return self._request("POST", "/ucls/analytics/decibel", json_body={"num": numerator, "den": denominator, "mode": mode})
def magnitude(self, f2: float, f1: float):
return self._request("POST", "/ucls/analytics/magnitude", json_body={"F2": f2, "F1": f1})class VizaaS(BaseAPI):
def tree(self, graph: dict, layout: dict | None = None):
return self._request(“POST”, “/ucls/viz/tree”, json_body={“graph”: graph, “layout”: layout or {}})
def kg(self, graph: dict, query: dict | None = None):
return self._request("POST", "/ucls/viz/kg", json_body={"graph": graph, "query": query or {}})
def diff(self, old: dict, new: dict, diff_report: dict):
return self._request("POST", "/ucls/viz/diff", json_body={"old": old, "new": new, "diff": diff_report})
# Log visuals:
def log_axis(self, min_val: float, max_val: float, base: float):
return self._request("POST", "/ucls/viz/log-axis", json_body={"min": min_val, "max": max_val, "base": base})
def log_hist(self, xs: list[float], base: float, bins: int = 20):
return self._request("POST", "/ucls/viz/log-hist", json_body={"x": xs, "base": base, "bins": bins})
def bode(self, freqs: list[float], magnitudes: list[float]):
return self._request("POST", "/ucls/viz/bode", json_body={"freqs": freqs, "magnitudes": magnitudes})———- Aggregated Client ———-
class UCLSClient:
def init(self, base_url: str | None = None, api_key: str | None = None):
base_url = base_url or os.environ.get(“UCLS_BASE_URL”, “”).strip()
api_key = api_key or os.environ.get(“UCLS_API_KEY”, “”).strip()
if not base_url or not api_key:
raise RuntimeError(“Set UCLS_BASE_URL and UCLS_API_KEY environment variables.”)
session = requests.Session()
# Construct all service clients sharing a single HTTP session:
self.naas = NaaS(base_url, api_key, session)
self.aaas = AaaS(base_url, api_key, session)
self.regaas = RegaaS(base_url, api_key, session)
self.idaas = IDaaS(base_url, api_key, session)
self.valaas = ValaaS(base_url, api_key, session)
self.praas = PRaaS(base_url, api_key, session)
self.gaas = GaaS(base_url, api_key, session)
self.oaas = OaaS(base_url, api_key, session)
self.taas = TaaS(base_url, api_key, session)
self.raas = RaaS(base_url, api_key, session)
self.tlaas = TLaaS(base_url, api_key, session)
self.provaaas = ProvAaaS(base_url, api_key, session)
self.vaas = VaaS(base_url, api_key, session)
self.interop = InteropaaS(base_url, api_key, session)
self.acaas = ACaaS(base_url, api_key, session)
self.daas = DaaS(base_url, api_key, session)
self.anaas = AnaaS(base_url, api_key, session)
self.vizaas = VizaaS(base_url, api_key, session)———- Demo Pipeline touching every service once ———-
def demo_pipeline():
c = UCLSClient()
# 1) NaaS — normalize & parse
norm = c.naas.normalize_name("Panthera leo", code_hint="ICZN")
token_id = norm.get("token_id", "token:panthera-leo")
parsed = c.naas.parse_binomial("Panthera leo", code_hint="ICZN")
# 2) AaaS — anchor (specimen)
anchor = c.aaas.bind_anchor(token_id, {"kind": "specimen", "catalogNumber": "BMNH 12345", "institutionCode": "NHMUK"})
# 3) ValaaS — validate under a code
val = c.valaas.validate(token_id, code="ICZN", evidence={"publication": {"doi":"10.1234/abcd"}, "type": anchor})
# 4) RegaaS — register in an authority (ZooBank as example)
reg = c.regaas.register(token_id, "ZooBank", {"kind":"name", "authorship": parsed.get("authorship")})
# 5) IDaaS — mint PID
pid = c.idaas.mint({"token_id": token_id}, namespace="ucls:name")
# 6) PRaaS — priority decision (dummy conflict)
decision = c.praas.decide_priority(conflicts=[{"candidate": token_id, "date":"1758-01-01"}], code="ICZN")
# 7) GaaS — submit governance opinion
opinion = c.gaas.submit_opinion({"subject": token_id, "action":"conserve"}, {"evidence": [reg]})
# 8) OaaS — reasoner check on a tiny ontology
reason = c.oaas.reasoner_check({"@graph":[{"@id":"Taxon","@type":"owl:Class"}]})
# 9) TaaS — reconcile trees (toy)
trees = [{"tree":"(Panthera_leo,(Panthera_tigris,Panthera_onca));"}]
recon = c.taas.reconcile_trees(trees, evidence={"source":"literature"})
# 10) RaaS — link duplicate records
linked = c.raas.record_linkage([{"name":"Panthera leo"},{"name":"P. leo"}])
# 11) TLaaS — transliterate a vernacular label
translit = c.tlaas.transliterate("ле́в", "ru", "la")
# 12) ProvAaaS — provenance snapshot
snap = c.provaaas.snapshot({"token": token_id, "registry": reg})
# 13) VaaS — bump version after changes
bump = c.vaas.bump({"changed":["name.status","tree"]})
# 14) InteropaaS — convert logarithm base for an analytics consumer
base_conv = c.interop.log_base_convert(1.0, from_base=10.0, to_base=2.0)
# 15) ACaaS — check permission to register
authz = c.acaas.evaluate({"user":"ron"}, "register", {"registry":"ZooBank"}, {"role":"curator"})
# 16) DaaS — index and search
c.daas.index([{"id": token_id, "label":"Panthera leo"}])
search = c.daas.search("Panthera leo", k=3)
# 17) AnaaS — quick entropy calc (logarithm-based)
ent = c.anaas.entropy([0.5, 0.25, 0.25], base=2.0)
# 18) VizaaS — render a tiny hierarchy
tree_svg = c.vizaas.tree({"nodes":[{"id":"Panthera"},{"id":"Panthera leo"}], "edges":[["Panthera","Panthera leo"]]})
# Return a compact report
return {
"token_id": token_id,
"pid": pid,
"registration": reg,
"validation": val,
"priority": decision,
"opinion": opinion,
"search_top": search,
"entropy_bits": ent,
"viz_tree": tree_svg,
"authz": authz,
"log_base_convert": base_conv,
}if name == “main“:
try:
report = demo_pipeline()
print(json.dumps(report, indent=2))
except Exception as e:
print(f”[UCLS DEMO ERROR] {e}”)
raise
Key terms in plain language
Open a term for a concise explanation of language used on this page.
API
An application programming interface is a defined way for software systems to exchange data or request functions from one another.
Artificial Intelligence (AI)
Software designed to perform tasks involving prediction, classification, generation, reasoning, or decision support. Business use still requires clear data, governance, security, and human accountability.
Cloud Computing
Computing resources—such as applications, servers, storage, or databases—delivered from remote infrastructure and scaled as requirements change.
Cybersecurity
The practices and controls used to protect identities, devices, networks, applications, and data from unauthorized access, disruption, or manipulation.
Identity and Access Management (IAM)
The systems and policies that determine who a user is, what resources they may access, and how that access is authenticated and reviewed.
Bandwidth
The amount of data a connection can carry in a given time, usually measured in Mbps or Gbps. More bandwidth supports more users, devices, and simultaneous applications.