docs(examples): add Python client and usage examples (#702)

Add examples/python: a dependency-light (requests-only) SkillHubClient plus
runnable examples covering search, resolve, download and publish against the
REST API, and link them from the README Documentation sections. Serves the
large Python-leaning audience and doubles as a reference SDK seed (see #701).

Signed-off-by: FenjuFu <fufenjupku@gmail.com>
This commit is contained in:
FenjuFu 2026-08-26 19:43:13 +08:00 committed by GitHub
parent 16306dd4f4
commit 0587c55f8b
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
6 changed files with 422 additions and 0 deletions

View file

@ -64,6 +64,7 @@ with the Skill's source and the problem it solves, or submit a PR by following t
- 📖 **[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

View file

@ -50,6 +50,7 @@ Skill欢迎分享给 SkillHub 社区,与大家一起丰富开放、实用
- 📖 **[用户指南](https://iflytek.github.io/skillhub/)** — 技能发布、搜索、CLI 使用等用户操作指南
- 🛠️ **[开发者文档](https://zread.ai/iflytek/skillhub)** — 架构设计、API 参考、本地开发、部署运维等技术文档
- 🐍 **[Python 示例](./examples/python)** — 使用 REST API 在 Python 中搜索、下载和发布技能
## 核心特性

85
examples/python/README.md Normal file
View file

@ -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=<your-api-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).

View file

@ -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=<your-api-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 <zip_path> <namespace>")
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()

View file

@ -0,0 +1 @@
requests>=2.25

View file

@ -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,
)
)