| 1 | """Skills loader for agent capabilities.""" |
| 2 | |
| 3 | import json |
| 4 | import os |
| 5 | import re |
| 6 | import shutil |
| 7 | from pathlib import Path |
| 8 | |
| 9 | import yaml |
| 10 | |
| 11 | # Default builtin skills directory (relative to this file) |
| 12 | BUILTIN_SKILLS_DIR = Path(__file__).parent.parent / "skills" |
| 13 | |
| 14 | # Extra skills directory (project-level, sibling to nanobot package) |
| 15 | EXTRA_SKILLS_DIR = Path(__file__).parent.parent.parent / "extra_skills" |
| 16 | |
| 17 | # Opening ---, YAML body (group 1), closing --- on its own line; supports CRLF. |
| 18 | _STRIP_SKILL_FRONTMATTER = re.compile( |
| 19 | r"^---\s*\r?\n(.*?)\r?\n---\s*\r?\n?", |
| 20 | re.DOTALL, |
| 21 | ) |
| 22 | |
| 23 | |
| 24 | class SkillsLoader: |
| 25 | """ |
| 26 | Loader for agent skills. |
| 27 | |
| 28 | Skills are markdown files (SKILL.md) that teach the agent how to use |
| 29 | specific tools or perform certain tasks. |
| 30 | """ |
| 31 | |
| 32 | def __init__(self, workspace: Path, builtin_skills_dir: Path | None = None, disabled_skills: set[str] | None = None): |
| 33 | self.workspace = workspace |
| 34 | self.workspace_skills = workspace / "skills" |
| 35 | self._include_pe_skills = builtin_skills_dir is None |
| 36 | self.builtin_skills = builtin_skills_dir or BUILTIN_SKILLS_DIR |
| 37 | self.extra_skills = EXTRA_SKILLS_DIR |
| 38 | self.disabled_skills = disabled_skills or set() |
| 39 | |
| 40 | def _pe_skill_dirs(self) -> list[Path]: |
| 41 | """Active-then-default PE ``skills/`` dirs, read live so hot-switch applies.""" |
| 42 | if not self._include_pe_skills: |
| 43 | return [] |
| 44 | try: |
| 45 | from nanobot.prompts import PEManager |
| 46 | |
| 47 | return [Path(p) for p in PEManager.instance().skills_dir()] |
| 48 | except Exception: |
| 49 | return [] |
| 50 | |
| 51 | def _skill_entries_from_dir(self, base: Path, source: str, *, skip_names: set[str] | None = None) -> list[dict[str, str]]: |
| 52 | if not base.exists(): |
| 53 | return [] |
| 54 | entries: list[dict[str, str]] = [] |
| 55 | for skill_dir in base.iterdir(): |
| 56 | if not skill_dir.is_dir(): |
| 57 | continue |
| 58 | skill_file = skill_dir / "SKILL.md" |
| 59 | if not skill_file.exists(): |
| 60 | continue |
| 61 | name = skill_dir.name |
| 62 | if skip_names is not None and name in skip_names: |
| 63 | continue |
| 64 | entries.append({"name": name, "path": str(skill_file), "source": source}) |
| 65 | return entries |
| 66 | |
| 67 | def list_skills(self, filter_unavailable: bool = True) -> list[dict[str, str]]: |
| 68 | """ |
| 69 | List all available skills. |
| 70 | |
| 71 | Args: |
| 72 | filter_unavailable: If True, filter out skills with unmet requirements. |
| 73 | |
| 74 | Returns: |
| 75 | List of skill info dicts with 'name', 'path', 'source'. |
| 76 | """ |
| 77 | skills: list[dict[str, str]] = [] |
| 78 | seen_names: set[str] = set() |
| 79 | for pe_dir in self._pe_skill_dirs(): |
| 80 | skills.extend(self._skill_entries_from_dir(pe_dir, "pe", skip_names=seen_names)) |
| 81 | seen_names = {entry["name"] for entry in skills} |
| 82 | skills.extend(self._skill_entries_from_dir(self.workspace_skills, "workspace", skip_names=seen_names)) |
| 83 | workspace_names = {entry["name"] for entry in skills} |
| 84 | if self.extra_skills and self.extra_skills.exists(): |
| 85 | skills.extend( |
| 86 | self._skill_entries_from_dir(self.extra_skills, "extra", skip_names=workspace_names) |
| 87 | ) |
| 88 | known_names = {entry["name"] for entry in skills} |
| 89 | if self.builtin_skills and self.builtin_skills.exists(): |
| 90 | skills.extend( |
| 91 | self._skill_entries_from_dir(self.builtin_skills, "builtin", skip_names=known_names) |
| 92 | ) |
| 93 | |
| 94 | if self.disabled_skills: |
| 95 | skills = [s for s in skills if s["name"] not in self.disabled_skills] |
| 96 | |
| 97 | if filter_unavailable: |
| 98 | return [skill for skill in skills if self._check_requirements(self._get_skill_meta(skill["name"]))] |
| 99 | return skills |
| 100 | |
| 101 | def load_skill(self, name: str) -> str | None: |
| 102 | """ |
| 103 | Load a skill by name. |
| 104 | |
| 105 | Args: |
| 106 | name: Skill name (directory name). |
| 107 | |
| 108 | Returns: |
| 109 | Skill content or None if not found. |
| 110 | """ |
| 111 | roots = list(self._pe_skill_dirs()) |
| 112 | roots.append(self.workspace_skills) |
| 113 | if self.extra_skills: |
| 114 | roots.append(self.extra_skills) |
| 115 | if self.builtin_skills: |
| 116 | roots.append(self.builtin_skills) |
| 117 | for root in roots: |
| 118 | path = root / name / "SKILL.md" |
| 119 | if path.exists(): |
| 120 | return path.read_text(encoding="utf-8") |
| 121 | return None |
| 122 | |
| 123 | def load_skills_for_context(self, skill_names: list[str]) -> str: |
| 124 | """ |
| 125 | Load specific skills for inclusion in agent context. |
| 126 | |
| 127 | Args: |
| 128 | skill_names: List of skill names to load. |
| 129 | |
| 130 | Returns: |
| 131 | Formatted skills content. |
| 132 | """ |
| 133 | parts = [ |
| 134 | f"### Skill: {name}\n\n{self._strip_frontmatter(markdown)}" |
| 135 | for name in skill_names |
| 136 | if (markdown := self.load_skill(name)) |
| 137 | ] |
| 138 | return "\n\n---\n\n".join(parts) |
| 139 | |
| 140 | def build_skills_summary(self, exclude: set[str] | None = None) -> str: |
| 141 | """ |
| 142 | Build a summary of all skills (name, description, path, availability). |
| 143 | |
| 144 | This is used for progressive loading - the agent can read the full |
| 145 | skill content using read_file when needed. |
| 146 | |
| 147 | Args: |
| 148 | exclude: Set of skill names to omit from the summary. |
| 149 | |
| 150 | Returns: |
| 151 | Markdown-formatted skills summary. |
| 152 | """ |
| 153 | all_skills = self.list_skills(filter_unavailable=False) |
| 154 | if not all_skills: |
| 155 | return "" |
| 156 | |
| 157 | lines: list[str] = [] |
| 158 | for entry in all_skills: |
| 159 | skill_name = entry["name"] |
| 160 | if exclude and skill_name in exclude: |
| 161 | continue |
| 162 | meta = self._get_skill_meta(skill_name) |
| 163 | available = self._check_requirements(meta) |
| 164 | desc = self._get_skill_description(skill_name) |
| 165 | if available: |
| 166 | lines.append(f"- **{skill_name}** — {desc} `{entry['path']}`") |
| 167 | else: |
| 168 | missing = self._get_missing_requirements(meta) |
| 169 | suffix = f" (unavailable: {missing})" if missing else " (unavailable)" |
| 170 | lines.append(f"- **{skill_name}** — {desc}{suffix} `{entry['path']}`") |
| 171 | return "\n".join(lines) |
| 172 | |
| 173 | def _get_missing_requirements(self, skill_meta: dict) -> str: |
| 174 | """Get a description of missing requirements.""" |
| 175 | requires = skill_meta.get("requires", {}) |
| 176 | required_bins = requires.get("bins", []) |
| 177 | required_env_vars = requires.get("env", []) |
| 178 | return ", ".join( |
| 179 | [f"CLI: {command_name}" for command_name in required_bins if not shutil.which(command_name)] |
| 180 | + [f"ENV: {env_name}" for env_name in required_env_vars if not os.environ.get(env_name)] |
| 181 | ) |
| 182 | |
| 183 | def _get_skill_description(self, name: str) -> str: |
| 184 | """Get the description of a skill from its frontmatter.""" |
| 185 | meta = self.get_skill_metadata(name) |
| 186 | if meta and meta.get("description"): |
| 187 | return meta["description"] |
| 188 | return name # Fallback to skill name |
| 189 | |
| 190 | def _strip_frontmatter(self, content: str) -> str: |
| 191 | """Remove YAML frontmatter from markdown content.""" |
| 192 | if not content.startswith("---"): |
| 193 | return content |
| 194 | match = _STRIP_SKILL_FRONTMATTER.match(content) |
| 195 | if match: |
| 196 | return content[match.end():].strip() |
| 197 | return content |
| 198 | |
| 199 | def _parse_nanobot_metadata(self, raw: object) -> dict: |
| 200 | """Extract nanobot/openclaw metadata from a frontmatter field. |
| 201 | |
| 202 | ``raw`` may be a dict (already parsed by yaml.safe_load) or a JSON str. |
| 203 | """ |
| 204 | if isinstance(raw, dict): |
| 205 | data = raw |
| 206 | elif isinstance(raw, str): |
| 207 | try: |
| 208 | data = json.loads(raw) |
| 209 | except (json.JSONDecodeError, TypeError): |
| 210 | return {} |
| 211 | else: |
| 212 | return {} |
| 213 | if not isinstance(data, dict): |
| 214 | return {} |
| 215 | payload = data.get("nanobot", data.get("openclaw", {})) |
| 216 | return payload if isinstance(payload, dict) else {} |
| 217 | |
| 218 | def _check_requirements(self, skill_meta: dict) -> bool: |
| 219 | """Check if skill requirements are met (bins, env vars).""" |
| 220 | requires = skill_meta.get("requires", {}) |
| 221 | required_bins = requires.get("bins", []) |
| 222 | required_env_vars = requires.get("env", []) |
| 223 | return all(shutil.which(cmd) for cmd in required_bins) and all( |
| 224 | os.environ.get(var) for var in required_env_vars |
| 225 | ) |
| 226 | |
| 227 | def _get_skill_meta(self, name: str) -> dict: |
| 228 | """Get nanobot metadata for a skill (cached in frontmatter).""" |
| 229 | raw_meta = self.get_skill_metadata(name) or {} |
| 230 | return self._parse_nanobot_metadata(raw_meta.get("metadata")) |
| 231 | |
| 232 | def get_always_skills(self) -> list[str]: |
| 233 | """Get skills marked as always=true that meet requirements.""" |
| 234 | return [ |
| 235 | entry["name"] |
| 236 | for entry in self.list_skills(filter_unavailable=True) |
| 237 | if (meta := self.get_skill_metadata(entry["name"]) or {}) |
| 238 | and ( |
| 239 | self._parse_nanobot_metadata(meta.get("metadata")).get("always") |
| 240 | or meta.get("always") |
| 241 | ) |
| 242 | ] |
| 243 | |
| 244 | def get_skill_metadata(self, name: str) -> dict | None: |
| 245 | """ |
| 246 | Get metadata from a skill's frontmatter. |
| 247 | |
| 248 | Args: |
| 249 | name: Skill name. |
| 250 | |
| 251 | Returns: |
| 252 | Metadata dict or None. |
| 253 | """ |
| 254 | content = self.load_skill(name) |
| 255 | if not content or not content.startswith("---"): |
| 256 | return None |
| 257 | match = _STRIP_SKILL_FRONTMATTER.match(content) |
| 258 | if not match: |
| 259 | return None |
| 260 | try: |
| 261 | parsed = yaml.safe_load(match.group(1)) |
| 262 | except yaml.YAMLError: |
| 263 | return None |
| 264 | if not isinstance(parsed, dict): |
| 265 | return None |
| 266 | # yaml.safe_load returns native types (int, bool, list, etc.); |
| 267 | # keep values as-is so downstream consumers get correct types. |
| 268 | metadata: dict[str, object] = {} |
| 269 | for key, value in parsed.items(): |
| 270 | metadata[str(key)] = value |
| 271 | return metadata |
| 272 |