Spaces:
Sleeping
Sleeping
| """Skills management endpoints.""" | |
| from __future__ import annotations | |
| import os | |
| from fastapi import APIRouter, File, HTTPException, UploadFile | |
| from pydantic import BaseModel | |
| from app.config import get_settings | |
| from app.skills import user_uploads | |
| from app.skills.registry import REGISTRY | |
| router = APIRouter() | |
| def _is_env_configured(name: str) -> bool: | |
| """Return True if the env var is present and not the placeholder value.""" | |
| if not name: | |
| return True | |
| val = os.environ.get(name, "") | |
| if not val: | |
| return False | |
| # Treat any "your-X-key-here" placeholder as missing | |
| if val.startswith("your-") and val.endswith("-here"): | |
| return False | |
| return True | |
| async def list_skills() -> dict: | |
| """List all registered skills, with enabled/disabled state, runtime | |
| env status, an `uploaded` flag distinguishing user-uploaded skills | |
| (which can be deleted) from built-ins (which cannot), and a `kind` | |
| field ("code" / "prompt" / "builtin") so the UI can render | |
| prompt-only skills differently.""" | |
| settings = get_settings() | |
| user_root = settings.user_skills_dir | |
| return { | |
| "skills": [ | |
| { | |
| "spec": s.model_dump(), | |
| "enabled": REGISTRY.is_enabled(s.name), | |
| "requirements_met": { | |
| env: _is_env_configured(env) for env in s.requires | |
| }, | |
| "uploaded": os.path.isdir(os.path.join(user_root, s.name)), | |
| "kind": _classify_kind(s.name, user_root), | |
| } | |
| for s in REGISTRY.list_specs() | |
| ] | |
| } | |
| def _classify_kind(name: str, user_root: str) -> str: | |
| """Return "builtin", "code" (uploaded with a .py), or "prompt" | |
| (uploaded with only SKILL.md).""" | |
| skill_dir = os.path.join(user_root, name) | |
| if not os.path.isdir(skill_dir): | |
| return "builtin" | |
| has_py = os.path.isfile(os.path.join(skill_dir, f"{name}.py")) | |
| return "code" if has_py else "prompt" | |
| class SkillToggleRequest(BaseModel): | |
| enabled: bool | |
| async def toggle_skill(name: str, body: SkillToggleRequest) -> dict: | |
| if not REGISTRY.get_spec(name): | |
| raise HTTPException(404, f"Unknown skill '{name}'") | |
| REGISTRY.set_enabled(name, body.enabled) | |
| return {"name": name, "enabled": body.enabled} | |
| class SkillDebugRequest(BaseModel): | |
| args: dict = {} | |
| async def debug_skill(name: str, body: SkillDebugRequest) -> dict: | |
| """Manually invoke a skill (bypassing the LLM). Useful for testing.""" | |
| from app.skills.registry import REGISTRY as R | |
| if not R.get_spec(name): | |
| raise HTTPException(404, f"Unknown skill '{name}'") | |
| if not R.is_enabled(name): | |
| raise HTTPException(400, f"Skill '{name}' is disabled") | |
| result = await R.dispatch(name, body.args) | |
| return result.to_dict() | |
| async def upload_skill(file: UploadFile = File(...)) -> dict: | |
| """Upload a new skill as a zip file. See backend/app/skills/user_uploads.py | |
| for the expected zip layout (one top-level directory containing | |
| SKILL.md and a handler module).""" | |
| settings = get_settings() | |
| blob = await file.read() | |
| if not blob: | |
| raise HTTPException(400, "Empty upload") | |
| if len(blob) > settings.max_skill_upload_bytes: | |
| raise HTTPException( | |
| 413, | |
| f"Upload exceeds {settings.max_skill_upload_bytes // (1024*1024)} MB limit", | |
| ) | |
| try: | |
| return user_uploads.install_skill_from_zip(blob) | |
| except ValueError as e: | |
| # 409 for name conflicts (caller can rebrand), 400 for everything else | |
| msg = str(e) | |
| if "conflicts with a built-in" in msg or "already" in msg: | |
| raise HTTPException(409, msg) | |
| raise HTTPException(400, msg) | |
| async def delete_skill(name: str) -> dict: | |
| """Delete an uploaded skill. Built-in skills cannot be deleted.""" | |
| try: | |
| user_uploads.uninstall_skill(name) | |
| except ValueError as e: | |
| raise HTTPException(400, str(e)) | |
| return {"deleted": name} | |