99 lines
3.0 KiB
Python
99 lines
3.0 KiB
Python
"""API routes for AI assistant skills & slash-commands (``/api/ai/skills``)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from typing import Any
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from pydantic import BaseModel, Field
|
|
|
|
from backend.auth.middleware import require_auth
|
|
from backend.skills import (
|
|
create_user_skill,
|
|
delete_user_skill,
|
|
list_skills,
|
|
)
|
|
|
|
logger = logging.getLogger("obsigate.skills_routes")
|
|
router = APIRouter(prefix="/api/ai/skills", tags=["AI"])
|
|
|
|
|
|
class SkillModel(BaseModel):
|
|
"""A single skill or admin command."""
|
|
|
|
model_config = {"extra": "allow"}
|
|
|
|
id: str
|
|
label: str
|
|
icon: str = "🧩"
|
|
type: str = Field(default="skill", description="'skill' or 'admin'")
|
|
description: str = ""
|
|
prompt: str | None = None
|
|
usage: str | None = None
|
|
special: str | None = None
|
|
custom: bool = False
|
|
|
|
|
|
class SkillsResponse(BaseModel):
|
|
"""Response for ``GET /api/ai/skills``."""
|
|
|
|
skills: list[SkillModel]
|
|
commands: list[SkillModel]
|
|
|
|
|
|
class CreateSkillRequest(BaseModel):
|
|
"""Body for ``POST /api/ai/skills``."""
|
|
|
|
id: str = Field(description="Stable identifier (a-z, 0-9, '-', '_')")
|
|
label: str = Field(description="Display name")
|
|
prompt: str = Field(description="Instruction injected into the system prompt")
|
|
description: str = ""
|
|
icon: str = "🧩"
|
|
|
|
|
|
class DeleteSkillResponse(BaseModel):
|
|
"""Response for ``DELETE /api/ai/skills/{skill_id}``."""
|
|
|
|
status: str = "deleted"
|
|
id: str
|
|
|
|
|
|
@router.get("", response_model=SkillsResponse)
|
|
async def api_list_skills(current_user=Depends(require_auth)):
|
|
"""List built-in skills, admin commands and the user's custom skills."""
|
|
data = list_skills(current_user)
|
|
return data
|
|
|
|
|
|
@router.post("", response_model=SkillModel)
|
|
async def api_create_skill(body: CreateSkillRequest, current_user=Depends(require_auth)):
|
|
"""Create a custom skill for the current user."""
|
|
try:
|
|
skill = create_user_skill(current_user, body.model_dump())
|
|
except ValueError as exc:
|
|
raise HTTPException(status_code=400, detail=str(exc)) from exc
|
|
return skill
|
|
|
|
|
|
@router.delete("/{skill_id}", response_model=DeleteSkillResponse)
|
|
async def api_delete_skill(skill_id: str, current_user=Depends(require_auth)):
|
|
"""Delete a custom skill owned by the current user."""
|
|
removed = delete_user_skill(current_user, skill_id)
|
|
if not removed:
|
|
raise HTTPException(status_code=404, detail="Skill introuvable")
|
|
return {"status": "deleted", "id": skill_id}
|
|
|
|
|
|
# Expose the resolved prompt of a skill (used by tests / clients that only
|
|
# need the instruction text without fetching the whole list).
|
|
@router.get("/{skill_id}/prompt")
|
|
async def api_skill_prompt(skill_id: str, current_user=Depends(require_auth)) -> dict[str, Any]:
|
|
"""Return the prompt text for a given skill id."""
|
|
from backend.skills import get_skill_prompt
|
|
|
|
prompt = get_skill_prompt(skill_id, current_user)
|
|
if prompt is None:
|
|
raise HTTPException(status_code=404, detail="Skill introuvable")
|
|
return {"id": skill_id, "prompt": prompt}
|