MCP Hub

from agentscope.app.hub import GitHubMCPHub

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

from agentscope.app.hub import MCPCard, MCPHubBase, MCPHubPage


class InternalMCPHub(MCPHubBase):
    """The MCP servers approved for use inside our company."""

    def __init__(self) -> None:
        super().__init__(
            hub_id="internal",
            display_name="Internal Registry",
            description="MCP servers approved by the platform team.",
        )

    async def list_mcps(
        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,
    ) -> MCPHubPage:
        """Return one page of listings."""
        page = await fetch_our_catalog(query=q, after=cursor, limit=limit)
        return MCPHubPage(
            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_mcp(self, user_id: str, card_id: str) -> MCPCard:
        """Return one listing, or raise `KeyError` if there is no such card."""
        return self._to_card(await fetch_one(card_id))

class InternalMCPHub(MCPHubBase):
    async def list_mcps(
        self,
        user_id: str,
        q: str | None = None,
        cursor: str | None = None,
        limit: int = 20,
    ) -> MCPHubPage:
        """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 MCPHubPage(
            cards=[
                self._to_card(c) for c in page.items if c.id in allowed
            ],
            next_cursor=page.next_cursor,
        )

    async def get_mcp(self, user_id: str, card_id: str) -> MCPCard:
        """Refuse a hidden card, so guessing an id gains nothing."""
        if card_id not in await our_entitlements(user_id):
            raise KeyError(card_id)
        return self._to_card(await fetch_one(card_id))

MCPCard(
    hub_id=self.hub_id,
    id="weather",
    name="weather",
    display_name="Weather",
    description="Current conditions and forecasts.",
    config_template=HttpMCPConfig(
        url="https://weather.example.com/mcp",
        # Placeholders work in any string: URLs, headers, env vars, args.
        headers={"Authorization": "Bearer ${api_key}"},
    ),
    inputs_schema={
        "type": "object",
        "required": ["api_key"],
        "properties": {
            "api_key": {
                "type": "string",
                "title": "API key",
                "description": "Found under Settings, API in your account.",
                # Renders as a masked field, and is never echoed back.
                "writeOnly": True,
                "format": "password",
            },
        },
    },
)

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

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