Add experimental Server Cards support (SEP-2127)

src/mcp/client/experimental/__init__.py

@@ -0,0 +1,4 @@
+"""Experimental client-side MCP features.
+
+WARNING: These APIs are experimental and may change without notice.
+"""

src/mcp/client/experimental/ai_catalog.py

@@ -0,0 +1,67 @@
+"""Ingest AI Catalogs.
+
+WARNING: These APIs are experimental and may change without notice.
+
+A client discovers the AI artifacts a host advertises by fetching its catalog
+from the well-known location::
+
+    from mcp.client.experimental.ai_catalog import fetch_ai_catalog, well_known_ai_catalog_url
+
+    catalog = await fetch_ai_catalog(well_known_ai_catalog_url("https://dice.example.com"))
+    for entry in catalog.entries:
+        print(entry.identifier, entry.media_type, entry.url)
+
+For the MCP-specific flow — fetch the catalog and the Server Cards it
+advertises in one call — see
+``mcp.client.experimental.server_card.discover_server_cards``.
+"""
+
+from __future__ import annotations
+
+from urllib.parse import urljoin, urlsplit
+
+import httpx
+
+from mcp.shared._httpx_utils import create_mcp_http_client
+from mcp.shared.experimental.ai_catalog.types import (
+    AI_CATALOG_MEDIA_TYPE,
+    AI_CATALOG_WELL_KNOWN_PATH,
+    AICatalog,
+)
+
+__all__ = ["well_known_ai_catalog_url", "fetch_ai_catalog"]
+
+
+def well_known_ai_catalog_url(url: str, *, well_known_path: str = AI_CATALOG_WELL_KNOWN_PATH) -> str:
+    """Resolve the well-known AI Catalog URL for a server's origin.
+
+    Accepts either a bare origin (``https://example.com``) or any URL on the
+    server (e.g. its ``/mcp`` endpoint); the catalog lives at the host root.
+
+    Raises:
+        ValueError: If ``url`` is not an absolute http(s) URL.
+    """
+    parts = urlsplit(url)
+    if parts.scheme not in ("http", "https") or not parts.netloc:
+        raise ValueError(f"Expected an absolute http(s) URL, got {url!r}")
+    return urljoin(f"{parts.scheme}://{parts.netloc}", well_known_path)
+
+
+async def fetch_ai_catalog(url: str, *, http_client: httpx.AsyncClient | None = None) -> AICatalog:
+    """Fetch and validate the AI Catalog at ``url``.
+
+    ``url`` is fetched as-is — catalogs are location-independent; use
+    :func:`well_known_ai_catalog_url` to resolve a host's conventional
+    location. Pass an existing ``http_client`` to reuse connection pooling /
+    auth, otherwise a short-lived client with MCP defaults is used.
+
+    Raises:
+        httpx.HTTPError: If the request fails or returns a non-2xx status.
+        pydantic.ValidationError: If the document is not a valid AI Catalog.
+    """
+    if http_client is None:
+        async with create_mcp_http_client() as client:
+            return await fetch_ai_catalog(url, http_client=client)
+    response = await http_client.get(url, headers={"Accept": f"{AI_CATALOG_MEDIA_TYPE}, application/json"})
+    response.raise_for_status()
+    return AICatalog.model_validate(response.json())

src/mcp/client/experimental/server_card.py

