import json import logging import os 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.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") # Base directory for thread data (relative to backend/) THREAD_DATA_BASE_DIR = ".deer-flow/threads" # Virtual path prefix used in sandbox environments (without leading slash for URL path matching) VIRTUAL_PATH_PREFIX = "mnt/user-data" # Allowed properties in SKILL.md frontmatter ALLOWED_FRONTMATTER_PROPERTIES = {"name", "description", "license", "allowed-tools", "metadata"} def _resolve_skill_file_path(thread_id: str, virtual_path: str) -> Path: """Resolve a virtual skill file path to the actual filesystem path. Args: thread_id: The thread ID. virtual_path: The virtual path (e.g., mnt/user-data/outputs/my-skill.skill). Returns: The resolved filesystem path. Raises: HTTPException: If the path is invalid or outside allowed directories. """ # Remove leading slash if present virtual_path = virtual_path.lstrip("/") # Validate and remove virtual path prefix if not virtual_path.startswith(VIRTUAL_PATH_PREFIX): raise HTTPException(status_code=400, detail=f"Path must start with /{VIRTUAL_PATH_PREFIX}") relative_path = virtual_path[len(VIRTUAL_PATH_PREFIX) :].lstrip("/") # Build the actual path base_dir = Path(os.getcwd()) / THREAD_DATA_BASE_DIR / thread_id / "user-data" actual_path = base_dir / relative_path # Security check: ensure the path is within the thread's user-data directory try: actual_path = actual_path.resolve() base_dir_resolved = base_dir.resolve() if not str(actual_path).startswith(str(base_dir_resolved)): raise HTTPException(status_code=403, detail="Access denied: path traversal detected") except (ValueError, RuntimeError): raise HTTPException(status_code=400, detail="Invalid path") return actual_path 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_skill_file_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)}")