MCP 中心

from agentscope.app.hub import GitHubMCPHub

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

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


class InternalMCPHub(MCPHubBase):
    """公司内部审核通过的 MCP 服务。"""

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

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

    async def get_mcp(self, user_id: str, card_id: str) -> MCPCard:
        """返回单个条目，不存在时抛出 `KeyError`。"""
        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:
        """只返回该用户有权看到的条目。"""
        # 由开发者自己的系统判定该用户可安装哪些条目。
        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:
        """对隐藏条目同样拒绝，使猜测 id 无法奏效。"""
        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="实时天气与预报。",
    config_template=HttpMCPConfig(
        url="https://weather.example.com/mcp",
        # 占位符在任意字符串中都有效：URL、请求头、环境变量、命令行参数。
        headers={"Authorization": "Bearer ${api_key}"},
    ),
    inputs_schema={
        "type": "object",
        "required": ["api_key"],
        "properties": {
            "api_key": {
                "type": "string",
                "title": "API 密钥",
                "description": "在账号的「设置 - API」中获取。",
                # 渲染为密码输入框，且不会回传给浏览器。
                "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()