import os
import yaml
import json
import logging
import collections
from typing import List, Dict, Any, Optional
from app.config import settings
logger = logging.getLogger(__name__)
class FileSystemSkillLoader:
def __init__(self, base_dirs: List[str]):
self.base_dirs = base_dirs
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 directories and parses all SKILL.md rules
into the common dictionary schema used by the orchestrator.
Supports both nested hierarchy (feature/skill/SKILL.md) and flat (skill/SKILL.md).
"""
skills = []
for base_dir in self.base_dirs:
if not os.path.exists(base_dir):
continue
# Root entries can be features (nested) OR skills (flat)
for entry in os.listdir(base_dir):
entry_path = os.path.join(base_dir, entry)
if not os.path.isdir(entry_path):
continue
# Check if this is a Skill directly (Flat Structure)
skill_md_path = os.path.join(entry_path, "SKILL.md")
if os.path.exists(skill_md_path):
# It's a skill! Try to load it.
# We assume it belongs to the "chat" feature if not nested,
# but frontmatter in SKILL.md will override this later.
loaded = self._load_skill(entry_path, entry, default_features=[entry, "chat"])
if loaded:
skills.append(loaded)
continue
# Check if it's a Feature folder (Nested Structure)
for skill_id in os.listdir(entry_path):
skill_path = os.path.join(entry_path, skill_id)
if not os.path.isdir(skill_path):
continue
skill_md_path = os.path.join(skill_path, "SKILL.md")
if os.path.exists(skill_md_path):
loaded = self._load_skill(skill_path, skill_id, default_features=[entry])
if loaded:
skills.append(loaded)
return skills
def _load_skill(self, skill_path: str, skill_id: str, default_features: List[str]) -> Optional[Dict[str, Any]]:
"""Parses a skill folder into the standardized dictionary schema."""
skill_md_path = os.path.join(skill_path, "SKILL.md")
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('\\', '/')
class LazyFileContent(collections.UserString):
def __init__(self, fpath):
self._fpath = fpath
self._content = None
@property
def data(self):
if self._content is None:
try:
with open(self._fpath, "r", encoding='utf-8') as ff:
self._content = ff.read()
except Exception:
self._content = ""
return self._content
vfs_files.append({"file_path": file_rel, "content": LazyFileContent(file_abs), "absolute_path": file_abs})
# 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 = ""
skill_features = default_features
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", "")
# M6: Prioritize features defined in SKILL.md frontmatter!
if frontmatter.get("features"):
skill_features = frontmatter.get("features")
# Generate the legacy internal dict schema
return {
"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": skill_features,
"is_system": metadata.get("is_system", False) or "/app/skills" in skill_path,
"owner_id": metadata.get("owner_id", "admin"),
"extra_metadata": metadata.get("extra_metadata", {"emoji": "🛠️"}),
"files": vfs_files
}
except Exception as e:
logger.error(f"Error parsing FS Skill at {skill_path}: {e}")
return None
# Create a global instance initialized with both system and data directory configs
fs_loader = FileSystemSkillLoader(base_dirs=[
os.path.abspath(os.path.join(os.path.dirname(__file__), "../../../skills")),
os.path.join(settings.DATA_DIR, "skills")
])