skills

class Skills(ClientExtension):
    """The client-side Skills extension: register, then access its typed verbs.

    Pass an instance to `Client(extensions=[Skills()])` — this advertises
    `io.modelcontextprotocol/skills` under the client's capabilities — and call
    `client.extension(Skills)` once connected for a `SkillsClient` handle.
    """

    identifier = EXTENSION_ID

    def bind(self, session: ClientSession) -> SkillsClient:
        """Return the SEP-2640 verbs bound to `session`."""
        return SkillsClient(session)

class SkillsClient:
    """The SEP-2640 verbs bound to one connected session.

    Obtain it from `client.extension(Skills)`. The catalog verbs — `list_skills`,
    `get_skill`, and `read_directory` — check that the server advertises the
    extension and validate its response against the SEP-2640 conformance rules
    before returning; `list_skills` and `read_directory` also follow
    `nextCursor` to completion. Read a skill's files with the ordinary
    `Client.read_resource` method.
    """

    def __init__(self, session: ClientSession) -> None:
        self._session = session

    def _require_extension(self, *, directory_read: bool = False) -> None:
        capabilities = self._session.server_capabilities
        settings = (capabilities.extensions or {}).get(EXTENSION_ID) if capabilities else None
        if settings is None:
            raise ValueError(f"server does not advertise the {EXTENSION_ID!r} extension")
        if directory_read and not settings.get("directoryRead"):
            raise ValueError(f"server does not advertise {EXTENSION_ID!r}'s directoryRead setting")

    async def list_skills(self, params: ListSkillsParams | None = None) -> list[Skill]:
        """Call `skills/list`, following `nextCursor` to completion, and validate the result.

        A skill a changing catalog surfaces on more than one page is returned once, in the order
        first seen.

        Raises:
            ValueError: If the server doesn't advertise the Skills extension, its response is not
                SEP-2640 conformant, or it repeats a pagination cursor.
            MCPError: If the server returns an error response.
        """
        self._require_extension()
        base = params if params is not None else ListSkillsParams()
        cursor = base.cursor
        skills: list[Skill] = []
        seen_cursors: set[str] = {cursor} if cursor is not None else set()
        seen_uris: set[str] = set()
        while True:
            # `send_request` parses each page into `ListSkillsResult`, whose validators reject a
            # non-conformant skill or a duplicate URI — no separate conformance call is needed.
            page = await self._session.send_request(
                ListSkillsRequest(params=base.model_copy(update={"cursor": cursor})), ListSkillsResult
            )
            # A server stuck repeating its cursor is a loop; catch that before the content checks.
            if page.next_cursor is not None and page.next_cursor in seen_cursors:
                raise ValueError(f"server repeated skills/list pagination cursor {page.next_cursor!r}")
            # A catalog that changes between page fetches can legitimately repeat a skill across
            # pages; keep the first occurrence rather than treating it as an error.
            for skill in page.skills:
                if skill.uri not in seen_uris:
                    seen_uris.add(skill.uri)
                    skills.append(skill)
            if page.next_cursor is None:
                return skills
            seen_cursors.add(page.next_cursor)
            cursor = page.next_cursor

    async def get_skill(self, uri: str) -> Skill:
        """Call `skills/get` for `uri` and validate the result.

        Unlike `list_skills`, this succeeds for a skill absent from any listing —
        per SEP-2640, a server MUST answer `skills/get` for every skill it serves.

        Raises:
            ValueError: If the server doesn't advertise the Skills extension, its
                response names a different skill, or the skill is not conformant.
            MCPError: If the server returns an error response, such as `-32602`
                for a URI it does not serve.
        """
        self._require_extension()
        # Parsing `GetSkillResult` validates the skill's own conformance; the requested-URI match
        # is the one rule the skill body can't carry, so it stays an explicit check here.
        result = await self._session.send_request(GetSkillRequest(params=GetSkillParams(uri=uri)), GetSkillResult)
        if result.skill.uri != uri:
            raise ValueError(f"server returned skill {result.skill.uri!r} for requested {uri!r}")
        return result.skill

    async def read_directory(self, uri: str, params: ReadDirectoryParams | None = None) -> list[Resource]:
        """Call `resources/directory/read` for `uri`, following `nextCursor` to completion.

        A child a changing directory surfaces on more than one page is returned once, in the order
        first seen.

        Raises:
            ValueError: If the server doesn't advertise the `directoryRead` setting, its response
                is not a valid child listing of `uri`, or it repeats a pagination cursor.
            MCPError: If the server returns an error response.
        """
        self._require_extension(directory_read=True)
        base = params if params is not None else ReadDirectoryParams(uri=uri)
        cursor = base.cursor
        resources: list[Resource] = []
        seen_cursors: set[str] = {cursor} if cursor is not None else set()
        seen_uris: set[str] = set()
        while True:
            page = await self._session.send_request(
                ReadDirectoryRequest(params=base.model_copy(update={"uri": uri, "cursor": cursor})),
                ReadDirectoryResult,
            )
            # A server stuck repeating its cursor is a loop; catch that before the content checks.
            if page.next_cursor is not None and page.next_cursor in seen_cursors:
                raise ValueError(f"server repeated resources/directory/read pagination cursor {page.next_cursor!r}")
            validate_directory_result(uri, page)
            # A changing directory can legitimately repeat a child across pages; keep the first.
            for resource in page.resources:
                if resource.uri not in seen_uris:
                    seen_uris.add(resource.uri)
                    resources.append(resource)
            if page.next_cursor is None:
                return resources
            seen_cursors.add(page.next_cursor)
            cursor = page.next_cursor