技能中心

from agentscope.app.hub import ClawSkillHub

hub = ClawSkillHub(
    # 用于接口路径和前端 URL，重启前后必须保持不变。
    hub_id="clawhub",
    # 显示在前端的中心切换列表中。
    display_name="ClawHub",
    # 可选令牌。匿名请求同样可用，只是有频率限制。
    api_token=None,
)

from agentscope.app.hub import (
    SkillArchive,
    SkillCard,
    SkillHubBase,
    SkillHubPage,
)


class InternalSkillHub(SkillHubBase):
    """平台团队内部发布的技能。"""

    def __init__(self) -> None:
        super().__init__(
            hub_id="internal",
            display_name="内部技能目录",
            description="由平台团队发布的技能。",
        )

    async def list_skills(
        self,
        user_id: str,
        # 用户输入的搜索词，为 `None` 时浏览全部。
        q: str | None = None,
        # 上一页返回的游标，为 `None` 时从第一页开始。
        cursor: str | None = None,
        limit: int = 20,
    ) -> SkillHubPage:
        """返回一页条目，不含各自的 `SKILL.md` 正文。"""
        page = await fetch_our_catalog(query=q, after=cursor, limit=limit)
        return SkillHubPage(
            cards=[self._to_card(c) for c in page.items],
            # 返回 `None` 表示已无更多内容可加载。
            next_cursor=page.next_cursor,
        )

    async def get_skill(self, user_id: str, card_id: str) -> SkillCard:
        """返回单个条目，这一次带上 `SKILL.md` 正文。"""
        detail = await fetch_one(card_id)
        return self._to_card(detail, markdown=detail.readme)

    async def download(
        self,
        user_id: str,
        card_id: str,
        version: str | None = None,
    ) -> SkillArchive:
        """打开该技能的压缩包以供流式读取。"""
        response = await self._client.get(f"/skills/{card_id}/archive")
        # 取值为 "zip"、"tar" 或 "tar.gz" 之一。
        return SkillArchive(format="tar.gz", stream=response.aiter_bytes())

class InternalSkillHub(SkillHubBase):
    async def list_skills(
        self,
        user_id: str,
        q: str | None = None,
        cursor: str | None = None,
        limit: int = 20,
    ) -> SkillHubPage:
        """只返回该用户有权看到的条目。"""
        # 由开发者自己的系统判定该用户可安装哪些条目。
        allowed = await our_entitlements(user_id)
        page = await fetch_our_catalog(query=q, after=cursor, limit=limit)
        return SkillHubPage(
            cards=[
                self._to_card(c) for c in page.items if c.id in allowed
            ],
            next_cursor=page.next_cursor,
        )

    async def get_skill(self, user_id: str, card_id: str) -> SkillCard:
        """对隐藏条目同样拒绝，使猜测 id 无法奏效。"""
        if card_id not in await our_entitlements(user_id):
            raise KeyError(card_id)
        detail = await fetch_one(card_id)
        return self._to_card(detail, markdown=detail.readme)

    async def download(
        self,
        user_id: str,
        card_id: str,
        version: str | None = None,
    ) -> SkillArchive:
        """隐藏条目的压缩包也一并拒绝。"""
        if card_id not in await our_entitlements(user_id):
            raise KeyError(card_id)
        response = await self._client.get(f"/skills/{card_id}/archive")
        return SkillArchive(format="tar.gz", stream=response.aiter_bytes())

class InternalSkillHub(SkillHubBase):
    async def __aenter__(self) -> "InternalSkillHub":
        self._client = httpx.AsyncClient(timeout=30.0)
        return self

    async def __aexit__(self, *exc: object) -> None:
        await self._client.aclose()