#!/usr/bin/env python3 """MediaAtlas decision models from the command line (Python 3, standard library only). python3 decide.py packs the packs and use cases (no key needed) python3 decide.py decide core routing "text to decide" one decision; or --file FILE, --image FILE python3 decide.py usage plan, tokens used and left, agent budget python3 decide.py agent-key "my agent" --budget 5000000 a key for an agent under your plan key (shown once) python3 decide.py agent-keys your agent keys; agent-revoke PREFIX removes one python3 decide.py buy developer|team a Stripe link for a person to pay The key is read from DECISION_API_KEY (starts with ma_). It is never printed. Exit codes: 0 done, 1 error, 2 payment needed (402), 3 wait and retry (429/503). """ import argparse import base64 import json import mimetypes import os import sys import urllib.error import urllib.request BASE = os.environ.get("DECISION_URL", "https://decision.mediaatlas.si").rstrip("/") PRICING = "https://decision.mediaatlas.si/#pricing" def key() -> str: k = os.environ.get("DECISION_API_KEY", "").strip() if not k: sys.exit(f"No key: set DECISION_API_KEY (an ma_... key). To get one, a person pays once: {PRICING}") return k def call(method: str, path: str, body: dict | None = None, auth: bool = True) -> tuple[int, dict, dict]: headers = {"User-Agent": "decision-skill/1", "Content-Type": "application/json"} if auth: headers["Authorization"] = f"Bearer {key()}" req = urllib.request.Request(BASE + path, json.dumps(body).encode() if body is not None else None, headers, method=method) try: with urllib.request.urlopen(req, timeout=120) as r: return r.status, json.loads(r.read() or b"{}"), dict(r.headers) except urllib.error.HTTPError as e: try: data = json.loads(e.read()) except ValueError: data = {} return e.code, data, dict(e.headers) def fail(status: int, body: dict, headers: dict) -> None: err = body.get("error") or {} msg = err.get("message") if isinstance(err, dict) else err code = err.get("code") if isinstance(err, dict) else "" print(f"Error {status} {code}: {msg}", file=sys.stderr) if status == 402: print(f"The plan needs paying. A person can pay here: {PRICING}\nOr make a link: python3 {sys.argv[0]} buy developer", file=sys.stderr) sys.exit(2) if status in (429, 503): print(f"Wait {headers.get('Retry-After', '10')} s and try again (quota_exceeded lasts until the 1st).", file=sys.stderr) sys.exit(3) sys.exit(1) def show(path: str, auth: bool = True) -> None: status, body, headers = call("GET", path, auth=auth) if status != 200: fail(status, body, headers) print(json.dumps(body, indent=2, ensure_ascii=False)) def packs(_: argparse.Namespace) -> None: status, body, headers = call("GET", "/v1/packs", auth=False) if status != 200: fail(status, body, headers) for p in body["packs"]: print(f"{p['id']}: {p['title']}") for u in p["usecases"]: print(f" {u['id']}: {u['title']}") def decide(a: argparse.Namespace) -> None: body: dict = {"pack": a.pack, "usecase": a.usecase} if a.model: body["model"] = a.model text = open(a.file, encoding="utf-8", errors="replace").read() if a.file else a.text if text: body["input"] = text if a.image: mime = mimetypes.guess_type(a.image)[0] or "image/jpeg" body["images"] = [f"data:{mime};base64," + base64.b64encode(open(a.image, "rb").read()).decode()] body.setdefault("input", text or "") if "input" not in body: sys.exit("Give a text, --file or --image.") status, res, headers = call("POST", "/v1/decisions", body) if status != 200: fail(status, res, headers) for r in res["results"]: if "ranking" in r: print("ranking: " + ", ".join(f"{x['id']} {x['score']:.2f}" for x in r["ranking"][:10])) continue d = r.get("decision") or {} print(f"{d.get('option')} confidence {d.get('confidence')} band {d.get('band')}") for qid, q in (r.get("answers") or {}).items(): probs = q.get("probabilities") or {} print(f" {qid}: " + ", ".join(f"{o} {p:.2f}" for o, p in sorted(probs.items(), key=lambda i: -i[1])[:5])) print(f"tokens {res['usage']['input_tokens']}, {res['latency_ms']} ms", file=sys.stderr) def agent_key(a: argparse.Namespace) -> None: body: dict = {"name": a.name} if a.budget: body["monthly_tokens"] = a.budget if a.packs: body["packs"] = a.packs status, res, headers = call("POST", "/v1/keys", body) if status != 201: fail(status, res, headers) print(f"Agent key for {a.name} (shown once; give it to one agent only):\n{res['key']}") def agent_keys(_: argparse.Namespace) -> None: show("/v1/keys") def agent_revoke(a: argparse.Namespace) -> None: status, res, headers = call("DELETE", f"/v1/keys/{a.prefix}") if status != 200: fail(status, res, headers) print("revoked", a.prefix) def buy(a: argparse.Namespace) -> None: status, res, headers = call("POST", "/api/checkout", {"plan": a.plan}, auth=False) if status != 200: fail(status, res, headers) print(f"Ask the person to open this link and pay for the {a.plan} plan:\n{res['url']}\nAfter paying they see an API key once; they send it to you (DECISION_API_KEY).") def main() -> None: p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) sub = p.add_subparsers(dest="cmd", required=True) sub.add_parser("packs").set_defaults(fn=packs) sub.add_parser("usage").set_defaults(fn=lambda a: show("/v1/usage")) d = sub.add_parser("decide") d.add_argument("pack") d.add_argument("usecase") d.add_argument("text", nargs="?") d.add_argument("--file") d.add_argument("--image") d.add_argument("--model", choices=["fast"]) d.set_defaults(fn=decide) k = sub.add_parser("agent-key") k.add_argument("name") k.add_argument("--budget", type=int, help="monthly input tokens this agent may spend") k.add_argument("--packs", nargs="*") k.set_defaults(fn=agent_key) sub.add_parser("agent-keys").set_defaults(fn=agent_keys) r = sub.add_parser("agent-revoke") r.add_argument("prefix") r.set_defaults(fn=agent_revoke) b = sub.add_parser("buy") b.add_argument("plan", choices=["developer", "team"]) b.set_defaults(fn=buy) a = p.parse_args() a.fn(a) if __name__ == "__main__": main()