import os
import yaml
import json
import logging
from typing import List, Dict, Any
from app.config import settings
logger = logging.getLogger(__name__)
class FileSystemSkillLoader:
def __init__(self, base_dir: str):
self.base_dir = base_dir
def _ensure_metadata(self, folder_path: str, skill_name: str, is_system: bool = False):
"""
Auto-generates a .metadata.json file if it's missing.
Assigns 'admin' as default owner if it's auto-generated during boot.
"""
meta_path = os.path.join(folder_path, ".metadata.json")
if not os.path.exists(meta_path):
default_metadata = {
"owner_id": "admin",
"is_system": is_system,
"extra_metadata": {"emoji": "🛠️"}
}
try:
with open(meta_path, "w") as f:
json.dump(default_metadata, f, indent=4)
logger.info(f"Generated default .metadata.json for skill '{skill_name}'")
except Exception as e:
logger.error(f"Failed to generate metadata for {skill_name}: {e}")
return default_metadata
try:
with open(meta_path, "r") as f:
return json.load(f)
except Exception as e:
logger.error(f"Error reading {meta_path}: {e}")
return {}
def get_all_skills(self) -> List[Dict[str, Any]]:
"""
Recursively walks the base direction and parses all SKILL.md rules
into the common dictionary schema used by the orchestrator.
"""
skills = []
if not os.path.exists(self.base_dir):
return skills
# Level 1 directories define Features (e.g., /data/skills/swarm_control)
for feature in os.listdir(self.base_dir):
feature_path = os.path.join(self.base_dir, feature)
if not os.path.isdir(feature_path):
continue
# Level 2 directories define the specific Skills
for skill_id in os.listdir(feature_path):
skill_path = os.path.join(feature_path, skill_id)
if not os.path.isdir(skill_path):
continue
skill_md_path = os.path.join(skill_path, "SKILL.md")
if not os.path.exists(skill_md_path):
continue
try:
with open(skill_md_path, "r", encoding='utf-8') as f:
skill_content = f.read()
# Create virtual file schema
vfs_files = []
for root, _, files in os.walk(skill_path):
for file in files:
# Hide .metadata.json and macOS .DS_Store from the VFS visible to LLMs
if file == ".metadata.json" or file.startswith('.'):
continue
file_abs = os.path.join(root, file)
file_rel = os.path.relpath(file_abs, skill_path).replace('\\', '/')
try:
with open(file_abs, "r", encoding='utf-8') as ff:
vfs_files.append({"file_path": file_rel, "content": ff.read()})
except Exception:
pass # skip binary or unreadable
# Extract or Generate Metadata
metadata = self._ensure_metadata(skill_path, skill_id, is_system=False)
# Extract Description directly from SKILL.md (Fallback or yaml frontmatter)
skill_name = skill_id
skill_desc = ""
if skill_content.startswith("---"):
parts = skill_content.split("---", 2)
if len(parts) >= 3:
frontmatter = yaml.safe_load(parts[1]) or {}
skill_name = frontmatter.get("name", skill_name)
skill_desc = frontmatter.get("description", "")
# Generate the legacy internal dict schema
skill_def = {
"id": f"fs-{skill_id}",
"name": skill_name,
"description": skill_desc or f"Skill loaded dynamically from {skill_id}",
"skill_type": "local",
"is_enabled": True,
"features": [feature], # Sourced directly from folder hierarchy
"is_system": metadata.get("is_system", False),
"owner_id": metadata.get("owner_id", "admin"),
"extra_metadata": metadata.get("extra_metadata", {"emoji": "🛠️"}),
"files": vfs_files
}
skills.append(skill_def)
except Exception as e:
logger.error(f"Error parsing FS Skill at {skill_path}: {e}")
return skills
# Create a global instance initialized with the data directory config
fs_loader = FileSystemSkillLoader(base_dir=os.path.join(settings.DATA_DIR, "skills"))