127 lines
4.0 KiB
Python
127 lines
4.0 KiB
Python
"""Agent Skills API 路由 - Phase 9.6
|
|
|
|
使用新的 SkillRegistry (file-based) 而不是 DB-based skill 系统。
|
|
"""
|
|
|
|
from typing import Any
|
|
|
|
from fastapi import APIRouter, HTTPException
|
|
|
|
from app.agents.skills.registry import get_skill_registry, SkillRegistry
|
|
|
|
router = APIRouter(prefix="/api/agent/skills", tags=["Agent Skills"])
|
|
|
|
|
|
def _skill_to_dict(skill) -> dict[str, Any]:
|
|
"""将 SkillMetadata 转换为字典"""
|
|
return {
|
|
"name": skill.name,
|
|
"description": skill.description,
|
|
"tags": skill.tags,
|
|
"triggers": skill.triggers,
|
|
"enabled": skill.enabled,
|
|
"content_preview": skill.content[:200] + "..."
|
|
if len(skill.content) > 200
|
|
else skill.content,
|
|
}
|
|
|
|
|
|
@router.get("", response_model=dict[str, Any])
|
|
async def list_agent_skills() -> dict[str, Any]:
|
|
"""列出所有已加载的 Agent Skills"""
|
|
registry = get_skill_registry()
|
|
skills = registry.list_all()
|
|
return {
|
|
"skills": [_skill_to_dict(s) for s in skills],
|
|
"count": len(skills),
|
|
}
|
|
|
|
|
|
@router.get("/search", response_model=dict[str, Any])
|
|
async def search_agent_skills(
|
|
query: str,
|
|
) -> dict[str, Any]:
|
|
"""搜索 Skills"""
|
|
registry = get_skill_registry()
|
|
results = registry.search(query)
|
|
return {
|
|
"skills": [_skill_to_dict(s) for s in results],
|
|
"count": len(results),
|
|
"query": query,
|
|
}
|
|
|
|
|
|
@router.get("/{skill_name}", response_model=dict[str, Any])
|
|
async def get_agent_skill(skill_name: str) -> dict[str, Any]:
|
|
"""获取指定 Skill 详情"""
|
|
registry = get_skill_registry()
|
|
skill = registry.get_skill(skill_name)
|
|
if not skill:
|
|
raise HTTPException(status_code=404, detail=f"Skill '{skill_name}' not found")
|
|
return {
|
|
"name": skill.name,
|
|
"description": skill.description,
|
|
"tags": skill.tags,
|
|
"triggers": skill.triggers,
|
|
"enabled": skill.enabled,
|
|
"content": skill.content,
|
|
}
|
|
|
|
|
|
@router.get("/{skill_name}/context", response_model=dict[str, str])
|
|
async def get_skill_context(skill_name: str) -> dict[str, str]:
|
|
"""获取 Skill 上下文字符串"""
|
|
registry = get_skill_registry()
|
|
context = registry.get_skill_context([skill_name])
|
|
if not context:
|
|
raise HTTPException(
|
|
status_code=404, detail=f"Skill '{skill_name}' not found or not enabled"
|
|
)
|
|
return {"skill_name": skill_name, "context": context}
|
|
|
|
|
|
@router.post("/context/batch", response_model=dict[str, str])
|
|
async def get_batch_skill_context(
|
|
skill_names: list[str],
|
|
) -> dict[str, str]:
|
|
"""批量获取多个 Skill 的上下文"""
|
|
registry = get_skill_registry()
|
|
context = registry.get_skill_context(skill_names)
|
|
return {"skills": skill_names, "context": context}
|
|
|
|
|
|
@router.post("/reload", response_model=dict[str, Any])
|
|
async def reload_skills(
|
|
skills_dir: str | None = None,
|
|
) -> dict[str, Any]:
|
|
"""重新加载所有 Skills"""
|
|
registry = get_skill_registry()
|
|
# 清除旧 skills
|
|
for name in list(registry._skills.keys()):
|
|
registry.unregister(name)
|
|
# 重新加载
|
|
count = registry.load_all(skills_dir)
|
|
return {"loaded": count, "message": f"Loaded {count} skills"}
|
|
|
|
|
|
@router.post("/{skill_name}/enable", response_model=dict[str, str])
|
|
async def enable_skill(skill_name: str) -> dict[str, str]:
|
|
"""启用 Skill"""
|
|
registry = get_skill_registry()
|
|
skill = registry.get_skill(skill_name)
|
|
if not skill:
|
|
raise HTTPException(status_code=404, detail=f"Skill '{skill_name}' not found")
|
|
skill.enabled = True
|
|
return {"status": "enabled", "skill_name": skill_name}
|
|
|
|
|
|
@router.post("/{skill_name}/disable", response_model=dict[str, str])
|
|
async def disable_skill(skill_name: str) -> dict[str, str]:
|
|
"""禁用 Skill"""
|
|
registry = get_skill_registry()
|
|
skill = registry.get_skill(skill_name)
|
|
if not skill:
|
|
raise HTTPException(status_code=404, detail=f"Skill '{skill_name}' not found")
|
|
skill.enabled = False
|
|
return {"status": "disabled", "skill_name": skill_name}
|