Agent/cyber_agent/tools/builtin/skill.py

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

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

import os
import subprocess
from pathlib import Path
from typing import Optional

from cyber_agent.tools import tool, ToolResult
from cyber_agent.skill.skill_loader import SkillLoader

# 默认 skills 目录（优先级：项目 skills > 框架 skills）
DEFAULT_SKILLS_DIRS = [
    os.getenv("SKILLS_DIR", "./skills"),      # 项目特定 skills（优先）
    "./cyber_agent/skill/skills"                    # 框架内置 skills
]

# 默认单一目录（用于 list_skills）
DEFAULT_SKILLS_DIR = DEFAULT_SKILLS_DIRS[0]


def _check_skill_setup(skill_name: str) -> Optional[str]:
    """
    检查 skill 的环境配置，返回缺失依赖的警告信息

    Args:
        skill_name: Skill 名称

    Returns:
        警告信息（如果有缺失的依赖），否则返回 None
    """
    # 特殊处理：browser-use skill
    if skill_name in ["browser-use", "browser_use"]:
        try:
            # 动态导入 browser-use skill 的 setup 模块
            from cyber_agent.skill.skills.browser_use.setup import (
                _check_browser_use_cli,
                _check_chromium_installed
            )

            cli_installed = _check_browser_use_cli()
            chromium_installed = _check_chromium_installed()

            if not cli_installed or not chromium_installed:
                warning = "\n⚠️ **Setup Required**\n\n"
                warning += "The following dependencies are missing:\n\n"

                if not cli_installed:
                    warning += "- `pip install browser-use`\n"
                if not chromium_installed:
                    warning += "- `uvx browser-use install`\n"

                warning += "\nYou can also use the setup tools:\n"
                warning += "- `check_browser_use()` - Check dependency status\n"
                warning += "- `install_browser_use_chromium()` - Auto-install Chromium\n\n"

                return warning
        except ImportError:
            # Setup 模块不存在，跳过检查
            pass

    return None


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

    Args:
        skill_name: Skill 名称（如 "browser-use", "error-handling"）
        skills_dir: Skills 目录路径（可选，默认按优先级查找）

    Returns:
        ToolResult: 包含 skill 的详细内容

    加载顺序：
    1. 如果指定 skills_dir，只在该目录查找
    2. 否则按优先级查找：./skills/ (项目) -> ./config/skills/ (框架)
    """
    # 确定要搜索的目录列表
    if skills_dir:
        search_paths = [Path(skills_dir)]
    else:
        search_paths = [Path(d) for d in DEFAULT_SKILLS_DIRS]

    # 在目录中查找 skill 文件
    skill_file = None
    found_in_dir = None

    for skills_path in search_paths:
        if not skills_path.exists():
            continue

        # 查找文件（支持 skill-name.md 或 skill_name.md）
        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
                    found_in_dir = skills_path
                    break
            if skill_file:
                break

        if skill_file:
            break

    if not skill_file:
        # 列出所有可用的 skills
        available_skills = []
        for skills_path in search_paths:
            if skills_path.exists():
                available_skills.extend([f.stem for f in skills_path.glob("**/*.md")])

        return ToolResult(
            title=f"Skill '{skill_name}' 未找到",
            output=f"可用的 skills: {', '.join(set(available_skills))}\n\n"
                   f"查找路径: {', '.join([str(p) for p in search_paths])}",
            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"

        # 检查 skill 的环境配置
        setup_warning = _check_skill_setup(skill_name)
        if setup_warning:
            output += setup_warning

        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}) from {found_in_dir}",
            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),
                "loaded_from": str(found_in_dir)
            }
        )

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


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

    Args:
        skills_dir: Skills 目录路径（可选）

    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)
        )