feat(skills): centralized skills index — eliminate GitHub API calls for search/install
Add a CI-built skills index served from the docs site. The index is crawled daily by GitHub Actions, resolves all GitHub paths upfront, and is cached locally by the client. When the index is available: - Search uses the cached index (0 GitHub API calls, was 23+) - Install uses resolved paths from index (6 API calls for file downloads only, was 31-45 for discovery + downloads) Total: 68 → 6 GitHub API calls for a typical search + install flow. Unauthenticated users (60 req/hr) can now search and install without hitting rate limits. Components: - scripts/build_skills_index.py: Crawl all sources (skills.sh, GitHub taps, official, clawhub, lobehub), batch-resolve GitHub paths via tree API, output JSON index - tools/skills_hub.py: HermesIndexSource class — search/fetch/inspect backed by the index, with lazy GitHubSource for file downloads - parallel_search_sources() skips external API sources when index is available (0 GitHub calls for search) - .github/workflows/skills-index.yml: twice-daily CI build + deploy - .github/workflows/deploy-site.yml: also builds index during docs deploy Graceful degradation: when the index is unavailable (first run, network down, stale), all methods return empty/None and downstream sources handle the request via direct API as before.
This commit is contained in:
Teknium
@@ -2698,6 +2698,222 @@ def check_for_skill_updates(
|
||||
return results
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Hermes centralized index source
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
HERMES_INDEX_URL = "https://hermes-agent.nousresearch.com/docs/api/skills-index.json"
|
||||
HERMES_INDEX_CACHE_FILE = INDEX_CACHE_DIR / "hermes-index.json"
|
||||
HERMES_INDEX_TTL = 6 * 3600 # 6 hours
|
||||
|
||||
|
||||
def _load_hermes_index() -> Optional[dict]:
|
||||
"""Fetch the centralized skills index, with local cache.
|
||||
|
||||
The index is a JSON file hosted on the docs site, rebuilt daily by CI.
|
||||
We cache it locally for HERMES_INDEX_TTL seconds to avoid repeated
|
||||
downloads within a session.
|
||||
"""
|
||||
# Check local cache
|
||||
if HERMES_INDEX_CACHE_FILE.exists():
|
||||
try:
|
||||
age = time.time() - HERMES_INDEX_CACHE_FILE.stat().st_mtime
|
||||
if age < HERMES_INDEX_TTL:
|
||||
return json.loads(HERMES_INDEX_CACHE_FILE.read_text())
|
||||
except (OSError, json.JSONDecodeError):
|
||||
pass
|
||||
|
||||
# Fetch from docs site
|
||||
try:
|
||||
resp = httpx.get(HERMES_INDEX_URL, timeout=15, follow_redirects=True)
|
||||
if resp.status_code != 200:
|
||||
logger.debug("Hermes index fetch returned %d", resp.status_code)
|
||||
return _load_stale_index_cache()
|
||||
data = resp.json()
|
||||
except (httpx.HTTPError, json.JSONDecodeError) as e:
|
||||
logger.debug("Hermes index fetch failed: %s", e)
|
||||
return _load_stale_index_cache()
|
||||
|
||||
# Validate structure
|
||||
if not isinstance(data, dict) or "skills" not in data:
|
||||
return _load_stale_index_cache()
|
||||
|
||||
# Cache locally
|
||||
try:
|
||||
HERMES_INDEX_CACHE_FILE.parent.mkdir(parents=True, exist_ok=True)
|
||||
HERMES_INDEX_CACHE_FILE.write_text(json.dumps(data))
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
return data
|
||||
|
||||
|
||||
def _load_stale_index_cache() -> Optional[dict]:
|
||||
"""Fall back to stale cache when the network fetch fails."""
|
||||
if HERMES_INDEX_CACHE_FILE.exists():
|
||||
try:
|
||||
return json.loads(HERMES_INDEX_CACHE_FILE.read_text())
|
||||
except (OSError, json.JSONDecodeError):
|
||||
pass
|
||||
return None
|
||||
|
||||
|
||||
class HermesIndexSource(SkillSource):
|
||||
"""Skill source backed by the centralized Hermes Skills Index.
|
||||
|
||||
The index is a JSON catalog published to the docs site and rebuilt
|
||||
daily by CI. It contains metadata + resolved GitHub paths for every
|
||||
skill, eliminating the need for users to hit the GitHub API for
|
||||
search or path discovery.
|
||||
|
||||
When the index is unavailable, all methods return empty / None so
|
||||
downstream sources take over transparently.
|
||||
"""
|
||||
|
||||
def __init__(self, auth: GitHubAuth):
|
||||
self._index: Optional[dict] = None
|
||||
self._loaded = False
|
||||
self.auth = auth
|
||||
# Lazily create GitHubSource for fetch — only used when actually
|
||||
# downloading files, which requires real GitHub API calls.
|
||||
self._github: Optional[GitHubSource] = None
|
||||
|
||||
def _ensure_loaded(self) -> dict:
|
||||
if not self._loaded:
|
||||
self._index = _load_hermes_index()
|
||||
self._loaded = True
|
||||
return self._index or {}
|
||||
|
||||
def _get_github(self) -> GitHubSource:
|
||||
if self._github is None:
|
||||
self._github = GitHubSource(auth=self.auth)
|
||||
return self._github
|
||||
|
||||
def source_id(self) -> str:
|
||||
return "hermes-index"
|
||||
|
||||
@property
|
||||
def is_available(self) -> bool:
|
||||
"""Whether the index is loaded and has skills."""
|
||||
index = self._ensure_loaded()
|
||||
return bool(index.get("skills"))
|
||||
|
||||
def trust_level_for(self, identifier: str) -> str:
|
||||
index = self._ensure_loaded()
|
||||
for skill in index.get("skills", []):
|
||||
if skill.get("identifier") == identifier:
|
||||
return skill.get("trust_level", "community")
|
||||
return "community"
|
||||
|
||||
def search(self, query: str, limit: int = 10) -> List[SkillMeta]:
|
||||
"""Search the cached index. Zero API calls."""
|
||||
index = self._ensure_loaded()
|
||||
skills = index.get("skills", [])
|
||||
if not skills:
|
||||
return []
|
||||
|
||||
if not query.strip():
|
||||
# No query — return featured/popular
|
||||
return [self._to_meta(s) for s in skills[:limit]]
|
||||
|
||||
query_lower = query.lower()
|
||||
results: List[SkillMeta] = []
|
||||
for s in skills:
|
||||
searchable = f"{s.get('name', '')} {s.get('description', '')} {' '.join(s.get('tags', []))}".lower()
|
||||
if query_lower in searchable:
|
||||
results.append(self._to_meta(s))
|
||||
if len(results) >= limit:
|
||||
break
|
||||
return results
|
||||
|
||||
def fetch(self, identifier: str) -> Optional[SkillBundle]:
|
||||
"""Fetch a skill using the resolved path from the index.
|
||||
|
||||
If the index has a ``resolved_github_id`` for this skill, we skip
|
||||
the entire candidate/discovery chain and go directly to GitHub
|
||||
with the exact path. This reduces install from ~31 API calls to
|
||||
just the file content downloads (~5-22 depending on skill size).
|
||||
"""
|
||||
index = self._ensure_loaded()
|
||||
entry = self._find_entry(identifier, index)
|
||||
if not entry:
|
||||
return None
|
||||
|
||||
# Use resolved path if available
|
||||
resolved = entry.get("resolved_github_id")
|
||||
if resolved:
|
||||
bundle = self._get_github().fetch(resolved)
|
||||
if bundle:
|
||||
bundle.source = entry.get("source", "hermes-index")
|
||||
bundle.identifier = identifier
|
||||
return bundle
|
||||
|
||||
# Fall back to identifier-based fetch via repo/path
|
||||
repo = entry.get("repo", "")
|
||||
path = entry.get("path", "")
|
||||
if repo and path:
|
||||
github_id = f"{repo}/{path}"
|
||||
bundle = self._get_github().fetch(github_id)
|
||||
if bundle:
|
||||
bundle.source = entry.get("source", "hermes-index")
|
||||
bundle.identifier = identifier
|
||||
return bundle
|
||||
|
||||
return None
|
||||
|
||||
def inspect(self, identifier: str) -> Optional[SkillMeta]:
|
||||
"""Return metadata from the index. Zero API calls."""
|
||||
index = self._ensure_loaded()
|
||||
entry = self._find_entry(identifier, index)
|
||||
if entry:
|
||||
return self._to_meta(entry)
|
||||
return None
|
||||
|
||||
def _find_entry(self, identifier: str, index: dict) -> Optional[dict]:
|
||||
"""Look up a skill in the index by identifier or name."""
|
||||
skills = index.get("skills", [])
|
||||
|
||||
# Exact identifier match
|
||||
for s in skills:
|
||||
if s.get("identifier") == identifier:
|
||||
return s
|
||||
|
||||
# Try without source prefix (e.g. "skills-sh/" stripped)
|
||||
normalized = identifier
|
||||
for prefix in ("skills-sh/", "skills.sh/", "official/", "github/", "clawhub/"):
|
||||
if identifier.startswith(prefix):
|
||||
normalized = identifier[len(prefix):]
|
||||
break
|
||||
|
||||
# Match on normalized identifier or name
|
||||
for s in skills:
|
||||
sid = s.get("identifier", "")
|
||||
# Strip prefix from stored identifier too
|
||||
stored_normalized = sid
|
||||
for prefix in ("skills-sh/", "skills.sh/", "official/", "github/", "clawhub/"):
|
||||
if sid.startswith(prefix):
|
||||
stored_normalized = sid[len(prefix):]
|
||||
break
|
||||
if stored_normalized == normalized:
|
||||
return s
|
||||
|
||||
return None
|
||||
|
||||
@staticmethod
|
||||
def _to_meta(entry: dict) -> SkillMeta:
|
||||
return SkillMeta(
|
||||
name=entry.get("name", ""),
|
||||
description=entry.get("description", ""),
|
||||
source=entry.get("source", "hermes-index"),
|
||||
identifier=entry.get("identifier", ""),
|
||||
trust_level=entry.get("trust_level", "community"),
|
||||
repo=entry.get("repo"),
|
||||
path=entry.get("path"),
|
||||
tags=entry.get("tags", []),
|
||||
extra=entry.get("extra", {}),
|
||||
)
|
||||
|
||||
|
||||
def create_source_router(auth: Optional[GitHubAuth] = None) -> List[SkillSource]:
|
||||
"""
|
||||
Create all configured source adapters.
|
||||
@@ -2711,6 +2927,7 @@ def create_source_router(auth: Optional[GitHubAuth] = None) -> List[SkillSource]
|
||||
|
||||
sources: List[SkillSource] = [
|
||||
OptionalSkillSource(), # Official optional skills (highest priority)
|
||||
HermesIndexSource(auth=auth), # Centralized index (search + resolved install paths)
|
||||
SkillsShSource(auth=auth),
|
||||
WellKnownSkillSource(),
|
||||
GitHubSource(auth=auth, extra_taps=extra_taps),
|
||||
@@ -2753,10 +2970,27 @@ def parallel_search_sources(
|
||||
per_source_limits = per_source_limits or {}
|
||||
|
||||
active: List[SkillSource] = []
|
||||
# When the centralized index is available and the user hasn't filtered
|
||||
# to a specific source, skip external API sources (github, skills-sh,
|
||||
# clawhub, etc.) — the index already has their data. This avoids
|
||||
# ~70 GitHub API calls per search for unauthenticated users.
|
||||
_index_available = False
|
||||
_api_source_ids = frozenset({"github", "skills-sh", "clawhub",
|
||||
"claude-marketplace", "lobehub", "well-known"})
|
||||
if source_filter == "all":
|
||||
for src in sources:
|
||||
if (src.source_id() == "hermes-index"
|
||||
and getattr(src, "is_available", False)):
|
||||
_index_available = True
|
||||
break
|
||||
|
||||
for src in sources:
|
||||
sid = src.source_id()
|
||||
if source_filter != "all" and sid != source_filter and sid != "official":
|
||||
continue
|
||||
# Skip external API sources when the index covers them
|
||||
if _index_available and sid in _api_source_ids:
|
||||
continue
|
||||
active.append(src)
|
||||
|
||||
all_results: List[SkillMeta] = []
|
||||
|
||||
Reference in New Issue
Block a user