Skill Hub

from agentscope.app.hub import ClawSkillHub

hub = ClawSkillHub(
    # Addresses this hub in the API and the frontend URL. Keep it stable.
    hub_id="clawhub",
    # Shown in the frontend's hub switcher.
    display_name="ClawHub",
    # Optional token. Anonymous requests work, but are rate-limited.
    api_token=None,
)

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


class InternalSkillHub(SkillHubBase):
    """The skills our platform team publishes internally."""

    def __init__(self) -> None:
        super().__init__(
            hub_id="internal",
            display_name="Internal Skills",
            description="Skills published by the platform team.",
        )

    async def list_skills(
        self,
        user_id: str,
        # The keyword the user typed, or `None` to browse everything.
        q: str | None = None,
        # The opaque cursor from the previous page, or `None` to start.
        cursor: str | None = None,
        limit: int = 20,
    ) -> SkillHubPage:
        """Return one page of listings, without their `SKILL.md` bodies."""
        page = await fetch_our_catalog(query=q, after=cursor, limit=limit)
        return SkillHubPage(
            cards=[self._to_card(c) for c in page.items],
            # `None` tells the frontend there is nothing more to load.
            next_cursor=page.next_cursor,
        )

    async def get_skill(self, user_id: str, card_id: str) -> SkillCard:
        """Return one listing, this time with its `SKILL.md` body."""
        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:
        """Open the skill's archive for streaming."""
        response = await self._client.get(f"/skills/{card_id}/archive")
        # One of "zip", "tar", or "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:
        """Return only the listings this user is allowed to see."""
        # Whatever your own system says this user may install.
        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:
        """Refuse a hidden card, so guessing an id gains nothing."""
        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:
        """Refuse the archive of a hidden card as well."""
        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()