@@ -0,0 +1,118 @@
+"""Ingest MCP Server Cards (SEP-2127).
+
+WARNING: These APIs are experimental and may change without notice.
+
+A client discovers how to connect to the servers a host advertises by
+fetching its AI Catalog and the Server Cards the catalog references::
+
+    from mcp.client.experimental.server_card import discover_server_cards
+
+    for card in await discover_server_cards("https://dice.example.com"):
+        for remote in card.remotes or []:
+            print(remote.type, remote.url, remote.supported_protocol_versions)
+
+Returned :class:`ServerCard` objects are validated; malformed documents raise
+``pydantic.ValidationError``. A missing ``$schema`` key is tolerated — see
+``ServerCard.schema_uri``.
+"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from urllib.parse import urljoin, urlsplit
+
+import httpx
+
+from mcp.client.experimental.ai_catalog import fetch_ai_catalog, well_known_ai_catalog_url
+from mcp.shared._httpx_utils import create_mcp_http_client
+from mcp.shared.experimental.ai_catalog.types import (
+    MCP_CATALOG_WELL_KNOWN_PATH,
+    MCP_SERVER_CARD_MEDIA_TYPE,
+)
+from mcp.shared.experimental.server_card.types import ServerCard
+
+__all__ = ["fetch_server_card", "load_server_card", "discover_server_cards"]
+
+
+async def fetch_server_card(url: str, *, http_client: httpx.AsyncClient | None = None) -> ServerCard:
+    """Fetch and validate the Server Card at ``url``.
+
+    ``url`` is the card's location, typically taken from an AI Catalog
+    entry's ``url``. Pass an existing ``http_client`` to reuse connection
+    pooling / auth, otherwise a short-lived client with MCP defaults is used.
+
+    Raises:
+        httpx.HTTPError: If the request fails or returns a non-2xx status.
+        pydantic.ValidationError: If the document is not a valid Server Card.
+    """
+    if http_client is None:
+        async with create_mcp_http_client() as client:
+            return await fetch_server_card(url, http_client=client)
+    response = await http_client.get(url, headers={"Accept": f"{MCP_SERVER_CARD_MEDIA_TYPE}, application/json"})
+    response.raise_for_status()
+    return ServerCard.model_validate(response.json())
+
+
+async def discover_server_cards(url: str, *, http_client: httpx.AsyncClient | None = None) -> list[ServerCard]:
+    """Discover the MCP servers advertised by the host of ``url``.
+
+    Fetches the host's AI Catalog from ``/.well-known/ai-catalog.json``
+    (falling back to the MCP-scoped ``/.well-known/mcp/catalog.json`` on a
+    404), then validates the Server Card of every MCP server entry — fetched
+    from the entry's ``url`` or read from its inline ``data``. Entries with
+    other media types are ignored.
+
+    Card URLs are taken from the fetched catalog and may point anywhere,
+    including other domains. Non-http(s) card URLs are rejected; beyond that,
+    applications discovering hosts they don't trust should pass an
+    ``http_client`` that enforces their network policy (e.g. rejecting
+    private address ranges or capping redirects) — the SDK imposes none
+    because loopback and intranet servers are legitimate discovery targets.
+
+    Raises:
+        ValueError: If ``url`` is not an absolute http(s) URL, or the catalog
+            references a card at a non-http(s) URL.
+        httpx.HTTPError: If a request fails or returns a non-2xx status.
+        pydantic.ValidationError: If the catalog or a referenced card is invalid.
+    """
+    if http_client is None:
+        async with create_mcp_http_client() as client:
+            return await discover_server_cards(url, http_client=client)
+
+    catalog_url = well_known_ai_catalog_url(url)
+    try:
+        catalog = await fetch_ai_catalog(catalog_url, http_client=http_client)
+    except httpx.HTTPStatusError as exc:
+        if exc.response.status_code != 404:
+            raise
+        catalog_url = well_known_ai_catalog_url(url, well_known_path=MCP_CATALOG_WELL_KNOWN_PATH)
+        catalog = await fetch_ai_catalog(catalog_url, http_client=http_client)
+
+    cards: list[ServerCard] = []
+    for entry in catalog.entries:
+        if entry.media_type != MCP_SERVER_CARD_MEDIA_TYPE:
+            continue
+        if entry.url is not None:
+            # Entry URLs are usually absolute; resolve relative ones against
+            # the catalog's location. The catalog is remote input — never
+            # follow it to a non-http(s) scheme.
+            card_url = urljoin(catalog_url, entry.url)
+            if urlsplit(card_url).scheme not in ("http", "https"):
+                raise ValueError(f"catalog entry {entry.identifier!r} has a non-http(s) card URL: {card_url!r}")
+            cards.append(await fetch_server_card(card_url, http_client=http_client))
+        else:
+            cards.append(ServerCard.model_validate(entry.data))
+    return cards
+
+
+def load_server_card(path: str | Path) -> ServerCard:
+    """Load and validate a Server Card from a JSON file.
+
+    Raises:
+        OSError: If the file cannot be read.
+        json.JSONDecodeError: If the file is not valid JSON.
+        pydantic.ValidationError: If the document is not a valid Server Card.
+    """
+    text = Path(path).read_text(encoding="utf-8")
+    return ServerCard.model_validate(json.loads(text))

src/mcp/server/experimental/__init__.py

@@ -0,0 +1,4 @@
+"""Experimental server-side MCP features.
+
+WARNING: These APIs are experimental and may change without notice.
+"""

src/mcp/server/experimental/ai_catalog.py

@@ -0,0 +1,97 @@
+"""Generate and serve AI Catalogs.
+
+WARNING: These APIs are experimental and may change without notice.
+
+A server advertises its MCP server(s) by serving an AI Catalog from the
+well-known path, with one entry per Server Card::
+
+    catalog = AICatalog(entries=[server_card_entry(card, "https://example.com/server-card")])
+    mount_ai_catalog(server.streamable_http_app(), catalog)   # GET /.well-known/ai-catalog.json
+
+To write a catalog to a file instead, use
+``catalog.model_dump_json(by_alias=True, exclude_none=True)``.
+"""
+
+from __future__ import annotations
+
+from starlette.applications import Starlette
+from starlette.requests import Request
+from starlette.responses import Response
+from starlette.routing import Route
+
+from mcp.shared.experimental.ai_catalog.types import (
+    AI_CATALOG_MEDIA_TYPE,
+    AI_CATALOG_URN_PREFIX,
+    AI_CATALOG_WELL_KNOWN_PATH,
+    MCP_SERVER_CARD_MEDIA_TYPE,
+    AICatalog,
+    CatalogEntry,
+)
+from mcp.shared.experimental.server_card.types import ServerCard
+
+__all__ = ["DISCOVERY_HEADERS", "server_card_entry", "ai_catalog_route", "mount_ai_catalog"]
+
+#: Response headers for discovery endpoints (catalogs and the artifacts they
+#: reference): CORS headers so browser clients can read them, plus a caching
+#: hint.
+DISCOVERY_HEADERS = {
+    "Access-Control-Allow-Origin": "*",
+    "Access-Control-Allow-Methods": "GET",
+    "Access-Control-Allow-Headers": "Content-Type",
+    "Cache-Control": "public, max-age=3600",
+}
+
+
+def _air_identifier(card_name: str) -> str:
+    """Derive an AI Catalog ``urn:air:`` identifier from a Server Card name.
+
+    The card ``name`` is ``namespace/suffix`` in reverse-DNS form
+    (``com.example/weather``); the namespace labels are reversed to forward-DNS
+    (``com.example`` -> ``example.com``) and the suffix appended:
+    ``urn:air:example.com:weather``.
+    """
+    namespace, _, suffix = card_name.partition("/")
+    publisher = ".".join(reversed(namespace.split(".")))
+    return f"{AI_CATALOG_URN_PREFIX}{publisher}:{suffix}"
+
+
+def server_card_entry(card: ServerCard, url: str) -> CatalogEntry:
+    """Build the catalog entry advertising ``card``, served at ``url``.
+
+    The entry's identifier is derived from the card's ``name``
+    (``urn:air:{publisher}:{name}``); display name, description and version are
+    taken from the card. ``url`` should be the absolute URL the card is
+    retrievable from, since catalogs may be fetched cross-domain.
+    """
+    return CatalogEntry(
+        identifier=_air_identifier(card.name),
+        display_name=card.title or card.name,
+        media_type=MCP_SERVER_CARD_MEDIA_TYPE,
+        url=url,
+        description=card.description,
+        version=card.version,
+    )
+
+
+def ai_catalog_route(catalog: AICatalog, *, path: str = AI_CATALOG_WELL_KNOWN_PATH) -> Route:
+    """Build a Starlette GET route that serves ``catalog`` at ``path``.
+
+    Add it to a new app — ``Starlette(routes=[ai_catalog_route(catalog)])`` —
+    or an existing one via :func:`mount_ai_catalog`. The payload is serialized
+    once and served with the CORS and caching headers discovery requires.
+    """
+    body = catalog.model_dump_json(by_alias=True, exclude_none=True).encode()
+
+    async def endpoint(_request: Request) -> Response:
+        return Response(body, media_type=AI_CATALOG_MEDIA_TYPE, headers=DISCOVERY_HEADERS)
+
+    return Route(path, endpoint=endpoint, methods=["GET"], name="ai_catalog")
+
+
+def mount_ai_catalog(app: Starlette, catalog: AICatalog, *, path: str = AI_CATALOG_WELL_KNOWN_PATH) -> None:
+    """Attach an AI Catalog route to an existing Starlette application.
+
+    Discovery expects the catalog to be reachable without authentication;
+    mount it outside any auth middleware.
+    """
+    app.router.routes.append(ai_catalog_route(catalog, path=path))