""" Skill manager for LangCode Agent Handles discovery, installation, and management of agent skills from skill registries like skills.sh """ import json import os import subprocess import shutil from pathlib import Path from typing import Dict, List, Any, Optional, Tuple import requests from config_manager import config_manager class SkillManager: """Manages agent skills discovery and installation""" def __init__(self, skills_dir: Optional[str] = None): """ Initialize the skill manager Args: skills_dir: Directory to store skills. If None, uses ~/.langcode/skills """ if skills_dir is None: self.skills_dir = Path.home() / ".langcode" / "skills" else: self.skills_dir = Path(skills_dir) # Ensure skills directory exists self.skills_dir.mkdir(parents=True, exist_ok=True) def search_skills(self, query: str, limit: int = 10) -> List[Dict[str, Any]]: """ Search for skills using skills.sh CLI or fallback to manual search Args: query: Search query limit: Maximum number of results Returns: List of skill information dictionaries """ # Try using skills.sh CLI first try: result = subprocess.run( ["skills", "search", query, "--limit", str(limit), "--json"], capture_output=True, text=True, timeout=30 ) if result.returncode == 0: skills_data = json.loads(result.stdout) return skills_data.get("skills", []) except (FileNotFoundError, subprocess.TimeoutExpired, json.JSONDecodeError): # Fall back to manual search if skills.sh is not available pass # Fallback: try to use known skill registries or return empty return self._fallback_skill_search(query, limit) def _fallback_skill_search(self, query: str, limit: int) -> List[Dict[str, Any]]: """ Fallback skill search when skills.sh CLI is not available Args: query: Search query limit: Maximum number of results Returns: List of skill information dictionaries """ # For now, return empty list - in a full implementation, this would # query known skill registries via their APIs return [] def get_skill_details(self, skill_name: str) -> Optional[Dict[str, Any]]: """ Get detailed information about a specific skill Args: skill_name: Name of the skill to get details for Returns: Skill details dictionary or None if not found """ try: result = subprocess.run( ["skills", "show", skill_name, "--json"], capture_output=True, text=True, timeout=30 ) if result.returncode == 0: return json.loads(result.stdout) except (FileNotFoundError, subprocess.TimeoutExpired, json.JSONDecodeError): pass return None def install_skill(self, skill_name: str, version: Optional[str] = None) -> Tuple[bool, str]: """ Install a skill from the skill registry Args: skill_name: Name of the skill to install version: Specific version to install (optional) Returns: Tuple of (success, message) """ try: cmd = ["skills", "add", skill_name] if version: cmd.extend(["--version", version]) result = subprocess.run( cmd, capture_output=True, text=True, timeout=60 # Skills might take time to install ) if result.returncode == 0: # Refresh installed skills list self._refresh_installed_skills() return True, f"Skill '{skill_name}' installed successfully" else: return False, f"Failed to install skill: {result.stderr}" except (FileNotFoundError, subprocess.TimeoutExpired) as e: return False, f"Error installing skill: {str(e)}" def uninstall_skill(self, skill_name: str) -> Tuple[bool, str]: """ Uninstall a skill Args: skill_name: Name of the skill to uninstall Returns: Tuple of (success, message) """ try: result = subprocess.run( ["skills", "rm", skill_name], capture_output=True, text=True, timeout=30 ) if result.returncode == 0: # Refresh installed skills list self._refresh_installed_skills() return True, f"Skill '{skill_name}' uninstalled successfully" else: return False, f"Failed to uninstall skill: {result.stderr}" except (FileNotFoundError, subprocess.TimeoutExpired) as e: return False, f"Error uninstalling skill: {str(e)}" def list_installed_skills(self) -> List[Dict[str, Any]]: """ List all installed skills Returns: List of installed skill information dictionaries """ try: result = subprocess.run( ["skills", "list", "--json"], capture_output=True, text=True, timeout=30 ) if result.returncode == 0: skills_data = json.loads(result.stdout) return skills_data.get("skills", []) except (FileNotFoundError, subprocess.TimeoutExpired, json.JSONDecodeError): pass # Fallback: check local skills directory return self._list_local_skills() def _list_local_skills(self) -> List[Dict[str, Any]]: """ List skills installed in the local skills directory Returns: List of skill information dictionaries """ skills = [] if not self.skills_dir.exists(): return skills for skill_dir in self.skills_dir.iterdir(): if skill_dir.is_dir(): skill_file = skill_dir / "SKILL.md" if skill_file.exists(): # Try to read basic info from SKILL.md try: content = skill_file.read_text(encoding='utf-8') # Extract name from first line if it's a markdown header name = skill_dir.name description = "Skill installed locally" # Try to extract more info from SKILL.md frontmatter if content.startswith('---'): end_frontmatter = content.find('---', 3) if end_frontmatter != -1: frontmatter = content[3:end_frontmatter] # Simple parsing for name and description for line in frontmatter.split('\n'): if line.startswith('name:'): name = line.split(':', 1)[1].strip() elif line.startswith('description:'): description = line.split(':', 1)[1].strip() skills.append({ "name": name, "description": description, "path": str(skill_dir), "installed": True }) except Exception: # If we can't read the skill properly, still list it skills.append({ "name": skill_dir.name, "description": "Skill installed locally", "path": str(skill_dir), "installed": True }) return skills def _refresh_installed_skills(self): """Refresh the cache of installed skills""" # This would typically update some internal cache # For now, we rely on calling list_installed_skills() which checks fresh pass def skill_is_installed(self, skill_name: str) -> bool: """ Check if a skill is installed Args: skill_name: Name of the skill to check Returns: True if skill is installed, False otherwise """ installed_skills = self.list_installed_skills() return any(skill.get("name") == skill_name for skill in installed_skills) def get_skill_path(self, skill_name: str) -> Optional[Path]: """ Get the filesystem path where a skill is installed Args: skill_name: Name of the skill Returns: Path to the skill directory or None if not installed """ installed_skills = self.list_installed_skills() for skill in installed_skills: if skill.get("name") == skill_name: return Path(skill.get("path", "")) return None # Global skill manager instance skill_manager = SkillManager()