feat: Add skill management capabilities to discover, install, and manage agent skills
- 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
This commit is contained in:
opencode committed
1 parent
753c7d491e
commit
6e7ae3a959
2 files changed
+575
No files matched your search
@@ -0,0 +1,275 @@
|
||||
"""
|
||||
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()
|
||||
Reference in new issue
Block a user