skills_tool.py
1 #!/usr/bin/env python3 2 """ 3 Skills Tool Module 4 5 This module provides tools for listing and viewing skill documents. 6 Skills are organized as directories containing a SKILL.md file (the main instructions) 7 and optional supporting files like references, templates, and examples. 8 9 Inspired by Anthropic's Claude Skills system with progressive disclosure architecture: 10 - Metadata (name ≤64 chars, description ≤1024 chars) - shown in skills_list 11 - Full Instructions - loaded via skill_view when needed 12 - Linked Files (references, templates) - loaded on demand 13 14 Directory Structure: 15 skills/ 16 ├── my-skill/ 17 │ ├── SKILL.md # Main instructions (required) 18 │ ├── references/ # Supporting documentation 19 │ │ ├── api.md 20 │ │ └── examples.md 21 │ ├── templates/ # Templates for output 22 │ │ └── template.md 23 │ └── assets/ # Supplementary files (agentskills.io standard) 24 └── category/ # Category folder for organization 25 └── another-skill/ 26 └── SKILL.md 27 28 SKILL.md Format (YAML Frontmatter, agentskills.io compatible): 29 --- 30 name: skill-name # Required, max 64 chars 31 description: Brief description # Required, max 1024 chars 32 version: 1.0.0 # Optional 33 license: MIT # Optional (agentskills.io) 34 platforms: [macos] # Optional — restrict to specific OS platforms 35 # Valid: macos, linux, windows 36 # Omit to load on all platforms (default) 37 prerequisites: # Optional — legacy runtime requirements 38 env_vars: [API_KEY] # Legacy env var names are normalized into 39 # required_environment_variables on load. 40 commands: [curl, jq] # Command checks remain advisory only. 41 compatibility: Requires X # Optional (agentskills.io) 42 metadata: # Optional, arbitrary key-value (agentskills.io) 43 hermes: 44 tags: [fine-tuning, llm] 45 related_skills: [peft, lora] 46 --- 47 48 # Skill Title 49 50 Full instructions and content here... 51 52 Available tools: 53 - skills_list: List skills with metadata (progressive disclosure tier 1) 54 - skill_view: Load full skill content (progressive disclosure tier 2-3) 55 56 Usage: 57 from tools.skills_tool import skills_list, skill_view, check_skills_requirements 58 59 # List all skills (returns metadata only - token efficient) 60 result = skills_list() 61 62 # View a skill's main content (loads full instructions) 63 content = skill_view("axolotl") 64 65 # View a reference file within a skill (loads linked file) 66 content = skill_view("axolotl", "references/dataset-formats.md") 67 """ 68 69 import json 70 import logging 71 72 from hermes_constants import get_hermes_home, display_hermes_home 73 import os 74 import re 75 from enum import Enum 76 from pathlib import Path 77 from typing import Dict, Any, List, Optional, Set, Tuple 78 79 from tools.registry import registry, tool_error 80 from hermes_cli.config import cfg_get 81 82 logger = logging.getLogger(__name__) 83 84 85 # All skills live in ~/.hermes/skills/ (seeded from bundled skills/ on install). 86 # This is the single source of truth -- agent edits, hub installs, and bundled 87 # skills all coexist here without polluting the git repo. 88 HERMES_HOME = get_hermes_home() 89 SKILLS_DIR = HERMES_HOME / "skills" 90 91 # Anthropic-recommended limits for progressive disclosure efficiency 92 MAX_NAME_LENGTH = 64 93 MAX_DESCRIPTION_LENGTH = 1024 94 95 # Platform identifiers for the 'platforms' frontmatter field. 96 # Maps user-friendly names to sys.platform prefixes. 97 _PLATFORM_MAP = { 98 "macos": "darwin", 99 "linux": "linux", 100 "windows": "win32", 101 } 102 _ENV_VAR_NAME_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$") 103 _EXCLUDED_SKILL_DIRS = frozenset((".git", ".github", ".hub", ".archive")) 104 _REMOTE_ENV_BACKENDS = frozenset( 105 {"docker", "singularity", "modal", "ssh", "daytona", "vercel_sandbox"} 106 ) 107 _secret_capture_callback = None 108 109 110 def load_env() -> Dict[str, str]: 111 """Load profile-scoped environment variables from HERMES_HOME/.env.""" 112 env_path = get_hermes_home() / ".env" 113 env_vars: Dict[str, str] = {} 114 if not env_path.exists(): 115 return env_vars 116 117 with env_path.open(encoding="utf-8") as f: 118 for line in f: 119 line = line.strip() 120 if line and not line.startswith("#") and "=" in line: 121 key, _, value = line.partition("=") 122 env_vars[key.strip()] = value.strip().strip("\"'") 123 return env_vars 124 125 126 class SkillReadinessStatus(str, Enum): 127 AVAILABLE = "available" 128 SETUP_NEEDED = "setup_needed" 129 UNSUPPORTED = "unsupported" 130 131 132 # Prompt injection detection — shared by local-skill and plugin-skill paths. 133 _INJECTION_PATTERNS: list = [ 134 "ignore previous instructions", 135 "ignore all previous", 136 "you are now", 137 "disregard your", 138 "forget your instructions", 139 "new instructions:", 140 "system prompt:", 141 "<system>", 142 "]]>", 143 ] 144 145 146 def set_secret_capture_callback(callback) -> None: 147 global _secret_capture_callback 148 _secret_capture_callback = callback 149 150 151 def skill_matches_platform(frontmatter: Dict[str, Any]) -> bool: 152 """Check if a skill is compatible with the current OS platform. 153 154 Delegates to ``agent.skill_utils.skill_matches_platform`` — kept here 155 as a public re-export so existing callers don't need updating. 156 """ 157 from agent.skill_utils import skill_matches_platform as _impl 158 return _impl(frontmatter) 159 160 161 def _normalize_prerequisite_values(value: Any) -> List[str]: 162 if not value: 163 return [] 164 if isinstance(value, str): 165 value = [value] 166 return [str(item) for item in value if str(item).strip()] 167 168 169 def _collect_prerequisite_values( 170 frontmatter: Dict[str, Any], 171 ) -> Tuple[List[str], List[str]]: 172 prereqs = frontmatter.get("prerequisites") 173 if not prereqs or not isinstance(prereqs, dict): 174 return [], [] 175 return ( 176 _normalize_prerequisite_values(prereqs.get("env_vars")), 177 _normalize_prerequisite_values(prereqs.get("commands")), 178 ) 179 180 181 def _normalize_setup_metadata(frontmatter: Dict[str, Any]) -> Dict[str, Any]: 182 setup = frontmatter.get("setup") 183 if not isinstance(setup, dict): 184 return {"help": None, "collect_secrets": []} 185 186 help_text = setup.get("help") 187 normalized_help = ( 188 str(help_text).strip() 189 if isinstance(help_text, str) and help_text.strip() 190 else None 191 ) 192 193 collect_secrets_raw = setup.get("collect_secrets") 194 if isinstance(collect_secrets_raw, dict): 195 collect_secrets_raw = [collect_secrets_raw] 196 if not isinstance(collect_secrets_raw, list): 197 collect_secrets_raw = [] 198 199 collect_secrets: List[Dict[str, Any]] = [] 200 for item in collect_secrets_raw: 201 if not isinstance(item, dict): 202 continue 203 204 env_var = str(item.get("env_var") or "").strip() 205 if not env_var: 206 continue 207 208 prompt = str(item.get("prompt") or f"Enter value for {env_var}").strip() 209 provider_url = str(item.get("provider_url") or item.get("url") or "").strip() 210 211 entry: Dict[str, Any] = { 212 "env_var": env_var, 213 "prompt": prompt, 214 "secret": bool(item.get("secret", True)), 215 } 216 if provider_url: 217 entry["provider_url"] = provider_url 218 collect_secrets.append(entry) 219 220 return { 221 "help": normalized_help, 222 "collect_secrets": collect_secrets, 223 } 224 225 226 def _get_required_environment_variables( 227 frontmatter: Dict[str, Any], 228 legacy_env_vars: List[str] | None = None, 229 ) -> List[Dict[str, Any]]: 230 setup = _normalize_setup_metadata(frontmatter) 231 required_raw = frontmatter.get("required_environment_variables") 232 if isinstance(required_raw, dict): 233 required_raw = [required_raw] 234 if not isinstance(required_raw, list): 235 required_raw = [] 236 237 required: List[Dict[str, Any]] = [] 238 seen: set[str] = set() 239 240 def _append_required(entry: Dict[str, Any]) -> None: 241 env_name = str(entry.get("name") or entry.get("env_var") or "").strip() 242 if not env_name or env_name in seen: 243 return 244 if not _ENV_VAR_NAME_RE.match(env_name): 245 return 246 247 normalized: Dict[str, Any] = { 248 "name": env_name, 249 "prompt": str(entry.get("prompt") or f"Enter value for {env_name}").strip(), 250 } 251 252 help_text = ( 253 entry.get("help") 254 or entry.get("provider_url") 255 or entry.get("url") 256 or setup.get("help") 257 ) 258 if isinstance(help_text, str) and help_text.strip(): 259 normalized["help"] = help_text.strip() 260 261 required_for = entry.get("required_for") 262 if isinstance(required_for, str) and required_for.strip(): 263 normalized["required_for"] = required_for.strip() 264 265 if entry.get("optional"): 266 normalized["optional"] = True 267 268 seen.add(env_name) 269 required.append(normalized) 270 271 for item in required_raw: 272 if isinstance(item, str): 273 _append_required({"name": item}) 274 continue 275 if isinstance(item, dict): 276 _append_required(item) 277 278 for item in setup["collect_secrets"]: 279 _append_required( 280 { 281 "name": item.get("env_var"), 282 "prompt": item.get("prompt"), 283 "help": item.get("provider_url") or setup.get("help"), 284 } 285 ) 286 287 if legacy_env_vars is None: 288 legacy_env_vars, _ = _collect_prerequisite_values(frontmatter) 289 for env_var in legacy_env_vars: 290 _append_required({"name": env_var}) 291 292 return required 293 294 295 def _capture_required_environment_variables( 296 skill_name: str, 297 missing_entries: List[Dict[str, Any]], 298 ) -> Dict[str, Any]: 299 if not missing_entries: 300 return { 301 "missing_names": [], 302 "setup_skipped": False, 303 "gateway_setup_hint": None, 304 } 305 306 missing_names = [entry["name"] for entry in missing_entries] 307 if _is_gateway_surface(): 308 return { 309 "missing_names": missing_names, 310 "setup_skipped": False, 311 "gateway_setup_hint": _gateway_setup_hint(), 312 } 313 314 if _secret_capture_callback is None: 315 return { 316 "missing_names": missing_names, 317 "setup_skipped": False, 318 "gateway_setup_hint": None, 319 } 320 321 setup_skipped = False 322 remaining_names: List[str] = [] 323 324 for entry in missing_entries: 325 metadata = {"skill_name": skill_name} 326 if entry.get("help"): 327 metadata["help"] = entry["help"] 328 if entry.get("required_for"): 329 metadata["required_for"] = entry["required_for"] 330 331 try: 332 callback_result = _secret_capture_callback( 333 entry["name"], 334 entry["prompt"], 335 metadata, 336 ) 337 except Exception: 338 logger.warning( 339 f"Secret capture callback failed for {entry['name']}", exc_info=True 340 ) 341 callback_result = { 342 "success": False, 343 "stored_as": entry["name"], 344 "validated": False, 345 "skipped": True, 346 } 347 348 success = isinstance(callback_result, dict) and bool( 349 callback_result.get("success") 350 ) 351 skipped = isinstance(callback_result, dict) and bool( 352 callback_result.get("skipped") 353 ) 354 if success and not skipped: 355 continue 356 357 setup_skipped = True 358 remaining_names.append(entry["name"]) 359 360 return { 361 "missing_names": remaining_names, 362 "setup_skipped": setup_skipped, 363 "gateway_setup_hint": None, 364 } 365 366 367 def _is_gateway_surface() -> bool: 368 if os.getenv("HERMES_GATEWAY_SESSION"): 369 return True 370 from gateway.session_context import get_session_env 371 return bool(get_session_env("HERMES_SESSION_PLATFORM")) 372 373 374 def _get_terminal_backend_name() -> str: 375 return str(os.getenv("TERMINAL_ENV", "local")).strip().lower() or "local" 376 377 378 def _is_env_var_persisted( 379 var_name: str, env_snapshot: Dict[str, str] | None = None 380 ) -> bool: 381 if env_snapshot is None: 382 env_snapshot = load_env() 383 if var_name in env_snapshot: 384 return bool(env_snapshot.get(var_name)) 385 return bool(os.getenv(var_name)) 386 387 388 def _remaining_required_environment_names( 389 required_env_vars: List[Dict[str, Any]], 390 capture_result: Dict[str, Any], 391 *, 392 env_snapshot: Dict[str, str] | None = None, 393 ) -> List[str]: 394 missing_names = set(capture_result["missing_names"]) 395 396 if env_snapshot is None: 397 env_snapshot = load_env() 398 remaining = [] 399 for entry in required_env_vars: 400 name = entry["name"] 401 if entry.get("optional"): 402 continue 403 if name in missing_names or not _is_env_var_persisted(name, env_snapshot): 404 remaining.append(name) 405 return remaining 406 407 408 def _gateway_setup_hint() -> str: 409 try: 410 from gateway.platforms.base import GATEWAY_SECRET_CAPTURE_UNSUPPORTED_MESSAGE 411 412 return GATEWAY_SECRET_CAPTURE_UNSUPPORTED_MESSAGE 413 except Exception: 414 return f"Secure secret entry is not available. Load this skill in the local CLI to be prompted, or add the key to {display_hermes_home()}/.env manually." 415 416 417 def _build_setup_note( 418 readiness_status: SkillReadinessStatus, 419 missing: List[str], 420 setup_help: str | None = None, 421 ) -> str | None: 422 if readiness_status == SkillReadinessStatus.SETUP_NEEDED: 423 missing_str = ", ".join(missing) if missing else "required prerequisites" 424 note = f"Setup needed before using this skill: missing {missing_str}." 425 if setup_help: 426 return f"{note} {setup_help}" 427 return note 428 return None 429 430 431 def check_skills_requirements() -> bool: 432 """Skills are always available -- the directory is created on first use if needed.""" 433 return True 434 435 436 def _parse_frontmatter(content: str) -> Tuple[Dict[str, Any], str]: 437 """Parse YAML frontmatter from markdown content. 438 439 Delegates to ``agent.skill_utils.parse_frontmatter`` — kept here 440 as a public re-export so existing callers don't need updating. 441 """ 442 from agent.skill_utils import parse_frontmatter 443 return parse_frontmatter(content) 444 445 446 def _get_category_from_path(skill_path: Path) -> Optional[str]: 447 """ 448 Extract category from skill path based on directory structure. 449 450 For paths like: ~/.hermes/skills/mlops/axolotl/SKILL.md -> "mlops" 451 Also works for external skill dirs configured via skills.external_dirs. 452 """ 453 # Try the module-level SKILLS_DIR first (respects monkeypatching in tests), 454 # then fall back to external dirs from config. 455 dirs_to_check = [SKILLS_DIR] 456 try: 457 from agent.skill_utils import get_external_skills_dirs 458 dirs_to_check.extend(get_external_skills_dirs()) 459 except Exception: 460 pass 461 for skills_dir in dirs_to_check: 462 try: 463 rel_path = skill_path.relative_to(skills_dir) 464 parts = rel_path.parts 465 if len(parts) >= 3: 466 return parts[0] 467 except ValueError: 468 continue 469 return None 470 471 472 def _parse_tags(tags_value) -> List[str]: 473 """ 474 Parse tags from frontmatter value. 475 476 Handles: 477 - Already-parsed list (from yaml.safe_load): [tag1, tag2] 478 - String with brackets: "[tag1, tag2]" 479 - Comma-separated string: "tag1, tag2" 480 481 Args: 482 tags_value: Raw tags value — may be a list or string 483 484 Returns: 485 List of tag strings 486 """ 487 if not tags_value: 488 return [] 489 490 # yaml.safe_load already returns a list for [tag1, tag2] 491 if isinstance(tags_value, list): 492 return [str(t).strip() for t in tags_value if t] 493 494 # String fallback — handle bracket-wrapped or comma-separated 495 tags_value = str(tags_value).strip() 496 if tags_value.startswith("[") and tags_value.endswith("]"): 497 tags_value = tags_value[1:-1] 498 499 return [t.strip().strip("\"'") for t in tags_value.split(",") if t.strip()] 500 501 502 503 def _get_disabled_skill_names() -> Set[str]: 504 """Load disabled skill names from config. 505 506 Delegates to ``agent.skill_utils.get_disabled_skill_names`` — kept here 507 as a public re-export so existing callers don't need updating. 508 """ 509 from agent.skill_utils import get_disabled_skill_names 510 return get_disabled_skill_names() 511 512 513 def _get_session_platform() -> str: 514 """Resolve the current platform from gateway session context. 515 516 Mirrors the platform-resolution logic in 517 ``agent.skill_utils.get_disabled_skill_names`` so that 518 ``_is_skill_disabled`` respects ``HERMES_SESSION_PLATFORM``. 519 """ 520 try: 521 from gateway.session_context import get_session_env 522 return get_session_env("HERMES_SESSION_PLATFORM") or "" 523 except Exception: 524 return "" 525 526 527 def _is_skill_disabled(name: str, platform: str = None) -> bool: 528 """Check if a skill is disabled in config. 529 530 Resolves the active platform from (in order of precedence): 531 1. Explicit ``platform`` argument 532 2. ``HERMES_PLATFORM`` environment variable 533 3. ``HERMES_SESSION_PLATFORM`` from gateway session context 534 """ 535 try: 536 from hermes_cli.config import load_config 537 config = load_config() 538 skills_cfg = config.get("skills", {}) 539 resolved_platform = platform or os.getenv("HERMES_PLATFORM") or _get_session_platform() 540 if resolved_platform: 541 platform_disabled = cfg_get(skills_cfg, "platform_disabled", resolved_platform) 542 if platform_disabled is not None: 543 return name in platform_disabled 544 return name in skills_cfg.get("disabled", []) 545 except Exception: 546 return False 547 548 549 def _find_all_skills(*, skip_disabled: bool = False) -> List[Dict[str, Any]]: 550 """Recursively find all skills in ~/.hermes/skills/ and external dirs. 551 552 Args: 553 skip_disabled: If True, return ALL skills regardless of disabled 554 state (used by ``hermes skills`` config UI). Default False 555 filters out disabled skills. 556 557 Returns: 558 List of skill metadata dicts (name, description, category). 559 """ 560 from agent.skill_utils import get_external_skills_dirs, iter_skill_index_files 561 562 skills = [] 563 seen_names: set = set() 564 565 # Load disabled set once (not per-skill) 566 disabled = set() if skip_disabled else _get_disabled_skill_names() 567 568 # Scan local dir first, then external dirs (local takes precedence) 569 dirs_to_scan = [] 570 if SKILLS_DIR.exists(): 571 dirs_to_scan.append(SKILLS_DIR) 572 dirs_to_scan.extend(get_external_skills_dirs()) 573 574 for scan_dir in dirs_to_scan: 575 for skill_md in iter_skill_index_files(scan_dir, "SKILL.md"): 576 if any(part in _EXCLUDED_SKILL_DIRS for part in skill_md.parts): 577 continue 578 579 skill_dir = skill_md.parent 580 581 try: 582 content = skill_md.read_text(encoding="utf-8")[:4000] 583 frontmatter, body = _parse_frontmatter(content) 584 585 if not skill_matches_platform(frontmatter): 586 continue 587 588 name = frontmatter.get("name", skill_dir.name)[:MAX_NAME_LENGTH] 589 if name in seen_names: 590 continue 591 if name in disabled: 592 continue 593 594 description = frontmatter.get("description", "") 595 if not description: 596 for line in body.strip().split("\n"): 597 line = line.strip() 598 if line and not line.startswith("#"): 599 description = line 600 break 601 602 if len(description) > MAX_DESCRIPTION_LENGTH: 603 description = description[:MAX_DESCRIPTION_LENGTH - 3] + "..." 604 605 category = _get_category_from_path(skill_md) 606 607 seen_names.add(name) 608 skills.append({ 609 "name": name, 610 "description": description, 611 "category": category, 612 }) 613 614 except (UnicodeDecodeError, PermissionError) as e: 615 logger.debug("Failed to read skill file %s: %s", skill_md, e) 616 continue 617 except Exception as e: 618 logger.debug( 619 "Skipping skill at %s: failed to parse: %s", skill_md, e, exc_info=True 620 ) 621 continue 622 623 return skills 624 625 626 def _sort_skills(skills: List[Dict[str, Any]]) -> List[Dict[str, Any]]: 627 """Keep every skill listing path ordered the same way.""" 628 return sorted(skills, key=lambda s: (s.get("category") or "", s["name"])) 629 630 631 def _load_category_description(category_dir: Path) -> Optional[str]: 632 """ 633 Load category description from DESCRIPTION.md if it exists. 634 635 Args: 636 category_dir: Path to the category directory 637 638 Returns: 639 Description string or None if not found 640 """ 641 desc_file = category_dir / "DESCRIPTION.md" 642 if not desc_file.exists(): 643 return None 644 645 try: 646 content = desc_file.read_text(encoding="utf-8") 647 # Parse frontmatter if present 648 frontmatter, body = _parse_frontmatter(content) 649 650 # Prefer frontmatter description, fall back to first non-header line 651 description = frontmatter.get("description", "") 652 if not description: 653 for line in body.strip().split("\n"): 654 line = line.strip() 655 if line and not line.startswith("#"): 656 description = line 657 break 658 659 # Truncate to reasonable length 660 if len(description) > MAX_DESCRIPTION_LENGTH: 661 description = description[: MAX_DESCRIPTION_LENGTH - 3] + "..." 662 663 return description if description else None 664 except (UnicodeDecodeError, PermissionError) as e: 665 logger.debug("Failed to read category description %s: %s", desc_file, e) 666 return None 667 except Exception as e: 668 logger.warning( 669 "Error parsing category description %s: %s", desc_file, e, exc_info=True 670 ) 671 return None 672 673 674 def skills_list(category: str = None, task_id: str = None) -> str: 675 """ 676 List all available skills (progressive disclosure tier 1 - minimal metadata). 677 678 Returns only name + description to minimize token usage. Use skill_view() to 679 load full content, tags, related files, etc. 680 681 Args: 682 category: Optional category filter (e.g., "mlops") 683 task_id: Optional task identifier used to probe the active backend 684 685 Returns: 686 JSON string with minimal skill info: name, description, category 687 """ 688 try: 689 if not SKILLS_DIR.exists(): 690 SKILLS_DIR.mkdir(parents=True, exist_ok=True) 691 return json.dumps( 692 { 693 "success": True, 694 "skills": [], 695 "categories": [], 696 "message": f"No skills found. Skills directory created at {display_hermes_home()}/skills/", 697 }, 698 ensure_ascii=False, 699 ) 700 701 # Find all skills 702 all_skills = _find_all_skills() 703 704 if not all_skills: 705 return json.dumps( 706 { 707 "success": True, 708 "skills": [], 709 "categories": [], 710 "message": "No skills found in skills/ directory.", 711 }, 712 ensure_ascii=False, 713 ) 714 715 # Filter by category if specified 716 if category: 717 all_skills = [s for s in all_skills if s.get("category") == category] 718 719 # Sort by category then name 720 all_skills = _sort_skills(all_skills) 721 722 # Extract unique categories 723 categories = sorted( 724 set(s.get("category") for s in all_skills if s.get("category")) 725 ) 726 727 return json.dumps( 728 { 729 "success": True, 730 "skills": all_skills, 731 "categories": categories, 732 "count": len(all_skills), 733 "hint": "Use skill_view(name) to see full content, tags, and linked files", 734 }, 735 ensure_ascii=False, 736 ) 737 738 except Exception as e: 739 return tool_error(str(e), success=False) 740 741 742 # ── Plugin skill serving ────────────────────────────────────────────────── 743 744 745 def _serve_plugin_skill( 746 skill_md: Path, 747 namespace: str, 748 bare: str, 749 *, 750 preprocess: bool = True, 751 session_id: str | None = None, 752 ) -> str: 753 """Read a plugin-provided skill, apply guards, return JSON.""" 754 from hermes_cli.plugins import _get_disabled_plugins, get_plugin_manager 755 756 if namespace in _get_disabled_plugins(): 757 return json.dumps( 758 { 759 "success": False, 760 "error": ( 761 f"Plugin '{namespace}' is disabled. " 762 f"Re-enable with: hermes plugins enable {namespace}" 763 ), 764 }, 765 ensure_ascii=False, 766 ) 767 768 try: 769 content = skill_md.read_text(encoding="utf-8") 770 except Exception as e: 771 return json.dumps( 772 {"success": False, "error": f"Failed to read skill '{namespace}:{bare}': {e}"}, 773 ensure_ascii=False, 774 ) 775 776 parsed_frontmatter: Dict[str, Any] = {} 777 try: 778 parsed_frontmatter, _ = _parse_frontmatter(content) 779 except Exception: 780 pass 781 782 if not skill_matches_platform(parsed_frontmatter): 783 return json.dumps( 784 { 785 "success": False, 786 "error": f"Skill '{namespace}:{bare}' is not supported on this platform.", 787 "readiness_status": SkillReadinessStatus.UNSUPPORTED.value, 788 }, 789 ensure_ascii=False, 790 ) 791 792 # Injection scan — log but still serve (matches local-skill behaviour) 793 if any(p in content.lower() for p in _INJECTION_PATTERNS): 794 logger.warning( 795 "Plugin skill '%s:%s' contains patterns that may indicate prompt injection", 796 namespace, bare, 797 ) 798 799 description = str(parsed_frontmatter.get("description", "")) 800 if len(description) > MAX_DESCRIPTION_LENGTH: 801 description = description[: MAX_DESCRIPTION_LENGTH - 3] + "..." 802 803 # Bundle context banner — tells the agent about sibling skills 804 try: 805 siblings = [ 806 s for s in get_plugin_manager().list_plugin_skills(namespace) 807 if s != bare 808 ] 809 if siblings: 810 sib_list = ", ".join(siblings) 811 banner = ( 812 f"[Bundle context: This skill is part of the '{namespace}' plugin.\n" 813 f"Sibling skills: {sib_list}.\n" 814 f"Use qualified form to invoke siblings (e.g. {namespace}:{siblings[0]}).]\n\n" 815 ) 816 else: 817 banner = f"[Bundle context: This skill is part of the '{namespace}' plugin.]\n\n" 818 except Exception: 819 banner = "" 820 821 rendered_content = content 822 if preprocess: 823 try: 824 from agent.skill_preprocessing import preprocess_skill_content 825 826 rendered_content = preprocess_skill_content( 827 content, 828 skill_md.parent, 829 session_id=session_id, 830 ) 831 except Exception: 832 logger.debug( 833 "Could not preprocess plugin skill %s:%s", namespace, bare, exc_info=True 834 ) 835 836 return json.dumps( 837 { 838 "success": True, 839 "name": f"{namespace}:{bare}", 840 "content": f"{banner}{rendered_content}" if banner else rendered_content, 841 "description": description, 842 "linked_files": None, 843 "readiness_status": SkillReadinessStatus.AVAILABLE.value, 844 }, 845 ensure_ascii=False, 846 ) 847 848 849 def skill_view( 850 name: str, 851 file_path: str = None, 852 task_id: str = None, 853 preprocess: bool = True, 854 ) -> str: 855 """ 856 View the content of a skill or a specific file within a skill directory. 857 858 Args: 859 name: Name or path of the skill (e.g., "axolotl" or "03-fine-tuning/axolotl"). 860 Qualified names like "plugin:skill" resolve to plugin-provided skills. 861 file_path: Optional path to a specific file within the skill (e.g., "references/api.md") 862 task_id: Optional task identifier used to probe the active backend 863 preprocess: Apply configured SKILL.md template and inline shell rendering 864 to main skill content. Internal slash/preload callers disable this 865 because they render the skill message themselves. 866 867 Returns: 868 JSON string with skill content or error message 869 """ 870 try: 871 # ── Qualified name dispatch (plugin skills) ────────────────── 872 # Names containing ':' are routed to the plugin skill registry. 873 # Bare names fall through to the existing flat-tree scan below. 874 if ":" in name: 875 from agent.skill_utils import is_valid_namespace, parse_qualified_name 876 from hermes_cli.plugins import discover_plugins, get_plugin_manager 877 878 namespace, bare = parse_qualified_name(name) 879 if not is_valid_namespace(namespace): 880 return json.dumps( 881 { 882 "success": False, 883 "error": ( 884 f"Invalid namespace '{namespace}' in '{name}'. " 885 f"Namespaces must match [a-zA-Z0-9_-]+." 886 ), 887 }, 888 ensure_ascii=False, 889 ) 890 891 discover_plugins() # idempotent 892 pm = get_plugin_manager() 893 plugin_skill_md = pm.find_plugin_skill(name) 894 895 if plugin_skill_md is not None: 896 if not plugin_skill_md.exists(): 897 # Stale registry entry — file deleted out of band 898 pm.remove_plugin_skill(name) 899 return json.dumps( 900 { 901 "success": False, 902 "error": ( 903 f"Skill '{name}' file no longer exists at " 904 f"{plugin_skill_md}. The registry entry has " 905 f"been cleaned up — try again after the " 906 f"plugin is reloaded." 907 ), 908 }, 909 ensure_ascii=False, 910 ) 911 return _serve_plugin_skill( 912 plugin_skill_md, 913 namespace, 914 bare, 915 preprocess=preprocess, 916 session_id=task_id, 917 ) 918 919 # Plugin exists but this specific skill is missing? 920 available = pm.list_plugin_skills(namespace) 921 if available: 922 return json.dumps( 923 { 924 "success": False, 925 "error": f"Skill '{bare}' not found in plugin '{namespace}'.", 926 "available_skills": [f"{namespace}:{s}" for s in available], 927 "hint": f"The '{namespace}' plugin provides {len(available)} skill(s).", 928 }, 929 ensure_ascii=False, 930 ) 931 # Plugin itself not found — fall through to flat-tree scan 932 # which will return a normal "not found" with suggestions. 933 934 from agent.skill_utils import get_external_skills_dirs 935 936 # Build list of all skill directories to search 937 all_dirs = [] 938 if SKILLS_DIR.exists(): 939 all_dirs.append(SKILLS_DIR) 940 all_dirs.extend(get_external_skills_dirs()) 941 942 if not all_dirs: 943 return json.dumps( 944 { 945 "success": False, 946 "error": "Skills directory does not exist yet. It will be created on first install.", 947 }, 948 ensure_ascii=False, 949 ) 950 951 skill_dir = None 952 skill_md = None 953 954 # Search all dirs: local first, then external (first match wins) 955 for search_dir in all_dirs: 956 # Try direct path first (e.g., "mlops/axolotl") 957 direct_path = search_dir / name 958 if direct_path.is_dir() and (direct_path / "SKILL.md").exists(): 959 skill_dir = direct_path 960 skill_md = direct_path / "SKILL.md" 961 break 962 elif direct_path.with_suffix(".md").exists(): 963 skill_md = direct_path.with_suffix(".md") 964 break 965 966 # Search by directory name across all dirs 967 if not skill_md: 968 for search_dir in all_dirs: 969 from agent.skill_utils import iter_skill_index_files 970 971 for found_skill_md in iter_skill_index_files(search_dir, "SKILL.md"): 972 if found_skill_md.parent.name == name: 973 skill_dir = found_skill_md.parent 974 skill_md = found_skill_md 975 break 976 if skill_md: 977 break 978 979 # Legacy: flat .md files 980 if not skill_md: 981 for search_dir in all_dirs: 982 for found_md in search_dir.rglob(f"{name}.md"): 983 if found_md.name != "SKILL.md": 984 skill_md = found_md 985 break 986 if skill_md: 987 break 988 989 if not skill_md or not skill_md.exists(): 990 available = [s["name"] for s in _sort_skills(_find_all_skills())[:20]] 991 return json.dumps( 992 { 993 "success": False, 994 "error": f"Skill '{name}' not found.", 995 "available_skills": available, 996 "hint": "Use skills_list to see all available skills", 997 }, 998 ensure_ascii=False, 999 ) 1000 1001 # Read the file once — reused for platform check and main content below 1002 try: 1003 content = skill_md.read_text(encoding="utf-8") 1004 except Exception as e: 1005 return json.dumps( 1006 { 1007 "success": False, 1008 "error": f"Failed to read skill '{name}': {e}", 1009 }, 1010 ensure_ascii=False, 1011 ) 1012 1013 # Security: warn if skill is loaded from outside trusted directories 1014 # (local skills dir + configured external_dirs are all trusted) 1015 _outside_skills_dir = True 1016 _trusted_dirs = [SKILLS_DIR.resolve()] 1017 try: 1018 _trusted_dirs.extend(d.resolve() for d in all_dirs[1:]) 1019 except Exception: 1020 pass 1021 for _td in _trusted_dirs: 1022 try: 1023 skill_md.resolve().relative_to(_td) 1024 _outside_skills_dir = False 1025 break 1026 except ValueError: 1027 continue 1028 1029 # Security: detect common prompt injection patterns 1030 # (pattern list at module level as _INJECTION_PATTERNS) 1031 _content_lower = content.lower() 1032 _injection_detected = any(p in _content_lower for p in _INJECTION_PATTERNS) 1033 1034 if _outside_skills_dir or _injection_detected: 1035 _warnings = [] 1036 if _outside_skills_dir: 1037 _warnings.append(f"skill file is outside the trusted skills directory (~/.hermes/skills/): {skill_md}") 1038 if _injection_detected: 1039 _warnings.append("skill content contains patterns that may indicate prompt injection") 1040 logging.getLogger(__name__).warning("Skill security warning for '%s': %s", name, "; ".join(_warnings)) 1041 1042 parsed_frontmatter: Dict[str, Any] = {} 1043 try: 1044 parsed_frontmatter, _ = _parse_frontmatter(content) 1045 except Exception: 1046 parsed_frontmatter = {} 1047 1048 if not skill_matches_platform(parsed_frontmatter): 1049 return json.dumps( 1050 { 1051 "success": False, 1052 "error": f"Skill '{name}' is not supported on this platform.", 1053 "readiness_status": SkillReadinessStatus.UNSUPPORTED.value, 1054 }, 1055 ensure_ascii=False, 1056 ) 1057 1058 # Check if the skill is disabled by the user 1059 resolved_name = parsed_frontmatter.get("name", skill_md.parent.name) 1060 if _is_skill_disabled(resolved_name): 1061 return json.dumps( 1062 { 1063 "success": False, 1064 "error": ( 1065 f"Skill '{resolved_name}' is disabled. " 1066 "Enable it with `hermes skills` or inspect the files directly on disk." 1067 ), 1068 }, 1069 ensure_ascii=False, 1070 ) 1071 1072 # If a specific file path is requested, read that instead 1073 if file_path and skill_dir: 1074 from tools.path_security import validate_within_dir, has_traversal_component 1075 1076 # Security: Prevent path traversal attacks 1077 if has_traversal_component(file_path): 1078 return json.dumps( 1079 { 1080 "success": False, 1081 "error": "Path traversal ('..') is not allowed.", 1082 "hint": "Use a relative path within the skill directory", 1083 }, 1084 ensure_ascii=False, 1085 ) 1086 1087 target_file = skill_dir / file_path 1088 1089 # Security: Verify resolved path is still within skill directory 1090 traversal_error = validate_within_dir(target_file, skill_dir) 1091 if traversal_error: 1092 return json.dumps( 1093 { 1094 "success": False, 1095 "error": traversal_error, 1096 "hint": "Use a relative path within the skill directory", 1097 }, 1098 ensure_ascii=False, 1099 ) 1100 if not target_file.exists(): 1101 # List available files in the skill directory, organized by type 1102 available_files = { 1103 "references": [], 1104 "templates": [], 1105 "assets": [], 1106 "scripts": [], 1107 "other": [], 1108 } 1109 1110 # Scan for all readable files 1111 for f in skill_dir.rglob("*"): 1112 if f.is_file() and f.name != "SKILL.md": 1113 rel = str(f.relative_to(skill_dir)) 1114 if rel.startswith("references/"): 1115 available_files["references"].append(rel) 1116 elif rel.startswith("templates/"): 1117 available_files["templates"].append(rel) 1118 elif rel.startswith("assets/"): 1119 available_files["assets"].append(rel) 1120 elif rel.startswith("scripts/"): 1121 available_files["scripts"].append(rel) 1122 elif f.suffix in [ 1123 ".md", 1124 ".py", 1125 ".yaml", 1126 ".yml", 1127 ".json", 1128 ".tex", 1129 ".sh", 1130 ]: 1131 available_files["other"].append(rel) 1132 1133 # Remove empty categories 1134 available_files = {k: v for k, v in available_files.items() if v} 1135 1136 return json.dumps( 1137 { 1138 "success": False, 1139 "error": f"File '{file_path}' not found in skill '{name}'.", 1140 "available_files": available_files, 1141 "hint": "Use one of the available file paths listed above", 1142 }, 1143 ensure_ascii=False, 1144 ) 1145 1146 # Read the file content 1147 try: 1148 content = target_file.read_text(encoding="utf-8") 1149 except UnicodeDecodeError: 1150 # Binary file - return info about it instead 1151 return json.dumps( 1152 { 1153 "success": True, 1154 "name": name, 1155 "file": file_path, 1156 "content": f"[Binary file: {target_file.name}, size: {target_file.stat().st_size} bytes]", 1157 "is_binary": True, 1158 }, 1159 ensure_ascii=False, 1160 ) 1161 1162 return json.dumps( 1163 { 1164 "success": True, 1165 "name": name, 1166 "file": file_path, 1167 "content": content, 1168 "file_type": target_file.suffix, 1169 }, 1170 ensure_ascii=False, 1171 ) 1172 1173 # Reuse the parse from the platform check above 1174 frontmatter = parsed_frontmatter 1175 1176 # Get reference, template, asset, and script files if this is a directory-based skill 1177 reference_files = [] 1178 template_files = [] 1179 asset_files = [] 1180 script_files = [] 1181 1182 if skill_dir: 1183 references_dir = skill_dir / "references" 1184 if references_dir.exists(): 1185 reference_files = [ 1186 str(f.relative_to(skill_dir)) for f in references_dir.glob("*.md") 1187 ] 1188 1189 templates_dir = skill_dir / "templates" 1190 if templates_dir.exists(): 1191 for ext in [ 1192 "*.md", 1193 "*.py", 1194 "*.yaml", 1195 "*.yml", 1196 "*.json", 1197 "*.tex", 1198 "*.sh", 1199 ]: 1200 template_files.extend( 1201 [ 1202 str(f.relative_to(skill_dir)) 1203 for f in templates_dir.rglob(ext) 1204 ] 1205 ) 1206 1207 # assets/ — agentskills.io standard directory for supplementary files 1208 assets_dir = skill_dir / "assets" 1209 if assets_dir.exists(): 1210 for f in assets_dir.rglob("*"): 1211 if f.is_file(): 1212 asset_files.append(str(f.relative_to(skill_dir))) 1213 1214 scripts_dir = skill_dir / "scripts" 1215 if scripts_dir.exists(): 1216 for ext in ["*.py", "*.sh", "*.bash", "*.js", "*.ts", "*.rb"]: 1217 script_files.extend( 1218 [str(f.relative_to(skill_dir)) for f in scripts_dir.glob(ext)] 1219 ) 1220 1221 # Read tags/related_skills with backward compat: 1222 # Check metadata.hermes.* first (agentskills.io convention), fall back to top-level 1223 hermes_meta = {} 1224 metadata = frontmatter.get("metadata") 1225 if isinstance(metadata, dict): 1226 hermes_meta = metadata.get("hermes", {}) or {} 1227 1228 tags = _parse_tags(hermes_meta.get("tags") or frontmatter.get("tags", "")) 1229 related_skills = _parse_tags( 1230 hermes_meta.get("related_skills") or frontmatter.get("related_skills", "") 1231 ) 1232 1233 # Build linked files structure for clear discovery 1234 linked_files = {} 1235 if reference_files: 1236 linked_files["references"] = reference_files 1237 if template_files: 1238 linked_files["templates"] = template_files 1239 if asset_files: 1240 linked_files["assets"] = asset_files 1241 if script_files: 1242 linked_files["scripts"] = script_files 1243 1244 try: 1245 rel_path = str(skill_md.relative_to(SKILLS_DIR)) 1246 except ValueError: 1247 # External skill — use path relative to the skill's own parent dir 1248 rel_path = str(skill_md.relative_to(skill_md.parent.parent)) if skill_md.parent.parent else skill_md.name 1249 skill_name = frontmatter.get( 1250 "name", skill_md.stem if not skill_dir else skill_dir.name 1251 ) 1252 legacy_env_vars, _ = _collect_prerequisite_values(frontmatter) 1253 required_env_vars = _get_required_environment_variables( 1254 frontmatter, legacy_env_vars 1255 ) 1256 backend = _get_terminal_backend_name() 1257 env_snapshot = load_env() 1258 missing_required_env_vars = [ 1259 e 1260 for e in required_env_vars 1261 if not e.get("optional") 1262 and not _is_env_var_persisted(e["name"], env_snapshot) 1263 ] 1264 capture_result = _capture_required_environment_variables( 1265 skill_name, 1266 missing_required_env_vars, 1267 ) 1268 if missing_required_env_vars: 1269 env_snapshot = load_env() 1270 remaining_missing_required_envs = _remaining_required_environment_names( 1271 required_env_vars, 1272 capture_result, 1273 env_snapshot=env_snapshot, 1274 ) 1275 setup_needed = bool(remaining_missing_required_envs) 1276 1277 # Register available skill env vars so they pass through to sandboxed 1278 # execution environments (execute_code, terminal). Only vars that are 1279 # actually set get registered — missing ones are reported as setup_needed. 1280 available_env_names = [ 1281 e["name"] 1282 for e in required_env_vars 1283 if e["name"] not in remaining_missing_required_envs 1284 ] 1285 if available_env_names: 1286 try: 1287 from tools.env_passthrough import register_env_passthrough 1288 1289 register_env_passthrough(available_env_names) 1290 except Exception: 1291 logger.debug( 1292 "Could not register env passthrough for skill %s", 1293 skill_name, 1294 exc_info=True, 1295 ) 1296 1297 # Register credential files for mounting into remote sandboxes 1298 # (Modal, Docker). Files that exist on the host are registered; 1299 # missing ones are added to the setup_needed indicators. 1300 required_cred_files_raw = frontmatter.get("required_credential_files", []) 1301 if not isinstance(required_cred_files_raw, list): 1302 required_cred_files_raw = [] 1303 missing_cred_files: list = [] 1304 if required_cred_files_raw: 1305 try: 1306 from tools.credential_files import register_credential_files 1307 1308 missing_cred_files = register_credential_files(required_cred_files_raw) 1309 if missing_cred_files: 1310 setup_needed = True 1311 except Exception: 1312 logger.debug( 1313 "Could not register credential files for skill %s", 1314 skill_name, 1315 exc_info=True, 1316 ) 1317 1318 rendered_content = content 1319 if preprocess: 1320 try: 1321 from agent.skill_preprocessing import preprocess_skill_content 1322 1323 rendered_content = preprocess_skill_content( 1324 content, 1325 skill_dir, 1326 session_id=task_id, 1327 ) 1328 except Exception: 1329 logger.debug( 1330 "Could not preprocess skill content for %s", skill_name, exc_info=True 1331 ) 1332 1333 result = { 1334 "success": True, 1335 "name": skill_name, 1336 "description": frontmatter.get("description", ""), 1337 "tags": tags, 1338 "related_skills": related_skills, 1339 "content": rendered_content, 1340 "path": rel_path, 1341 "skill_dir": str(skill_dir) if skill_dir else None, 1342 "linked_files": linked_files if linked_files else None, 1343 "usage_hint": "To view linked files, call skill_view(name, file_path) where file_path is e.g. 'references/api.md' or 'assets/config.yaml'" 1344 if linked_files 1345 else None, 1346 "required_environment_variables": required_env_vars, 1347 "required_commands": [], 1348 "missing_required_environment_variables": remaining_missing_required_envs, 1349 "missing_credential_files": missing_cred_files, 1350 "missing_required_commands": [], 1351 "setup_needed": setup_needed, 1352 "setup_skipped": capture_result["setup_skipped"], 1353 "readiness_status": SkillReadinessStatus.SETUP_NEEDED.value 1354 if setup_needed 1355 else SkillReadinessStatus.AVAILABLE.value, 1356 } 1357 1358 setup_help = next((e["help"] for e in required_env_vars if e.get("help")), None) 1359 if setup_help: 1360 result["setup_help"] = setup_help 1361 1362 if capture_result["gateway_setup_hint"]: 1363 result["gateway_setup_hint"] = capture_result["gateway_setup_hint"] 1364 1365 if setup_needed: 1366 missing_items = [ 1367 f"env ${env_name}" for env_name in remaining_missing_required_envs 1368 ] + [ 1369 f"file {path}" for path in missing_cred_files 1370 ] 1371 setup_note = _build_setup_note( 1372 SkillReadinessStatus.SETUP_NEEDED, 1373 missing_items, 1374 setup_help, 1375 ) 1376 if backend in _REMOTE_ENV_BACKENDS and setup_note: 1377 setup_note = f"{setup_note} {backend.upper()}-backed skills need these requirements available inside the remote environment as well." 1378 if setup_note: 1379 result["setup_note"] = setup_note 1380 1381 # Surface agentskills.io optional fields when present 1382 if frontmatter.get("compatibility"): 1383 result["compatibility"] = frontmatter["compatibility"] 1384 if isinstance(metadata, dict): 1385 result["metadata"] = metadata 1386 1387 return json.dumps(result, ensure_ascii=False) 1388 1389 except Exception as e: 1390 return tool_error(str(e), success=False) 1391 1392 1393 1394 1395 if __name__ == "__main__": 1396 """Test the skills tool""" 1397 print("🎯 Skills Tool Test") 1398 print("=" * 60) 1399 1400 # Test listing skills 1401 print("\n📋 Listing all skills:") 1402 result = json.loads(skills_list()) 1403 if result["success"]: 1404 print( 1405 f"Found {result['count']} skills in {len(result.get('categories', []))} categories" 1406 ) 1407 print(f"Categories: {result.get('categories', [])}") 1408 print("\nFirst 10 skills:") 1409 for skill in result["skills"][:10]: 1410 cat = f"[{skill['category']}] " if skill.get("category") else "" 1411 print(f" • {cat}{skill['name']}: {skill['description'][:60]}...") 1412 else: 1413 print(f"Error: {result['error']}") 1414 1415 # Test viewing a skill 1416 print("\n📖 Viewing skill 'axolotl':") 1417 result = json.loads(skill_view("axolotl")) 1418 if result["success"]: 1419 print(f"Name: {result['name']}") 1420 print(f"Description: {result.get('description', 'N/A')[:100]}...") 1421 print(f"Content length: {len(result['content'])} chars") 1422 if result.get("linked_files"): 1423 print(f"Linked files: {result['linked_files']}") 1424 else: 1425 print(f"Error: {result['error']}") 1426 1427 # Test viewing a reference file 1428 print("\n📄 Viewing reference file 'axolotl/references/dataset-formats.md':") 1429 result = json.loads(skill_view("axolotl", "references/dataset-formats.md")) 1430 if result["success"]: 1431 print(f"File: {result['file']}") 1432 print(f"Content length: {len(result['content'])} chars") 1433 print(f"Preview: {result['content'][:150]}...") 1434 else: 1435 print(f"Error: {result['error']}") 1436 1437 1438 # --------------------------------------------------------------------------- 1439 # Registry 1440 # --------------------------------------------------------------------------- 1441 1442 SKILLS_LIST_SCHEMA = { 1443 "name": "skills_list", 1444 "description": "List available skills (name + description). Use skill_view(name) to load full content.", 1445 "parameters": { 1446 "type": "object", 1447 "properties": { 1448 "category": { 1449 "type": "string", 1450 "description": "Optional category filter to narrow results", 1451 } 1452 }, 1453 "required": [], 1454 }, 1455 } 1456 1457 SKILL_VIEW_SCHEMA = { 1458 "name": "skill_view", 1459 "description": "Skills allow for loading information about specific tasks and workflows, as well as scripts and templates. Load a skill's full content or access its linked files (references, templates, scripts). First call returns SKILL.md content plus a 'linked_files' dict showing available references/templates/scripts. To access those, call again with file_path parameter.", 1460 "parameters": { 1461 "type": "object", 1462 "properties": { 1463 "name": { 1464 "type": "string", 1465 "description": "The skill name (use skills_list to see available skills). For plugin-provided skills, use the qualified form 'plugin:skill' (e.g. 'superpowers:writing-plans').", 1466 }, 1467 "file_path": { 1468 "type": "string", 1469 "description": "OPTIONAL: Path to a linked file within the skill (e.g., 'references/api.md', 'templates/config.yaml', 'scripts/validate.py'). Omit to get the main SKILL.md content.", 1470 }, 1471 }, 1472 "required": ["name"], 1473 }, 1474 } 1475 1476 registry.register( 1477 name="skills_list", 1478 toolset="skills", 1479 schema=SKILLS_LIST_SCHEMA, 1480 handler=lambda args, **kw: skills_list( 1481 category=args.get("category"), task_id=kw.get("task_id") 1482 ), 1483 check_fn=check_skills_requirements, 1484 emoji="📚", 1485 ) 1486 def _skill_view_with_bump(args, **kw): 1487 """Invoke skill_view, then bump view_count on success. Best-effort: a 1488 telemetry failure never breaks the tool call.""" 1489 name = args.get("name", "") 1490 result = skill_view( 1491 name, file_path=args.get("file_path"), task_id=kw.get("task_id") 1492 ) 1493 try: 1494 parsed = json.loads(result) 1495 if isinstance(parsed, dict) and parsed.get("success"): 1496 # Use the resolved skill name from the payload when present — 1497 # qualified forms ("plugin:skill") return with the canonical name. 1498 resolved = parsed.get("name") or name 1499 if resolved: 1500 from tools.skill_usage import bump_use, bump_view 1501 bump_view(str(resolved)) 1502 # A skill_view tool call is the agent actively loading the skill 1503 # to act on it — that counts as use, not just a browse/view. 1504 # Curator's stale timer keys off last_used_at (see agent/curator.py). 1505 bump_use(str(resolved)) 1506 except Exception: 1507 pass 1508 return result 1509 1510 1511 registry.register( 1512 name="skill_view", 1513 toolset="skills", 1514 schema=SKILL_VIEW_SCHEMA, 1515 handler=_skill_view_with_bump, 1516 check_fn=check_skills_requirements, 1517 emoji="📚", 1518 ) 1519