Coverage for integrations / skills / registry.py: 0.0%
222 statements
« prev ^ index » next coverage.py v7.14.0, created at 2026-05-12 04:49 +0000
1"""
2Skill Registry - follows ServiceToolRegistry / MCPToolRegistry pattern.
4Ingests agent skills (Claude Code SKILL.md, plain Markdown, JSON) into the
5Hevolve pipeline so any HART agent can execute them during thought experiments.
7Design:
8- SkillInfo describes a skill's metadata, instructions, and capabilities
9- SkillRegistry manages discovery, storage, and LangChain tool generation
10- Global singleton: skill_registry (mirrors service_tool_registry)
11- Persists to skills.json for startup reload
12"""
14import json
15import logging
16import os
17import re
18import threading
19from dataclasses import dataclass, field
20from datetime import datetime
21from typing import Dict, List, Any, Optional
23logger = logging.getLogger(__name__)
25try:
26 from core.labeled_tool import labeled_tool
27except ImportError: # cx_Freeze / degraded test env
28 def labeled_tool(name, func, description, *, ui_label): # type: ignore
29 from langchain.agents import Tool as _Tool
30 return _Tool(name=name, func=func, description=description)
33# ---------------------------------------------------------------------------
34# YAML frontmatter parser (no PyYAML dependency - keep it self-contained)
35# ---------------------------------------------------------------------------
37def _parse_frontmatter(text: str) -> tuple:
38 """Parse YAML frontmatter from SKILL.md content.
40 Returns (frontmatter_dict, body_markdown).
41 If no frontmatter found, returns ({}, full_text).
42 """
43 text = text.lstrip()
44 if not text.startswith("---"):
45 return {}, text
47 end = text.find("---", 3)
48 if end == -1:
49 return {}, text
51 yaml_block = text[3:end].strip()
52 body = text[end + 3:].strip()
54 # Minimal YAML parser - handles key: value, key: [list], key: "quoted"
55 meta: Dict[str, Any] = {}
56 for line in yaml_block.splitlines():
57 line = line.strip()
58 if not line or line.startswith("#"):
59 continue
60 m = re.match(r'^([A-Za-z_][\w-]*)\s*:\s*(.*)', line)
61 if not m:
62 continue
63 key = m.group(1).strip()
64 val = m.group(2).strip()
66 # Strip quotes
67 if (val.startswith('"') and val.endswith('"')) or \
68 (val.startswith("'") and val.endswith("'")):
69 val = val[1:-1]
71 # Inline list [a, b, c]
72 if val.startswith("[") and val.endswith("]"):
73 val = [v.strip().strip("'\"") for v in val[1:-1].split(",") if v.strip()]
75 # Boolean
76 if isinstance(val, str):
77 if val.lower() in ("true", "yes"):
78 val = True
79 elif val.lower() in ("false", "no"):
80 val = False
82 meta[key] = val
84 return meta, body
87# ---------------------------------------------------------------------------
88# SkillInfo
89# ---------------------------------------------------------------------------
91@dataclass
92class SkillInfo:
93 """Metadata + content for an ingested skill."""
94 name: str
95 description: str
96 instructions: str # Markdown body - the actual skill prompt
97 source: str = "local" # local | github | http | builtin
98 source_path: str = "" # File path or URL it was loaded from
99 allowed_tools: List[str] = field(default_factory=list)
100 tags: List[str] = field(default_factory=list)
101 user_invocable: bool = True
102 context: str = "inline" # inline | fork
103 version: str = "1.0.0"
104 author: str = ""
105 registered_at: Optional[str] = None
107 def to_dict(self) -> Dict[str, Any]:
108 return {
109 "name": self.name,
110 "description": self.description,
111 "instructions": self.instructions,
112 "source": self.source,
113 "source_path": self.source_path,
114 "allowed_tools": self.allowed_tools,
115 "tags": self.tags,
116 "user_invocable": self.user_invocable,
117 "context": self.context,
118 "version": self.version,
119 "author": self.author,
120 "registered_at": self.registered_at,
121 }
123 @classmethod
124 def from_dict(cls, data: Dict[str, Any]) -> 'SkillInfo':
125 return cls(
126 name=data["name"],
127 description=data.get("description", ""),
128 instructions=data.get("instructions", ""),
129 source=data.get("source", "local"),
130 source_path=data.get("source_path", ""),
131 allowed_tools=data.get("allowed_tools", []),
132 tags=data.get("tags", []),
133 user_invocable=data.get("user_invocable", True),
134 context=data.get("context", "inline"),
135 version=data.get("version", "1.0.0"),
136 author=data.get("author", ""),
137 registered_at=data.get("registered_at"),
138 )
140 @classmethod
141 def from_skill_md(cls, content: str, source: str = "local",
142 source_path: str = "") -> 'SkillInfo':
143 """Parse a SKILL.md file into a SkillInfo."""
144 meta, body = _parse_frontmatter(content)
146 name = meta.get("name", "")
147 if not name and source_path:
148 # Derive name from path: .claude/skills/my-skill/SKILL.md → my-skill
149 parts = source_path.replace("\\", "/").split("/")
150 for i, p in enumerate(parts):
151 if p == "skills" and i + 1 < len(parts):
152 name = parts[i + 1]
153 break
154 if not name:
155 name = os.path.splitext(os.path.basename(source_path))[0]
157 allowed = meta.get("allowed-tools", meta.get("allowed_tools", []))
158 if isinstance(allowed, str):
159 allowed = [t.strip() for t in allowed.split(",")]
161 tags_raw = meta.get("tags", [])
162 if isinstance(tags_raw, str):
163 tags_raw = [t.strip() for t in tags_raw.split(",")]
165 return cls(
166 name=name,
167 description=meta.get("description", body[:120].replace("\n", " ").strip()),
168 instructions=body,
169 source=source,
170 source_path=source_path,
171 allowed_tools=allowed,
172 tags=tags_raw,
173 user_invocable=not meta.get("disable-model-invocation", False),
174 context=meta.get("context", "inline"),
175 version=meta.get("version", "1.0.0"),
176 author=meta.get("author", ""),
177 )
180# ---------------------------------------------------------------------------
181# SkillRegistry
182# ---------------------------------------------------------------------------
184class SkillRegistry:
185 """
186 Registry for agent skills - any skill definition becomes a HART tool.
188 Mirrors ServiceToolRegistry pattern:
189 - register_skill / unregister_skill
190 - discover_local / discover_github
191 - get_langchain_tools → plugs into hart_intelligence get_tools()
192 - Global singleton: skill_registry
193 """
195 def __init__(self, config_file: str = "skills.json"):
196 self._skills: Dict[str, SkillInfo] = {}
197 self._config_file = config_file
198 self._lock = threading.Lock()
200 # ---- Registration ----
202 def register_skill(self, skill: SkillInfo) -> bool:
203 """Register a skill. Returns True if new, False if already exists."""
204 with self._lock:
205 if skill.name in self._skills:
206 logger.debug(f"Skill '{skill.name}' already registered, updating")
207 skill.registered_at = datetime.now().isoformat()
208 self._skills[skill.name] = skill
209 logger.info(f"Registered skill: {skill.name} (source={skill.source})")
210 return True
212 def unregister_skill(self, name: str) -> bool:
213 with self._lock:
214 if name in self._skills:
215 del self._skills[name]
216 logger.info(f"Unregistered skill: {name}")
217 return True
218 return False
220 def get_skill(self, name: str) -> Optional[SkillInfo]:
221 return self._skills.get(name)
223 def list_skills(self) -> List[Dict[str, Any]]:
224 """List all registered skills (summary view)."""
225 return [
226 {
227 "name": s.name,
228 "description": s.description,
229 "source": s.source,
230 "tags": s.tags,
231 "user_invocable": s.user_invocable,
232 "registered_at": s.registered_at,
233 }
234 for s in self._skills.values()
235 ]
237 # ---- Discovery ----
239 def discover_local(self, search_paths: Optional[List[str]] = None) -> int:
240 """Discover skills from local filesystem.
242 Searches:
243 - ~/.claude/skills/*/SKILL.md (Claude Code user skills)
244 - .claude/skills/*/SKILL.md (project skills)
245 - Custom paths
246 """
247 if search_paths is None:
248 home = os.path.expanduser("~")
249 search_paths = [
250 os.path.join(home, ".claude", "skills"),
251 os.path.join(os.getcwd(), ".claude", "skills"),
252 ]
254 count = 0
255 for base in search_paths:
256 if not os.path.isdir(base):
257 continue
258 for entry in os.listdir(base):
259 skill_dir = os.path.join(base, entry)
260 skill_md = os.path.join(skill_dir, "SKILL.md")
261 if not os.path.isfile(skill_md):
262 # Also check for skill.md (lowercase)
263 skill_md = os.path.join(skill_dir, "skill.md")
264 if not os.path.isfile(skill_md):
265 continue
266 try:
267 with open(skill_md, "r", encoding="utf-8") as f:
268 content = f.read()
269 skill = SkillInfo.from_skill_md(content, source="local",
270 source_path=skill_md)
271 if skill.name:
272 self.register_skill(skill)
273 count += 1
274 except Exception as e:
275 logger.warning(f"Failed to parse {skill_md}: {e}")
277 logger.info(f"Discovered {count} local skills from {len(search_paths)} paths")
278 return count
280 def discover_github(self, repo_url: str, branch: str = "main",
281 skills_path: str = ".claude/skills") -> int:
282 """Discover skills from a GitHub repository.
284 Fetches the repo tree and downloads SKILL.md files.
285 """
286 import urllib.request
287 import urllib.error
289 # Parse owner/repo from URL
290 # Supports: https://github.com/owner/repo or owner/repo
291 repo_url = repo_url.rstrip("/")
292 parts = repo_url.replace("https://github.com/", "").split("/")
293 if len(parts) < 2:
294 logger.error(f"Invalid repo URL: {repo_url}")
295 return 0
296 owner, repo = parts[0], parts[1]
298 # Fetch directory listing via GitHub API
299 api_url = f"https://api.github.com/repos/{owner}/{repo}/contents/{skills_path}?ref={branch}"
300 try:
301 req = urllib.request.Request(api_url, headers={
302 "Accept": "application/vnd.github.v3+json",
303 "User-Agent": "Hevolve-HARTmind/1.0",
304 })
305 with urllib.request.urlopen(req, timeout=10) as resp:
306 entries = json.loads(resp.read().decode())
307 except Exception as e:
308 logger.warning(f"GitHub discovery failed for {repo_url}: {e}")
309 return 0
311 count = 0
312 for entry in entries:
313 if entry.get("type") != "dir":
314 continue
315 skill_name = entry["name"]
316 raw_url = (f"https://raw.githubusercontent.com/{owner}/{repo}"
317 f"/{branch}/{skills_path}/{skill_name}/SKILL.md")
318 try:
319 req = urllib.request.Request(raw_url, headers={
320 "User-Agent": "Hevolve-HARTmind/1.0"
321 })
322 with urllib.request.urlopen(req, timeout=10) as resp:
323 content = resp.read().decode("utf-8")
324 skill = SkillInfo.from_skill_md(content, source="github",
325 source_path=raw_url)
326 if skill.name:
327 self.register_skill(skill)
328 count += 1
329 except Exception as e:
330 logger.debug(f"Skipping {skill_name}: {e}")
332 logger.info(f"Discovered {count} skills from GitHub {owner}/{repo}")
333 return count
335 def ingest_markdown(self, name: str, markdown: str,
336 description: str = "", tags: Optional[List[str]] = None) -> bool:
337 """Ingest a raw Markdown string as a skill (for API / UI uploads)."""
338 # Check if it has frontmatter
339 if markdown.lstrip().startswith("---"):
340 skill = SkillInfo.from_skill_md(markdown, source="api")
341 if not skill.name:
342 skill.name = name
343 else:
344 skill = SkillInfo(
345 name=name,
346 description=description or markdown[:120].replace("\n", " ").strip(),
347 instructions=markdown,
348 source="api",
349 tags=tags or [],
350 )
351 return self.register_skill(skill)
353 # ---- LangChain integration ----
355 def get_langchain_tools(self) -> list:
356 """
357 Get all skills as LangChain Tool objects.
359 Plugs into hart_intelligence get_tools() alongside service_tool_registry.
360 Each skill becomes a tool that returns the skill's instructions with the
361 user's query injected, letting the LLM execute the skill as a thought experiment.
362 """
363 tools = []
364 for name, skill in self._skills.items():
365 if not skill.user_invocable:
366 continue
368 # Capture in closure
369 _skill = skill
371 def execute_skill(query: str, _s=_skill) -> str:
372 """Execute a HART skill by applying its instructions to the query."""
373 # Build the skill execution prompt
374 result = f"## Skill: {_s.name}\n\n"
375 result += f"{_s.instructions}\n\n"
376 result += f"## User Request\n\n{query}\n"
377 return result
379 tool_name = f"hart_skill_{name}"
380 # Friendly UI status label per #508 option C — registry-driven labels.
381 # Source hints: github → "Running Claude Code skill…", local SKILL.md → "Markdown",
382 # builtin → use description fragment. Keep ≤60 chars.
383 _src = (skill.source or "local").lower()
384 if _src == "github":
385 _label = f"Running {skill.name} (Claude Code skill)…"
386 elif _src == "http":
387 _label = f"Running {skill.name} skill from web…"
388 elif _src == "builtin":
389 _label = f"Running {skill.name} (built-in skill)…"
390 else:
391 _label = f"Running {skill.name} skill…"
393 tools.append(labeled_tool(
394 name=tool_name,
395 func=execute_skill,
396 description=(
397 f"[HART Skill] {skill.description}. "
398 f"Use this skill to help with: {', '.join(skill.tags) if skill.tags else skill.name}"
399 ),
400 ui_label=_label[:60],
401 ))
403 return tools
405 def get_autogen_tools(self) -> List[Dict[str, Any]]:
406 """Get all skills as autogen function descriptions.
408 Mirrors ServiceToolRegistry.get_all_tool_functions() for autogen registration.
409 """
410 functions = {}
411 for name, skill in self._skills.items():
412 if not skill.user_invocable:
413 continue
415 _skill = skill
417 def execute(query: str, _s=_skill) -> str:
418 return f"## Skill: {_s.name}\n\n{_s.instructions}\n\n## User Request\n\n{query}\n"
420 func_name = f"hart_skill_{name}"
421 execute.__name__ = func_name
422 execute.__doc__ = skill.description
423 functions[func_name] = execute
425 return functions
427 # ---- Persistence ----
429 def save_config(self) -> None:
430 """Persist registry to JSON."""
431 data = {name: skill.to_dict() for name, skill in self._skills.items()}
432 try:
433 with open(self._config_file, "w", encoding="utf-8") as f:
434 json.dump(data, f, indent=2, ensure_ascii=False)
435 logger.debug(f"Saved {len(data)} skills to {self._config_file}")
436 except Exception as e:
437 logger.warning(f"Failed to save skills config: {e}")
439 def load_config(self) -> int:
440 """Load registry from JSON. Returns number of skills loaded."""
441 if not os.path.exists(self._config_file):
442 return 0
443 try:
444 with open(self._config_file, "r", encoding="utf-8") as f:
445 data = json.load(f)
446 count = 0
447 for name, skill_data in data.items():
448 skill = SkillInfo.from_dict(skill_data)
449 self._skills[name] = skill
450 count += 1
451 logger.info(f"Loaded {count} skills from {self._config_file}")
452 return count
453 except Exception as e:
454 logger.warning(f"Failed to load skills config: {e}")
455 return 0
457 @property
458 def count(self) -> int:
459 return len(self._skills)
462# ---------------------------------------------------------------------------
463# Global singleton
464# ---------------------------------------------------------------------------
466skill_registry = SkillRegistry()