opencode
6e7ae3a959
- Created skill_manager.py for interacting with skills.sh and other skill registries - Added skill management tools (search_skills, install_skill, uninstall_skill, list_installed_skills, get_skill_details) as LangChain Tools - Integrated skill management into the agent's tool system via get_skill_tools() - Added /skills subcommand to CLI for direct terminal access to skill management - Updated system prompt to inform users about skill management capabilities - Created comprehensive documentation in docs/skills-management-design.md - Enables users to discover and install community skills to extend agent capabilities
275 lines
9.6 KiB
Python
275 lines
9.6 KiB
Python
"""
|
|
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() |