#!/usr/bin/env python3 """Minimal offline-friendly ClawHub CLI for SkillHub.""" from __future__ import annotations import argparse import getpass import http.cookiejar import json import mimetypes import os import shutil import sys import tempfile import uuid import zipfile from pathlib import Path from urllib import error, parse, request DEFAULT_REGISTRY = "http://127.0.0.1:8080" CONFIG_DIR = Path(os.environ.get("CLAWHUB_CONFIG_DIR", Path.home() / ".clawhub")) COOKIE_FILE = CONFIG_DIR / "cookies.txt" class CliError(Exception): pass def registry_url(value: str | None = None) -> str: raw = value or os.environ.get("CLAWHUB_REGISTRY") or os.environ.get("SKILLHUB_REGISTRY") or DEFAULT_REGISTRY return raw.rstrip("/") def read_json(path: Path) -> dict: try: return json.loads(path.read_text(encoding="utf-8")) except FileNotFoundError as exc: raise CliError(f"File not found: {path}") from exc except json.JSONDecodeError as exc: raise CliError(f"Invalid JSON in {path}: {exc}") from exc def response_payload(raw: bytes) -> object: if not raw: return None text = raw.decode("utf-8", errors="replace") try: return json.loads(text) except json.JSONDecodeError: return text def unwrap_data(payload: object) -> object: if isinstance(payload, dict) and "data" in payload: return payload["data"] return payload def print_json(payload: object) -> None: print(json.dumps(payload, ensure_ascii=False, indent=2)) def load_cookies() -> http.cookiejar.MozillaCookieJar: jar = http.cookiejar.MozillaCookieJar(str(COOKIE_FILE)) if COOKIE_FILE.exists(): try: jar.load(ignore_discard=True, ignore_expires=True) except http.cookiejar.LoadError: pass return jar def save_cookies(jar: http.cookiejar.MozillaCookieJar) -> None: CONFIG_DIR.mkdir(parents=True, exist_ok=True) jar.save(ignore_discard=True, ignore_expires=True) class Client: def __init__(self, base_url: str, token: str | None = None, use_cookies: bool = True): self.base_url = base_url.rstrip("/") self.token = token or os.environ.get("CLAWHUB_TOKEN") or os.environ.get("SKILLHUB_TOKEN") self.cookies = load_cookies() if use_cookies else http.cookiejar.MozillaCookieJar() self.opener = request.build_opener(request.HTTPCookieProcessor(self.cookies)) def request( self, method: str, path: str, *, data: bytes | None = None, headers: dict[str, str] | None = None, timeout: int = 120, ) -> object: url = f"{self.base_url}{path}" all_headers = {"User-Agent": "clawhub-offline-cli/0.1.0"} if headers: all_headers.update(headers) if self.token: all_headers["Authorization"] = f"Bearer {self.token}" req = request.Request(url, data=data, headers=all_headers, method=method) try: with self.opener.open(req, timeout=timeout) as resp: payload = response_payload(resp.read()) save_cookies(self.cookies) return payload except error.HTTPError as exc: body = exc.read().decode("utf-8", errors="replace") detail = body.strip() or exc.reason raise CliError(f"HTTP {exc.code} {method} {url}: {detail}") from exc except error.URLError as exc: raise CliError(f"Cannot reach {url}: {exc.reason}") from exc def make_package_zip(skill_dir: Path) -> Path: skill_dir = skill_dir.resolve() if not skill_dir.is_dir(): raise CliError(f"Skill directory not found: {skill_dir}") if not (skill_dir / "package.json").is_file(): raise CliError(f"Missing package.json in {skill_dir}") if not (skill_dir / "skill.md").is_file(): raise CliError(f"Missing skill.md in {skill_dir}") package = read_json(skill_dir / "package.json") if not package.get("name"): raise CliError(f"package.json must include name: {skill_dir / 'package.json'}") if not package.get("version"): raise CliError(f"package.json must include version: {skill_dir / 'package.json'}") temp_dir = Path(tempfile.mkdtemp(prefix="clawhub-publish-")) zip_path = temp_dir / f"{package['name']}-{package['version']}.zip" with zipfile.ZipFile(zip_path, "w", compression=zipfile.ZIP_DEFLATED) as zf: for path in sorted(skill_dir.rglob("*")): if not path.is_file(): continue if any(part in {".git", "__pycache__", "node_modules"} for part in path.relative_to(skill_dir).parts): continue zf.write(path, path.relative_to(skill_dir).as_posix()) return zip_path def multipart_form(fields: dict[str, str], files: dict[str, Path]) -> tuple[bytes, str]: boundary = f"----clawhub-{uuid.uuid4().hex}" chunks: list[bytes] = [] for name, value in fields.items(): chunks.extend( [ f"--{boundary}\r\n".encode(), f'Content-Disposition: form-data; name="{name}"\r\n\r\n'.encode(), str(value).encode(), b"\r\n", ] ) for name, path in files.items(): filename = path.name ctype = mimetypes.guess_type(filename)[0] or "application/octet-stream" chunks.extend( [ f"--{boundary}\r\n".encode(), f'Content-Disposition: form-data; name="{name}"; filename="{filename}"\r\n'.encode(), f"Content-Type: {ctype}\r\n\r\n".encode(), path.read_bytes(), b"\r\n", ] ) chunks.append(f"--{boundary}--\r\n".encode()) return b"".join(chunks), f"multipart/form-data; boundary={boundary}" def cmd_health(args: argparse.Namespace) -> int: client = Client(registry_url(getattr(args, "registry", None)), token=getattr(args, "token", None), use_cookies=False) payload = client.request("GET", "/actuator/health", timeout=10) print_json(payload) return 0 def cmd_whoami(args: argparse.Namespace) -> int: client = Client(registry_url(getattr(args, "registry", None)), token=getattr(args, "token", None)) payload = client.request("GET", "/api/cli/v1/auth") print_json(payload) return 0 def cmd_login(args: argparse.Namespace) -> int: username = args.username or input("Username: ") password = args.password or getpass.getpass("Password: ") client = Client(registry_url(getattr(args, "registry", None)), token=getattr(args, "token", None)) body = json.dumps({"username": username, "password": password}).encode("utf-8") payload = client.request( "POST", "/api/v1/auth/local/login", data=body, headers={"Content-Type": "application/json"}, ) save_cookies(client.cookies) print_json(payload) print(f"Saved session cookies to {COOKIE_FILE}", file=sys.stderr) return 0 def cmd_token_create(args: argparse.Namespace) -> int: client = Client(registry_url(getattr(args, "registry", None)), token=getattr(args, "token", None)) body = json.dumps( { "name": args.name, "scopes": args.scope, "expiresAt": args.expires_at, } ).encode("utf-8") payload = client.request( "POST", "/api/v1/tokens", data=body, headers={"Content-Type": "application/json"}, ) print_json(payload) return 0 def cmd_publish(args: argparse.Namespace) -> int: client = Client(registry_url(getattr(args, "registry", None)), token=getattr(args, "token", None)) zip_path = make_package_zip(Path(args.skill_dir)) try: fields = {"visibility": args.visibility} body, content_type = multipart_form(fields, {"file": zip_path}) path = f"/api/cli/v1/skills/{parse.quote(args.namespace)}/publish" payload = client.request( "POST", path, data=body, headers={"Content-Type": content_type}, timeout=args.timeout, ) data = unwrap_data(payload) if args.json: print_json(payload) elif isinstance(data, dict): namespace = data.get("namespace", args.namespace) slug = data.get("slug") or Path(args.skill_dir).name version = data.get("version") or "unknown" visibility = data.get("visibility") or args.visibility print(f"Published {namespace}/{slug}@{version} ({visibility})") else: print_json(payload) return 0 finally: shutil.rmtree(zip_path.parent, ignore_errors=True) def cmd_search(args: argparse.Namespace) -> int: client = Client(registry_url(getattr(args, "registry", None)), token=getattr(args, "token", None)) query = parse.urlencode( { "q": args.query, "limit": args.limit, "namespace": args.namespace or "", } ) payload = client.request("GET", f"/api/cli/v1/skills/search?{query}") print_json(payload) return 0 def build_parser() -> argparse.ArgumentParser: common = argparse.ArgumentParser(add_help=False) common.add_argument( "--registry", default=argparse.SUPPRESS, help=f"SkillHub API base URL. Default: {DEFAULT_REGISTRY}", ) common.add_argument( "--token", default=argparse.SUPPRESS, help="API token. Defaults to CLAWHUB_TOKEN or SKILLHUB_TOKEN.", ) parser = argparse.ArgumentParser( prog="clawhub", description="SkillHub/ClawHub offline CLI", parents=[common], ) sub = parser.add_subparsers(dest="command", required=True) health = sub.add_parser("health", help="Check SkillHub backend health", parents=[common]) health.set_defaults(func=cmd_health) whoami = sub.add_parser("whoami", help="Show authenticated user", parents=[common]) whoami.set_defaults(func=cmd_whoami) login = sub.add_parser("login", help="Login with local username/password and save session cookie", parents=[common]) login.add_argument("--username") login.add_argument("--password") login.set_defaults(func=cmd_login) token_create = sub.add_parser( "token-create", help="Create an API token using the current session", parents=[common], ) token_create.add_argument("--name", default="clawhub-cli") token_create.add_argument("--scope", action="append", default=[], help="Token scope; may be repeated") token_create.add_argument("--expires-at", default=None, help="ISO timestamp or server-supported date string") token_create.set_defaults(func=cmd_token_create) publish = sub.add_parser("publish", help="Publish a skill directory", parents=[common]) publish.add_argument("skill_dir") publish.add_argument("--namespace", required=True) publish.add_argument("--visibility", default="PUBLIC", choices=["PUBLIC", "PRIVATE", "INTERNAL"]) publish.add_argument("--timeout", type=int, default=180) publish.add_argument("--json", action="store_true", help="Print full JSON response") publish.set_defaults(func=cmd_publish) search = sub.add_parser("search", help="Search skills through the CLI API", parents=[common]) search.add_argument("query") search.add_argument("--namespace") search.add_argument("--limit", type=int, default=20) search.set_defaults(func=cmd_search) return parser def main(argv: list[str] | None = None) -> int: parser = build_parser() args = parser.parse_args(argv) try: return args.func(args) except CliError as exc: print(f"clawhub: {exc}", file=sys.stderr) return 1 except KeyboardInterrupt: print("clawhub: interrupted", file=sys.stderr) return 130 if __name__ == "__main__": raise SystemExit(main())