Agent/agent/tools/builtin/resource.py

"""
资源查询工具 - KnowHub 工具表 API 封装
"""

import os
import httpx
from typing import List, Dict, Optional, Any
from agent.tools import tool, ToolResult

KNOWHUB_API = os.getenv("KNOWHUB_API", "http://43.106.118.91:9999").rstrip("/")


@tool(
    description="列出知识库中的所有工具资源",
    groups=["resource"],
)
def resource_list_tools(
    category: Optional[str] = None,
) -> ToolResult:
    """列出所有工具资源，可选按分类过滤

    Args:
        category: 可选的分类过滤，如 "plugin", "model" 等
    """
    try:
        resp = httpx.get(f"{KNOWHUB_API}/api/resource", params={"limit": 1000}, timeout=60.0)
        resp.raise_for_status()
        data = resp.json()

        # 处理返回格式
        if isinstance(data, dict):
            data = data.get("results") or data.get("data") or []

        # 过滤工具
        tools = [r for r in data if isinstance(r, dict) and r.get("id", "").startswith("tools/")]

        # 按分类过滤
        if category:
            tools = [t for t in tools if t.get("metadata", {}).get("category") == category]

        result = {
            "total": len(tools),
            "tools": [{"id": t["id"], "title": t.get("title", ""), "category": t.get("metadata", {}).get("category")} for t in tools]
        }

        return ToolResult(title="工具列表", output=str(result))
    except Exception as e:
        return ToolResult(title="查询失败", output="", error=str(e))


@tool(
    description="获取指定工具的详细信息",
    groups=["resource"],
)
def resource_get_tool(tool_id: str) -> ToolResult:
    """获取工具详情

    Args:
        tool_id: 工具ID，如 "tools/image_gen/comfyui"
    """
    try:
        resp = httpx.get(f"{KNOWHUB_API}/api/resource/{tool_id}", timeout=10.0)
        resp.raise_for_status()
        return ToolResult(title=f"工具详情: {tool_id}", output=str(resp.json()))
    except Exception as e:
        return ToolResult(title="获取工具详情失败", output="", error=str(e))