#!/usr/bin/env python3 """crafthouse — CLI and Python SDK for Craft House Cloud (https://api.crafthouse.ai). Install: curl -fsSL https://api.crafthouse.ai/dl/crafthouse.py -o ~/.local/bin/crafthouse && chmod +x ~/.local/bin/crafthouse Login: crafthouse login (asks for an API key; or: crafthouse register you@example.com) Use: crafthouse catalog | order n8n-start --plan trial | status [ID] | creds ID | ssh ID | renew ID | cancel ID | balance | topup 50 Add --json for machine-readable output. The key is stored in ~/.config/crafthouse/key (mode 600) or taken from CRAFTHOUSE_API_KEY. SDK: from crafthouse import Client; c = Client(api_key); c.order("n8n-start", plan="trial") """ import argparse, getpass, json, os, subprocess, sys, time, urllib.error, urllib.request API = os.environ.get("CRAFTHOUSE_API", "https://api.crafthouse.ai/v1") KEYFILE = os.path.expanduser("~/.config/crafthouse/key") class ApiError(Exception): def __init__(self, status, message): super().__init__(f"{status}: {message}") self.status, self.message = status, message class Client: def __init__(self, api_key=None, base=API): self.key, self.base = api_key, base.rstrip("/") def _req(self, method, path, body=None): h = {"Content-Type": "application/json", "Accept": "application/json", "User-Agent": "crafthouse-cli/1.0"} if self.key: h["Authorization"] = "Bearer " + self.key req = urllib.request.Request(self.base + path, data=json.dumps(body).encode() if body is not None else None, method=method, headers=h) try: with urllib.request.urlopen(req, timeout=60) as r: return json.loads(r.read() or b"{}") except urllib.error.HTTPError as e: try: msg = json.loads(e.read()).get("error", "") except Exception: msg = e.reason raise ApiError(e.code, msg) from None def register(self, email, name=None): return self._req("POST", "/register", {"email": email, "name": name}) def catalog(self): return self._req("GET", "/catalog") def order(self, product=None, plan="monthly", settings=None, bundle=None): return self._req("POST", "/orders", {"product": product, "bundle": bundle, "plan": plan, "settings": settings or {}}) def services(self): return self._req("GET", "/services")["services"] def service(self, sid): return self._req("GET", f"/services/{int(sid)}") def credentials(self, sid): return self._req("GET", f"/services/{int(sid)}/credentials") def renew(self, sid): return self._req("POST", f"/services/{int(sid)}/renew") def cancel(self, sid, type="end_of_period", reason=None): return self._req("POST", f"/services/{int(sid)}/cancel", {"type": type, "reason": reason}) def balance(self): return self._req("GET", "/balance") def topup(self, amount): return self._req("POST", "/balance/topup", {"amount": amount}) def new_key(self, name="agent", webhook_url=None): return self._req("POST", "/keys", {"name": name, "webhook_url": webhook_url}) def wait_ready(self, sid, timeout=900, every=10): """Poll until the service state is ready (or failed); returns the last status.""" end = time.time() + timeout while True: s = self.service(sid) if s.get("state") in ("ready", "failed") or time.time() > end: return s time.sleep(every) def _key(): k = os.environ.get("CRAFTHOUSE_API_KEY") if k: return k try: return open(KEYFILE).read().strip() except OSError: sys.exit("No API key: run `crafthouse login` or `crafthouse register EMAIL`, or set CRAFTHOUSE_API_KEY.") def _save(k): os.makedirs(os.path.dirname(KEYFILE), exist_ok=True) fd = os.open(KEYFILE, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) with os.fdopen(fd, "w") as f: f.write(k) def _out(data, as_json, text=None): if as_json or text is None: print(json.dumps(data, indent=2, ensure_ascii=False)) else: print(text(data)) def main(): p = argparse.ArgumentParser(prog="crafthouse", description="Craft House Cloud CLI") p.add_argument("--json", action="store_true", help="JSON output") sp = p.add_subparsers(dest="cmd", required=True) sp.add_parser("login", help="save an API key") r = sp.add_parser("register", help="create an account and save its API key"); r.add_argument("email"); r.add_argument("--name") sp.add_parser("catalog", help="services, bundles, plans and prices") o = sp.add_parser("order", help="order a product or a bundle"); o.add_argument("product", help="product slug, or bundle id with --bundle") o.add_argument("--plan", default="monthly", help="trial | monthly | 3m | 6m | 12m"); o.add_argument("--bundle", action="store_true") o.add_argument("--set", action="append", default=[], metavar="KEY=VALUE", help="order setting (repeat)"); o.add_argument("--wait", action="store_true", help="wait until ready") st = sp.add_parser("status", help="all services or one"); st.add_argument("id", nargs="?") for c in ("creds", "ssh", "renew"): sp.add_parser(c).add_argument("id") cn = sp.add_parser("cancel"); cn.add_argument("id"); cn.add_argument("--now", action="store_true") sp.add_parser("balance") t = sp.add_parser("topup"); t.add_argument("amount", type=float) a = p.parse_args() try: if a.cmd == "login": k = getpass.getpass("API key (chk_…): ").strip() Client(k).balance(); _save(k); print("Saved to " + KEYFILE); return if a.cmd == "register": d = Client().register(a.email, a.name); _save(d["api_key"]) _out(d, a.json, lambda d: f"Account {d['email']} created. Password (shown once): {d['password']}\nAPI key saved to {KEYFILE}. Log in: {d['login_url']}"); return if a.cmd == "catalog": _out(Client().catalog(), a.json, lambda d: "\n".join([f"{x['product']:<22} {x['name'][:40]:<40} " + ", ".join(f"{p['plan']} ${p['price_usd']}" for p in x["plans"]) + ("" if x["available"] else " (sold out)") for x in d["products"]] + [""] + [f"bundle {b['bundle']:<22} ${b['price_usd_month']}/mo (instead of ${b['separately_usd_month']}): " + ", ".join(b["products"]) for b in d["bundles"]])); return c = Client(_key()) if a.cmd == "order": st = dict(x.split("=", 1) for x in a.set) d = c.order(bundle=a.product, plan=a.plan, settings=st) if a.bundle else c.order(a.product, a.plan, st) if a.wait and not a.bundle and d.get("status") == "provisioning": d = c.wait_ready(d["service_id"]) _out(d, a.json, lambda d: json.dumps(d, indent=2, ensure_ascii=False)) elif a.cmd == "status": if a.id: _out(c.service(a.id), a.json, lambda s: f"#{s['service_id']} {s['product']} {s['status']} state={s.get('state')} url={s.get('url') or '-'}") else: _out(c.services(), a.json, lambda l: "\n".join(f"#{s['service_id']:<6} {s['product']:<20} {s['status']:<10} ${s['price_usd']}" for s in l) or "No services") elif a.cmd == "creds": _out(c.credentials(a.id), a.json, lambda d: "\n".join(f"{k}: {v}" for k, v in d.items() if v)) elif a.cmd == "ssh": acc = c.credentials(a.id).get("access") or "" if not acc.startswith("ssh "): sys.exit("This service has no SSH access (only VPS has it).") os.execvp("ssh", acc.split()) elif a.cmd == "renew": _out(c.renew(a.id), a.json, lambda d: "Paid from balance." if d["paid"] else f"Not enough balance. Pay here: {d['invoice']['pay_url']}") elif a.cmd == "cancel": _out(c.cancel(a.id, "immediate" if a.now else "end_of_period"), a.json, lambda d: f"Cancellation requested: {d['cancellation']}") elif a.cmd == "balance": _out(c.balance(), a.json, lambda d: f"Balance: ${d['balance_usd']}; pending invoices: {len(d['pending_invoices'])}") elif a.cmd == "topup": _out(c.topup(a.amount), a.json, lambda d: f"Invoice #{d['invoice_id']} for ${d['total_usd']}. Pay: {d['pay_url']}") except ApiError as e: sys.exit(f"Error {e.status}: {e.message}") if __name__ == "__main__": main()