diff --git a/README.md b/README.md index cf803028c..cd0b7f341 100644 --- a/README.md +++ b/README.md @@ -38,6 +38,7 @@ firewall, with the same polish you'd expect from a public registry. - 📖 **[User Guide](https://iflytek.github.io/skillhub/)** — Skill publishing, search, CLI usage and other user guides - 🛠️ **[Developer Docs](https://zread.ai/iflytek/skillhub)** — Architecture, API reference, local development, deployment and operations +- 🐍 **[Python Examples](./examples/python)** — Search, download, and publish skills from Python via the REST API ## Highlights diff --git a/README_zh.md b/README_zh.md index edb58fc82..facf7c15f 100644 --- a/README_zh.md +++ b/README_zh.md @@ -27,6 +27,7 @@ SkillHub 是一个自托管平台,为团队提供私有的、受治理的智 - 📖 **[用户指南](https://iflytek.github.io/skillhub/)** — 技能发布、搜索、CLI 使用等用户操作指南 - 🛠️ **[开发者文档](https://zread.ai/iflytek/skillhub)** — 架构设计、API 参考、本地开发、部署运维等技术文档 +- 🐍 **[Python 示例](./examples/python)** — 使用 REST API 在 Python 中搜索、下载和发布技能 ## 核心特性 diff --git a/examples/python/README.md b/examples/python/README.md new file mode 100644 index 000000000..39f0e7a2a --- /dev/null +++ b/examples/python/README.md @@ -0,0 +1,85 @@ +# SkillHub Python Examples + +A minimal, dependency-light (`requests`-only) Python client and runnable +examples for the SkillHub REST API. Use it to **search, inspect, download, +and publish** skills from Python — the same operations the ClawHub CLI +performs, without shelling out to the CLI. + +> These are reference examples, not (yet) an officially published pip +> package. See [iflytek/skillhub#701](https://github.com/iflytek/skillhub/issues/701) +> for the discussion on whether to ship a full published SDK. + +## Files + +| File | What it is | +|------|------------| +| [`skillhub_client.py`](./skillhub_client.py) | A small `SkillHubClient` class wrapping the REST API | +| [`example_usage.py`](./example_usage.py) | Runnable script: search → resolve → download, and publish | +| [`requirements.txt`](./requirements.txt) | The only dependency: `requests` | + +## Setup + +```bash +pip install -r requirements.txt + +# Point at your SkillHub instance +export SKILLHUB_URL=https://skill.example.com +# Only needed for write operations (publish / star / rate) +export SKILLHUB_TOKEN= +``` + +Generate an API token from the SkillHub web UI (**Settings → API Tokens**) or +via `POST /api/v1/tokens`. + +## Quick start + +```python +from skillhub_client import SkillHubClient + +client = SkillHubClient() # reads SKILLHUB_URL / SKILLHUB_TOKEN from env + +# Search public skills +results = client.search(keyword="email", size=5) + +# Inspect and resolve a version +detail = client.get_skill("my-namespace", "my-skill") +resolved = client.resolve("my-namespace", "my-skill", tag="stable") + +# Download the latest package (returns the written file path) +path = client.download("my-namespace", "my-skill") + +# Publish a skill package (requires a token) +client.publish("./my-skill.zip", namespace="my-namespace") +``` + +Or run the end-to-end script: + +```bash +python example_usage.py # search + inspect + download +python example_usage.py publish ./my-skill.zip my-namespace +``` + +## Supported operations + +| Method | Endpoint | Auth | +|--------|----------|------| +| `search(keyword, namespace, page, size)` | `GET /api/v1/skills` | — | +| `get_skill(namespace, slug)` | `GET /api/v1/skills/{ns}/{slug}` | — | +| `list_versions(namespace, slug)` | `GET /api/v1/skills/{ns}/{slug}/versions` | — | +| `resolve(namespace, slug, version, tag)` | `GET /api/v1/skills/{ns}/{slug}/resolve` | — | +| `download(namespace, slug, version, dest)` | `GET /api/v1/skills/{ns}/{slug}[/versions/{v}]/download` | — | +| `whoami()` | `GET /api/v1/whoami` | Bearer | +| `publish(zip_path, namespace, request_id)` | `POST /api/v1/publish` | Bearer | +| `star(namespace, slug)` | `POST /api/v1/skills/{ns}/{slug}/star` | Bearer | +| `rate(namespace, slug, score)` | `POST /api/v1/skills/{ns}/{slug}/rating` | Bearer | + +The client unwraps the unified `{code, msg, data}` response envelope +automatically and raises `SkillHubError` on a non-zero business code. + +## Notes + +- Write operations accept an optional `request_id` (a UUID) that is sent as + the `X-Request-Id` header for idempotency. +- For the full API surface (namespaces, reviews, promotion, tags), see the + [Developer Docs → API](https://iflytek.github.io/skillhub/) and + [`document/docs/04-developer/api`](../../document/docs/04-developer/api). diff --git a/examples/python/example_usage.py b/examples/python/example_usage.py new file mode 100644 index 000000000..71540caa9 --- /dev/null +++ b/examples/python/example_usage.py @@ -0,0 +1,90 @@ +"""Runnable examples for the SkillHub Python client. + +Configure the target registry via environment variables: + + export SKILLHUB_URL=https://skill.example.com + export SKILLHUB_TOKEN= # only needed for write operations + +Then run: + + python example_usage.py # search + inspect + download + python example_usage.py publish ./my-skill.zip my-namespace +""" + +from __future__ import annotations + +import os +import sys + +from skillhub_client import SkillHubClient, SkillHubError + + +def _pick(obj, *keys, default=None): + """Best-effort field access across slightly different response shapes.""" + for key in keys: + if isinstance(obj, dict) and obj.get(key) is not None: + return obj[key] + return default + + +def demo_read(client: SkillHubClient) -> None: + print(f"Searching {client.base_url} for skills matching 'email'...\n") + result = client.search(keyword="email", size=5) + + # The search payload may expose the list under 'items' or 'results'. + items = _pick(result, "items", "results", default=result if isinstance(result, list) else []) + if not items: + print("No skills found. Try a different keyword or registry.") + return + + for skill in items: + name = _pick(skill, "name", "slug", default="(unnamed)") + ns = _pick(skill, "namespace", default="") + version = _pick(skill, "version", "latestVersion", default="?") + downloads = _pick(skill, "downloadCount", "downloads", default=0) + coord = f"{ns}/{name}" if ns else name + print(f" - {coord} v{version} ({downloads} downloads)") + + # Download the first result's latest package. + first = items[0] + ns = _pick(first, "namespace", default="") + slug = _pick(first, "slug", "name") + if ns and slug: + print(f"\nResolving latest version of {ns}/{slug}...") + resolved = client.resolve(ns, slug) + version = _pick(resolved, "version", default=None) + print(f" resolved version: {version}") + + dest = client.download(ns, slug, version=version) + size = os.path.getsize(dest) + print(f" downloaded -> {dest} ({size} bytes)") + + +def demo_publish(client: SkillHubClient, zip_path: str, namespace: str) -> None: + if not client.token: + sys.exit("Publishing requires SKILLHUB_TOKEN to be set.") + print(f"Publishing {zip_path} to namespace '{namespace}'...") + result = client.publish(zip_path, namespace) + print(f" published: {result}") + + +def main() -> None: + try: + client = SkillHubClient() + except ValueError as exc: + sys.exit(str(exc)) + + args = sys.argv[1:] + try: + if args and args[0] == "publish": + if len(args) != 3: + sys.exit("usage: python example_usage.py publish ") + demo_publish(client, args[1], args[2]) + else: + demo_read(client) + except SkillHubError as exc: + sys.exit(f"API error: {exc}") + + +if __name__ == "__main__": + main() diff --git a/examples/python/requirements.txt b/examples/python/requirements.txt new file mode 100644 index 000000000..e8691f91f --- /dev/null +++ b/examples/python/requirements.txt @@ -0,0 +1 @@ +requests>=2.25 diff --git a/examples/python/skillhub_client.py b/examples/python/skillhub_client.py new file mode 100644 index 000000000..f684ed958 --- /dev/null +++ b/examples/python/skillhub_client.py @@ -0,0 +1,244 @@ +"""A minimal Python client for the SkillHub REST API. + +This is a dependency-light reference client (only ``requests``) that mirrors +the operations the ClawHub CLI performs: search, inspect, resolve, download +and publish skills. It is meant as a copy-pasteable starting point for Python +integrations, not (yet) an officially published package. + +API reference: https://iflytek.github.io/skillhub/ (Developer Docs -> API) + +Endpoints used (see docs/04-developer/api): + Public (no auth): + GET /api/v1/skills?keyword=&namespace=&page=&size= + GET /api/v1/skills/{namespace}/{slug} + GET /api/v1/skills/{namespace}/{slug}/versions + GET /api/v1/skills/{namespace}/{slug}/resolve?version=&tag= + GET /api/v1/skills/{namespace}/{slug}/download + GET /api/v1/skills/{namespace}/{slug}/versions/{version}/download + Authenticated (Bearer token): + GET /api/v1/whoami + POST /api/v1/publish (multipart: file, namespace) + POST /api/v1/skills/{namespace}/{slug}/star + POST /api/v1/skills/{namespace}/{slug}/rating (json: {"score": 1-5}) +""" + +from __future__ import annotations + +import os +from typing import Any, Dict, Optional + +import requests + + +class SkillHubError(RuntimeError): + """Raised when the API returns a non-zero business code.""" + + def __init__(self, code: Any, message: str, request_id: Optional[str] = None): + self.code = code + self.request_id = request_id + super().__init__(f"SkillHub API error {code}: {message}" + + (f" (requestId={request_id})" if request_id else "")) + + +class SkillHubClient: + """Thin wrapper over the SkillHub REST API. + + Args: + base_url: Registry base URL, e.g. ``https://skill.example.com``. + token: Optional API token for authenticated calls (Bearer). + timeout: Per-request timeout in seconds. + session: Optional pre-configured ``requests.Session``. + """ + + def __init__( + self, + base_url: Optional[str] = None, + token: Optional[str] = None, + timeout: int = 30, + session: Optional[requests.Session] = None, + ): + base_url = base_url or os.environ.get("SKILLHUB_URL") + if not base_url: + raise ValueError( + "base_url is required (pass it explicitly or set SKILLHUB_URL)" + ) + self.base_url = base_url.rstrip("/") + self.token = token or os.environ.get("SKILLHUB_TOKEN") + self.timeout = timeout + self.session = session or requests.Session() + + # -- internals ------------------------------------------------------- + + def _headers(self, extra: Optional[Dict[str, str]] = None) -> Dict[str, str]: + headers: Dict[str, str] = {} + if self.token: + headers["Authorization"] = f"Bearer {self.token}" + if extra: + headers.update(extra) + return headers + + def _url(self, path: str) -> str: + return f"{self.base_url}{path}" + + def _unwrap(self, resp: requests.Response) -> Any: + """Return the payload, unwrapping the ``{code,msg,data}`` envelope. + + Native ``/api/v1`` endpoints wrap responses in a unified envelope, + while the CLI-compat endpoints return the object directly. This + handles both. + """ + resp.raise_for_status() + payload = resp.json() + if isinstance(payload, dict) and "code" in payload and "data" in payload: + if payload.get("code") not in (0, None): + raise SkillHubError( + payload.get("code"), payload.get("msg", ""), payload.get("requestId") + ) + return payload["data"] + return payload + + # -- public API ------------------------------------------------------ + + def search( + self, + keyword: Optional[str] = None, + namespace: Optional[str] = None, + page: int = 1, + size: int = 20, + ) -> Any: + """Search public skills.""" + params = {"keyword": keyword, "namespace": namespace, "page": page, "size": size} + params = {k: v for k, v in params.items() if v is not None} + return self._unwrap( + self.session.get( + self._url("/api/v1/skills"), + params=params, + headers=self._headers(), + timeout=self.timeout, + ) + ) + + def get_skill(self, namespace: str, slug: str) -> Any: + """Fetch a single skill's detail.""" + return self._unwrap( + self.session.get( + self._url(f"/api/v1/skills/{namespace}/{slug}"), + headers=self._headers(), + timeout=self.timeout, + ) + ) + + def list_versions(self, namespace: str, slug: str) -> Any: + """List all versions of a skill.""" + return self._unwrap( + self.session.get( + self._url(f"/api/v1/skills/{namespace}/{slug}/versions"), + headers=self._headers(), + timeout=self.timeout, + ) + ) + + def resolve( + self, + namespace: str, + slug: str, + version: Optional[str] = None, + tag: Optional[str] = None, + ) -> Any: + """Resolve a version constraint / tag to a concrete version.""" + params = {"version": version, "tag": tag} + params = {k: v for k, v in params.items() if v is not None} + return self._unwrap( + self.session.get( + self._url(f"/api/v1/skills/{namespace}/{slug}/resolve"), + params=params, + headers=self._headers(), + timeout=self.timeout, + ) + ) + + def download( + self, + namespace: str, + slug: str, + version: Optional[str] = None, + dest: Optional[str] = None, + ) -> str: + """Download a skill package (zip). Returns the written file path. + + If ``version`` is omitted the ``latest`` package is downloaded. If + ``dest`` is omitted a file named ``{slug}-{version}.zip`` (or + ``{slug}.zip``) is written to the current directory. + """ + if version: + path = f"/api/v1/skills/{namespace}/{slug}/versions/{version}/download" + else: + path = f"/api/v1/skills/{namespace}/{slug}/download" + if dest is None: + dest = f"{slug}-{version}.zip" if version else f"{slug}.zip" + with self.session.get( + self._url(path), headers=self._headers(), timeout=self.timeout, stream=True + ) as resp: + resp.raise_for_status() + with open(dest, "wb") as fh: + for chunk in resp.iter_content(chunk_size=8192): + if chunk: + fh.write(chunk) + return dest + + # -- authenticated API ---------------------------------------------- + + def whoami(self) -> Any: + """Return the authenticated principal (requires a token).""" + return self._unwrap( + self.session.get( + self._url("/api/v1/whoami"), + headers=self._headers(), + timeout=self.timeout, + ) + ) + + def publish( + self, zip_path: str, namespace: str, request_id: Optional[str] = None + ) -> Any: + """Publish a skill package (zip) to a namespace. Requires a token. + + Pass ``request_id`` (a UUID) to make the publish idempotent via the + ``X-Request-Id`` header. + """ + extra = {"X-Request-Id": request_id} if request_id else None + with open(zip_path, "rb") as fh: + files = {"file": (os.path.basename(zip_path), fh, "application/zip")} + data = {"namespace": namespace} + return self._unwrap( + self.session.post( + self._url("/api/v1/publish"), + files=files, + data=data, + headers=self._headers(extra), + timeout=self.timeout, + ) + ) + + def star(self, namespace: str, slug: str) -> Any: + """Star a skill. Requires a token.""" + return self._unwrap( + self.session.post( + self._url(f"/api/v1/skills/{namespace}/{slug}/star"), + headers=self._headers(), + timeout=self.timeout, + ) + ) + + def rate(self, namespace: str, slug: str, score: int) -> Any: + """Rate a skill from 1 to 5. Requires a token.""" + if not 1 <= score <= 5: + raise ValueError("score must be between 1 and 5") + return self._unwrap( + self.session.post( + self._url(f"/api/v1/skills/{namespace}/{slug}/rating"), + json={"score": score}, + headers=self._headers(), + timeout=self.timeout, + ) + )