168 lines
5.1 KiB
Python
168 lines
5.1 KiB
Python
import re
|
|
|
|
from strix.utils.resource_paths import get_strix_resource_path
|
|
|
|
|
|
_EXCLUDED_CATEGORIES = {"scan_modes", "coordination"}
|
|
_FRONTMATTER_PATTERN = re.compile(r"^---\s*\n.*?\n---\s*\n", re.DOTALL)
|
|
|
|
|
|
def get_available_skills() -> dict[str, list[str]]:
|
|
skills_dir = get_strix_resource_path("skills")
|
|
available_skills: dict[str, list[str]] = {}
|
|
|
|
if not skills_dir.exists():
|
|
return available_skills
|
|
|
|
for category_dir in skills_dir.iterdir():
|
|
if category_dir.is_dir() and not category_dir.name.startswith("__"):
|
|
category_name = category_dir.name
|
|
|
|
if category_name in _EXCLUDED_CATEGORIES:
|
|
continue
|
|
|
|
skills = []
|
|
|
|
for file_path in category_dir.glob("*.md"):
|
|
skill_name = file_path.stem
|
|
skills.append(skill_name)
|
|
|
|
if skills:
|
|
available_skills[category_name] = sorted(skills)
|
|
|
|
return available_skills
|
|
|
|
|
|
def get_all_skill_names() -> set[str]:
|
|
all_skills = set()
|
|
for category_skills in get_available_skills().values():
|
|
all_skills.update(category_skills)
|
|
return all_skills
|
|
|
|
|
|
def validate_skill_names(skill_names: list[str]) -> dict[str, list[str]]:
|
|
available_skills = get_all_skill_names()
|
|
valid_skills = []
|
|
invalid_skills = []
|
|
|
|
for skill_name in skill_names:
|
|
if skill_name in available_skills:
|
|
valid_skills.append(skill_name)
|
|
else:
|
|
invalid_skills.append(skill_name)
|
|
|
|
return {"valid": valid_skills, "invalid": invalid_skills}
|
|
|
|
|
|
def parse_skill_list(skills: str | None) -> list[str]:
|
|
if not skills:
|
|
return []
|
|
return [s.strip() for s in skills.split(",") if s.strip()]
|
|
|
|
|
|
def validate_requested_skills(skill_list: list[str], max_skills: int = 5) -> str | None:
|
|
if len(skill_list) > max_skills:
|
|
return "Cannot specify more than 5 skills for an agent (use comma-separated format)"
|
|
|
|
if not skill_list:
|
|
return None
|
|
|
|
validation = validate_skill_names(skill_list)
|
|
if validation["invalid"]:
|
|
available_skills = list(get_all_skill_names())
|
|
return (
|
|
f"Invalid skills: {validation['invalid']}. "
|
|
f"Available skills: {', '.join(available_skills)}"
|
|
)
|
|
|
|
return None
|
|
|
|
|
|
def generate_skills_description() -> str:
|
|
available_skills = get_available_skills()
|
|
|
|
if not available_skills:
|
|
return "No skills available"
|
|
|
|
all_skill_names = get_all_skill_names()
|
|
|
|
if not all_skill_names:
|
|
return "No skills available"
|
|
|
|
sorted_skills = sorted(all_skill_names)
|
|
skills_str = ", ".join(sorted_skills)
|
|
|
|
description = f"List of skills to load for this agent (max 5). Available skills: {skills_str}. "
|
|
|
|
example_skills = sorted_skills[:2]
|
|
if example_skills:
|
|
example = f"Example: {', '.join(example_skills)} for specialized agent"
|
|
description += example
|
|
|
|
return description
|
|
|
|
|
|
def _get_all_categories() -> dict[str, list[str]]:
|
|
"""Get all skill categories including internal ones (scan_modes, coordination)."""
|
|
skills_dir = get_strix_resource_path("skills")
|
|
all_categories: dict[str, list[str]] = {}
|
|
|
|
if not skills_dir.exists():
|
|
return all_categories
|
|
|
|
for category_dir in skills_dir.iterdir():
|
|
if category_dir.is_dir() and not category_dir.name.startswith("__"):
|
|
category_name = category_dir.name
|
|
skills = []
|
|
|
|
for file_path in category_dir.glob("*.md"):
|
|
skill_name = file_path.stem
|
|
skills.append(skill_name)
|
|
|
|
if skills:
|
|
all_categories[category_name] = sorted(skills)
|
|
|
|
return all_categories
|
|
|
|
|
|
def load_skills(skill_names: list[str]) -> dict[str, str]:
|
|
import logging
|
|
|
|
logger = logging.getLogger(__name__)
|
|
skill_content = {}
|
|
skills_dir = get_strix_resource_path("skills")
|
|
|
|
all_categories = _get_all_categories()
|
|
|
|
for skill_name in skill_names:
|
|
try:
|
|
skill_path = None
|
|
|
|
if "/" in skill_name:
|
|
skill_path = f"{skill_name}.md"
|
|
else:
|
|
for category, skills in all_categories.items():
|
|
if skill_name in skills:
|
|
skill_path = f"{category}/{skill_name}.md"
|
|
break
|
|
|
|
if not skill_path:
|
|
root_candidate = f"{skill_name}.md"
|
|
if (skills_dir / root_candidate).exists():
|
|
skill_path = root_candidate
|
|
|
|
if skill_path and (skills_dir / skill_path).exists():
|
|
full_path = skills_dir / skill_path
|
|
var_name = skill_name.split("/")[-1]
|
|
content = full_path.read_text(encoding="utf-8")
|
|
content = _FRONTMATTER_PATTERN.sub("", content).lstrip()
|
|
skill_content[var_name] = content
|
|
logger.info(f"Loaded skill: {skill_name} -> {var_name}")
|
|
else:
|
|
logger.warning(f"Skill not found: {skill_name}")
|
|
|
|
except (FileNotFoundError, OSError, ValueError) as e:
|
|
logger.warning(f"Failed to load skill {skill_name}: {e}")
|
|
|
|
return skill_content
|