Agent/tools/skill.py

"""
Skill 工具 - 按需加载 Skill 文件

Agent 可以调用此工具来加载特定的 skill 文档
"""

import os
from pathlib import Path
from typing import Optional

from agent.tools import tool, ToolResult
from agent.storage.skill_loader import SkillLoader

# 默认 skills 目录
DEFAULT_SKILLS_DIR = os.getenv("SKILLS_DIR", "./skills")


@tool(
    description="加载指定的 skill 文档。Skills 提供领域知识和最佳实践指导。"
)
async def skill(
    skill_name: str,
    skills_dir: Optional[str] = None,
    uid: str = ""
) -> ToolResult:
    """
    加载指定的 skill 文档

    Args:
        skill_name: Skill 名称（如 "browser-use", "error-handling"）
        skills_dir: Skills 目录路径（可选，默认使用环境变量或 ./skills）
        uid: 用户 ID（自动注入）

    Returns:
        ToolResult: 包含 skill 的详细内容
    """
    # 确定 skills 目录
    skills_path = Path(skills_dir or DEFAULT_SKILLS_DIR)

    if not skills_path.exists():
        return ToolResult(
            title="Skills 目录不存在",
            output=f"找不到 skills 目录: {skills_path}",
            error=f"Directory not found: {skills_path}"
        )

    # 查找文件（支持 skill-name.md 或 skill_name.md）
    skill_file = None
    for ext in [".md"]:
        for name_format in [skill_name, skill_name.replace("-", "_"), skill_name.replace("_", "-")]:
            candidate = skills_path / f"{name_format}{ext}"
            if candidate.exists():
                skill_file = candidate
                break
        if skill_file:
            break

    if not skill_file:
        # 列出可用的 skills
        available_skills = [f.stem for f in skills_path.glob("*.md")]
        return ToolResult(
            title=f"Skill '{skill_name}' 未找到",
            output=f"可用的 skills: {', '.join(available_skills)}",
            error=f"Skill not found: {skill_name}"
        )

    # 加载 skill
    try:
        loader = SkillLoader(str(skills_path))
        skill_obj = loader.load_file(skill_file)

        if not skill_obj:
            return ToolResult(
                title="加载失败",
                output=f"无法解析 skill 文件: {skill_file.name}",
                error="Failed to parse skill file"
            )

        # 格式化输出
        output = f"# {skill_obj.name}\n\n"
        output += f"**Category**: {skill_obj.category}\n\n"

        if skill_obj.description:
            output += f"## Description\n\n{skill_obj.description}\n\n"

        if skill_obj.guidelines:
            output += f"## Guidelines\n\n"
            for i, guideline in enumerate(skill_obj.guidelines, 1):
                output += f"{i}. {guideline}\n"
            output += "\n"

        return ToolResult(
            title=f"Skill: {skill_obj.name}",
            output=output,
            long_term_memory=f"已加载 skill: {skill_obj.name} ({skill_obj.category})",
            include_output_only_once=True,  # skill 内容只展示一次
            metadata={
                "skill_name": skill_obj.name,
                "category": skill_obj.category,
                "scope": skill_obj.scope,
                "guidelines_count": len(skill_obj.guidelines)
            }
        )

    except Exception as e:
        return ToolResult(
            title="加载错误",
            output=f"加载 skill 时出错: {str(e)}",
            error=str(e)
        )


@tool(
    description="列出所有可用的 skills"
)
async def list_skills(
    skills_dir: Optional[str] = None,
    uid: str = ""
) -> ToolResult:
    """
    列出所有可用的 skills

    Args:
        skills_dir: Skills 目录路径（可选）
        uid: 用户 ID（自动注入）

    Returns:
        ToolResult: 包含所有 skills 的列表
    """
    skills_path = Path(skills_dir or DEFAULT_SKILLS_DIR)

    if not skills_path.exists():
        return ToolResult(
            title="Skills 目录不存在",
            output=f"找不到 skills 目录: {skills_path}",
            error=f"Directory not found: {skills_path}"
        )

    try:
        loader = SkillLoader(str(skills_path))
        skills = loader.load_all()

        if not skills:
            return ToolResult(
                title="没有可用的 Skills",
                output="skills 目录中没有找到任何 .md 文件",
                metadata={"count": 0}
            )

        # 按 category 分组
        by_category = {}
        for skill_obj in skills:
            category = skill_obj.category or "general"
            if category not in by_category:
                by_category[category] = []
            by_category[category].append(skill_obj)

        # 格式化输出
        output = f"# 可用的 Skills ({len(skills)} 个)\n\n"

        for category in sorted(by_category.keys()):
            output += f"## {category.title()}\n\n"
            for skill_obj in by_category[category]:
                skill_id = skill_obj.skill_id or skill_obj.name.lower().replace(' ', '-')
                output += f"- **{skill_obj.name}** (`{skill_id}`)\n"
                if skill_obj.description:
                    desc = skill_obj.description.split('\n')[0]  # 第一行
                    output += f"  {desc[:100]}{'...' if len(desc) > 100 else ''}\n"
            output += "\n"

        output += "\n使用 `skill` 工具加载具体的 skill：`skill(skill_name=\"browser-use\")`"

        return ToolResult(
            title=f"可用 Skills ({len(skills)} 个)",
            output=output,
            long_term_memory=f"找到 {len(skills)} 个可用 skills，分为 {len(by_category)} 个类别",
            include_output_only_once=True,
            metadata={
                "count": len(skills),
                "categories": list(by_category.keys()),
                "skills": [{"name": s.name, "category": s.category} for s in skills]
            }
        )

    except Exception as e:
        return ToolResult(
            title="列出 Skills 错误",
            output=f"列出 skills 时出错: {str(e)}",
            error=str(e)
        )