import json import logging import re import shutil import tempfile import zipfile from pathlib import Path import yaml from fastapi import APIRouter, HTTPException from pydantic import BaseModel, Field from src.config.extensions_config import ExtensionsConfig, SkillStateConfig, get_extensions_config, reload_extensions_config from src.gateway.path_utils import resolve_thread_virtual_path from src.skills import Skill, load_skills from src.skills.loader import get_skills_root_path logger = logging.getLogger(__name__) router = APIRouter(prefix="/api", tags=["skills"]) class SkillResponse(BaseModel): """Response model for skill information.""" name: str = Field(..., description="Name of the skill") description: str = Field(..., description="Description of what the skill does") license: str | None = Field(None, description="License information") category: str = Field(..., description="Category of the skill (public or custom)") enabled: bool = Field(default=True, description="Whether this skill is enabled") class SkillsListResponse(BaseModel): """Response model for listing all skills.""" skills: list[SkillResponse] class SkillUpdateRequest(BaseModel): """Request model for updating a skill.""" enabled: bool = Field(..., description="Whether to enable or disable the skill") class SkillInstallRequest(BaseModel): """Request model for installing a skill from a .skill file.""" thread_id: str = Field(..., description="The thread ID where the .skill file is located") path: str = Field(..., description="Virtual path to the .skill file (e.g., mnt/user-data/outputs/my-skill.skill)") class SkillInstallResponse(BaseModel): """Response model for skill installation.""" success: bool = Field(..., description="Whether the installation was successful") skill_name: str = Field(..., description="Name of the installed skill") message: str = Field(..., description="Installation result message") # Allowed properties in SKILL.md frontmatter ALLOWED_FRONTMATTER_PROPERTIES = {"name", "description", "license", "allowed-tools", "metadata"} def _validate_skill_frontmatter(skill_dir: Path) -> tuple[bool, str, str | None]: """Validate a skill directory's SKILL.md frontmatter. Args: skill_dir: Path to the skill directory containing SKILL.md. Returns: Tuple of (is_valid, message, skill_name). """ skill_md = skill_dir / "SKILL.md" if not skill_md.exists(): return False, "SKILL.md not found", None content = skill_md.read_text() if not content.startswith("---"): return False, "No YAML frontmatter found", None # Extract frontmatter match = re.match(r"^---\n(.*?)\n---", content, re.DOTALL) if not match: return False, "Invalid frontmatter format", None frontmatter_text = match.group(1) # Parse YAML frontmatter try: frontmatter = yaml.safe_load(frontmatter_text) if not isinstance(frontmatter, dict): return False, "Frontmatter must be a YAML dictionary", None except yaml.YAMLError as e: return False, f"Invalid YAML in frontmatter: {e}", None # Check for unexpected properties unexpected_keys = set(frontmatter.keys()) - ALLOWED_FRONTMATTER_PROPERTIES if unexpected_keys: return False, f"Unexpected key(s) in SKILL.md frontmatter: {', '.join(sorted(unexpected_keys))}", None # Check required fields if "name" not in frontmatter: return False, "Missing 'name' in frontmatter", None if "description" not in frontmatter: return False, "Missing 'description' in frontmatter", None # Validate name name = frontmatter.get("name", "") if not isinstance(name, str): return False, f"Name must be a string, got {type(name).__name__}", None name = name.strip() if not name: return False, "Name cannot be empty", None # Check naming convention (hyphen-case: lowercase with hyphens) if not re.match(r"^[a-z0-9-]+$", name): return False, f"Name '{name}' should be hyphen-case (lowercase letters, digits, and hyphens only)", None if name.startswith("-") or name.endswith("-") or "--" in name: return False, f"Name '{name}' cannot start/end with hyphen or contain consecutive hyphens", None if len(name) > 64: return False, f"Name is too long ({len(name)} characters). Maximum is 64 characters.", None # Validate description description = frontmatter.get("description", "") if not isinstance(description, str): return False, f"Description must be a string, got {type(description).__name__}", None description = description.strip() if description: if "<" in description or ">" in description: return False, "Description cannot contain angle brackets (< or >)", None if len(description) > 1024: return False, f"Description is too long ({len(description)} characters). Maximum is 1024 characters.", None return True, "Skill is valid!", name def _skill_to_response(skill: Skill) -> SkillResponse: """Convert a Skill object to a SkillResponse.""" return SkillResponse( name=skill.name, description=skill.description, license=skill.license, category=skill.category, enabled=skill.enabled, ) @router.get( "/skills", response_model=SkillsListResponse, summary="List All Skills", description="Retrieve a list of all available skills from both public and custom directories.", ) async def list_skills() -> SkillsListResponse: """List all available skills. Returns all skills regardless of their enabled status. Returns: A list of all skills with their metadata. Example Response: ```json { "skills": [ { "name": "PDF Processing", "description": "Extract and analyze PDF content", "license": "MIT", "category": "public", "enabled": true }, { "name": "Frontend Design", "description": "Generate frontend designs and components", "license": null, "category": "custom", "enabled": false } ] } ``` """ try: # Load all skills (including disabled ones) skills = load_skills(enabled_only=False) return SkillsListResponse(skills=[_skill_to_response(skill) for skill in skills]) except Exception as e: logger.error(f"Failed to load skills: {e}", exc_info=True) raise HTTPException(status_code=500, detail=f"Failed to load skills: {str(e)}") @router.get( "/skills/{skill_name}", response_model=SkillResponse, summary="Get Skill Details", description="Retrieve detailed information about a specific skill by its name.", ) async def get_skill(skill_name: str) -> SkillResponse: """Get a specific skill by name. Args: skill_name: The name of the skill to retrieve. Returns: Skill information if found. Raises: HTTPException: 404 if skill not found. Example Response: ```json { "name": "PDF Processing", "description": "Extract and analyze PDF content", "license": "MIT", "category": "public", "enabled": true } ``` """ try: skills = load_skills(enabled_only=False) skill = next((s for s in skills if s.name == skill_name), None) if skill is None: raise HTTPException(status_code=404, detail=f"Skill '{skill_name}' not found") return _skill_to_response(skill) except HTTPException: raise except Exception as e: logger.error(f"Failed to get skill {skill_name}: {e}", exc_info=True) raise HTTPException(status_code=500, detail=f"Failed to get skill: {str(e)}") @router.put( "/skills/{skill_name}", response_model=SkillResponse, summary="Update Skill", description="Update a skill's enabled status by modifying the skills_state_config.json file.", ) async def update_skill(skill_name: str, request: SkillUpdateRequest) -> SkillResponse: """Update a skill's enabled status. This will modify the skills_state_config.json file to update the enabled state. The SKILL.md file itself is not modified. Args: skill_name: The name of the skill to update. request: The update request containing the new enabled status. Returns: The updated skill information. Raises: HTTPException: 404 if skill not found, 500 if update fails. Example Request: ```json { "enabled": false } ``` Example Response: ```json { "name": "PDF Processing", "description": "Extract and analyze PDF content", "license": "MIT", "category": "public", "enabled": false } ``` """ try: # Find the skill to verify it exists skills = load_skills(enabled_only=False) skill = next((s for s in skills if s.name == skill_name), None) if skill is None: raise HTTPException(status_code=404, detail=f"Skill '{skill_name}' not found") # Get or create config path config_path = ExtensionsConfig.resolve_config_path() if config_path is None: # Create new config file in parent directory (project root) config_path = Path.cwd().parent / "extensions_config.json" logger.info(f"No existing extensions config found. Creating new config at: {config_path}") # Load current configuration extensions_config = get_extensions_config() # Update the skill's enabled status extensions_config.skills[skill_name] = SkillStateConfig(enabled=request.enabled) # Convert to JSON format (preserve MCP servers config) config_data = { "mcpServers": {name: server.model_dump() for name, server in extensions_config.mcp_servers.items()}, "skills": {name: {"enabled": skill_config.enabled} for name, skill_config in extensions_config.skills.items()}, } # Write the configuration to file with open(config_path, "w") as f: json.dump(config_data, f, indent=2) logger.info(f"Skills configuration updated and saved to: {config_path}") # Reload the extensions config to update the global cache reload_extensions_config() # Reload the skills to get the updated status (for API response) skills = load_skills(enabled_only=False) updated_skill = next((s for s in skills if s.name == skill_name), None) if updated_skill is None: raise HTTPException(status_code=500, detail=f"Failed to reload skill '{skill_name}' after update") logger.info(f"Skill '{skill_name}' enabled status updated to {request.enabled}") return _skill_to_response(updated_skill) except HTTPException: raise except Exception as e: logger.error(f"Failed to update skill {skill_name}: {e}", exc_info=True) raise HTTPException(status_code=500, detail=f"Failed to update skill: {str(e)}") @router.post( "/skills/install", response_model=SkillInstallResponse, summary="Install Skill", description="Install a skill from a .skill file (ZIP archive) located in the thread's user-data directory.", ) async def install_skill(request: SkillInstallRequest) -> SkillInstallResponse: """Install a skill from a .skill file. The .skill file is a ZIP archive containing a skill directory with SKILL.md and optional resources (scripts, references, assets). Args: request: The install request containing thread_id and virtual path to .skill file. Returns: Installation result with skill name and status message. Raises: HTTPException: - 400 if path is invalid or file is not a valid .skill file - 403 if access denied (path traversal detected) - 404 if file not found - 409 if skill already exists - 500 if installation fails Example Request: ```json { "thread_id": "abc123-def456", "path": "/mnt/user-data/outputs/my-skill.skill" } ``` Example Response: ```json { "success": true, "skill_name": "my-skill", "message": "Skill 'my-skill' installed successfully" } ``` """ try: # Resolve the virtual path to actual file path skill_file_path = resolve_thread_virtual_path(request.thread_id, request.path) # Check if file exists if not skill_file_path.exists(): raise HTTPException(status_code=404, detail=f"Skill file not found: {request.path}") # Check if it's a file if not skill_file_path.is_file(): raise HTTPException(status_code=400, detail=f"Path is not a file: {request.path}") # Check file extension if not skill_file_path.suffix == ".skill": raise HTTPException(status_code=400, detail="File must have .skill extension") # Verify it's a valid ZIP file if not zipfile.is_zipfile(skill_file_path): raise HTTPException(status_code=400, detail="File is not a valid ZIP archive") # Get the custom skills directory skills_root = get_skills_root_path() custom_skills_dir = skills_root / "custom" # Create custom directory if it doesn't exist custom_skills_dir.mkdir(parents=True, exist_ok=True) # Extract to a temporary directory first for validation with tempfile.TemporaryDirectory() as temp_dir: temp_path = Path(temp_dir) # Extract the .skill file with zipfile.ZipFile(skill_file_path, "r") as zip_ref: zip_ref.extractall(temp_path) # Find the skill directory (should be the only top-level directory) extracted_items = list(temp_path.iterdir()) if len(extracted_items) == 0: raise HTTPException(status_code=400, detail="Skill archive is empty") # Handle both cases: single directory or files directly in root if len(extracted_items) == 1 and extracted_items[0].is_dir(): skill_dir = extracted_items[0] else: # Files are directly in the archive root skill_dir = temp_path # Validate the skill is_valid, message, skill_name = _validate_skill_frontmatter(skill_dir) if not is_valid: raise HTTPException(status_code=400, detail=f"Invalid skill: {message}") if not skill_name: raise HTTPException(status_code=400, detail="Could not determine skill name") # Check if skill already exists target_dir = custom_skills_dir / skill_name if target_dir.exists(): raise HTTPException(status_code=409, detail=f"Skill '{skill_name}' already exists. Please remove it first or use a different name.") # Move the skill directory to the custom skills directory shutil.copytree(skill_dir, target_dir) logger.info(f"Skill '{skill_name}' installed successfully to {target_dir}") return SkillInstallResponse(success=True, skill_name=skill_name, message=f"Skill '{skill_name}' installed successfully") except HTTPException: raise except Exception as e: logger.error(f"Failed to install skill: {e}", exc_info=True) raise HTTPException(status_code=500, detail=f"Failed to install skill: {str(e)}")