diff --git a/src/mcp/client/experimental/__init__.py b/src/mcp/client/experimental/__init__.py new file mode 100644 index 000000000..a0fbf5764 --- /dev/null +++ b/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. +""" diff --git a/src/mcp/client/experimental/ai_catalog.py b/src/mcp/client/experimental/ai_catalog.py new file mode 100644 index 000000000..60236a21f --- /dev/null +++ b/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()) diff --git a/src/mcp/client/experimental/server_card.py b/src/mcp/client/experimental/server_card.py new file mode 100644 index 000000000..28db303aa --- /dev/null +++ b/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)) diff --git a/src/mcp/server/experimental/__init__.py b/src/mcp/server/experimental/__init__.py new file mode 100644 index 000000000..2a83c96e9 --- /dev/null +++ b/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. +""" diff --git a/src/mcp/server/experimental/ai_catalog.py b/src/mcp/server/experimental/ai_catalog.py new file mode 100644 index 000000000..947dfacea --- /dev/null +++ b/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)) diff --git a/src/mcp/server/experimental/server_card.py b/src/mcp/server/experimental/server_card.py new file mode 100644 index 000000000..338fddf01 --- /dev/null +++ b/src/mcp/server/experimental/server_card.py @@ -0,0 +1,128 @@ +"""Generate and serve MCP Server Cards (SEP-2127). + +WARNING: These APIs are experimental and may change without notice. + +A server author builds a card from the server's identity and serves it. The +recommended location is ``/server-card`` — pass ``path`` to +match the MCP endpoint (e.g. ``/mcp/server-card`` when the server is mounted at +``/mcp``):: + + card = build_server_card(server, name="com.example/dice-roller", remotes=[...]) + mount_server_card(server.streamable_http_app(), card, path="/mcp/server-card") + +Clients learn the card's URL from a catalog entry, so any reachable path works; +the convention only matters for fallback probing. + +A hosted card is only discoverable once it is registered in an AI Catalog (see +``mcp.server.experimental.ai_catalog``); clients learn a card's URL from a +catalog entry rather than guessing it. To write a card to a file instead of +serving it, use ``card.model_dump_json(by_alias=True, exclude_none=True)``. +""" + +from __future__ import annotations + +from typing import Any, Protocol + +from starlette.applications import Starlette +from starlette.requests import Request +from starlette.responses import Response +from starlette.routing import Route + +from mcp.server.experimental.ai_catalog import DISCOVERY_HEADERS +from mcp.shared.experimental.ai_catalog.types import MCP_SERVER_CARD_MEDIA_TYPE +from mcp.shared.experimental.server_card.types import ( + Icon, + Remote, + Repository, + ServerCard, +) + +__all__ = ["build_server_card", "server_card_route", "mount_server_card"] + + +class _ServerIdentity(Protocol): + """The identity attributes shared by the low-level ``Server`` and ``MCPServer``.""" + + name: str + version: str | None + title: str | None + description: str | None + website_url: str | None + icons: list[Icon] | None + + +def build_server_card( + server: _ServerIdentity, + *, + name: str, + remotes: list[Remote] | None = None, + repository: Repository | None = None, + meta: dict[str, Any] | None = None, +) -> ServerCard: + """Build a Server Card from a running server's identity metadata. + + ``name`` is the card's reverse-DNS ``namespace/name`` identifier, passed + explicitly because a server's display ``name`` is free-form. The version, + title, description, website and icons are taken from ``server``. + + Args: + server: A low-level ``Server`` or high-level ``MCPServer`` (anything + exposing the standard identity attributes). + name: Reverse-DNS server name, e.g. ``"io.modelcontextprotocol/everything"``. + remotes: Remote endpoints to advertise. + repository: Optional source repository metadata. + meta: Optional ``_meta`` extension metadata. + + Returns: + A validated :class:`ServerCard`. + + Raises: + ValueError: If ``server`` has no ``version`` or ``description`` set; both + are required on a card. + pydantic.ValidationError: If the resulting card is invalid (e.g. ``name`` + is not reverse-DNS). + """ + if server.version is None: + raise ValueError("server.version must be set to build a Server Card") + if not server.description: + raise ValueError("server.description must be set to build a Server Card") + return ServerCard( + name=name, + version=server.version, + description=server.description, + title=server.title, + website_url=server.website_url, + icons=server.icons, + remotes=remotes, + repository=repository, + _meta=meta, + ) + + +def server_card_route(card: ServerCard, *, path: str = "/server-card") -> Route: + """Build a Starlette GET route that serves ``card`` at ``path``. + + ``path`` defaults to ``/server-card``, the recommended location + (``/server-card``). Add the route to + a new app — ``Starlette(routes=[server_card_route(card)])`` — or an existing + one via :func:`mount_server_card`, and advertise the resulting URL in an AI + Catalog entry. The payload is serialized once and served as + ``application/mcp-server-card+json`` with the CORS and caching headers + discovery requires. + """ + body = card.model_dump_json(by_alias=True, exclude_none=True).encode() + + async def endpoint(_request: Request) -> Response: + return Response(body, media_type=MCP_SERVER_CARD_MEDIA_TYPE, headers=DISCOVERY_HEADERS) + + return Route(path, endpoint=endpoint, methods=["GET"], name="mcp_server_card") + + +def mount_server_card(app: Starlette, card: ServerCard, *, path: str = "/server-card") -> None: + """Attach a Server Card route to an existing Starlette application. + + ``path`` defaults to ``/server-card``, the reserved location. Pre-connection + discovery expects the card to be reachable without authentication; mount it + outside any auth middleware. + """ + app.router.routes.append(server_card_route(card, path=path)) diff --git a/src/mcp/shared/experimental/__init__.py b/src/mcp/shared/experimental/__init__.py new file mode 100644 index 000000000..079b22cfb --- /dev/null +++ b/src/mcp/shared/experimental/__init__.py @@ -0,0 +1,4 @@ +"""Shared experimental MCP features. + +WARNING: These APIs are experimental and may change without notice. +""" diff --git a/src/mcp/shared/experimental/ai_catalog/__init__.py b/src/mcp/shared/experimental/ai_catalog/__init__.py new file mode 100644 index 000000000..1c8324a08 --- /dev/null +++ b/src/mcp/shared/experimental/ai_catalog/__init__.py @@ -0,0 +1,43 @@ +"""AI Catalogs — shared types. + +WARNING: These APIs are experimental and may change without notice. + +An AI Catalog is a JSON index of AI artifacts (MCP Server Cards among them) +published at ``/.well-known/ai-catalog.json`` for domain-level discovery. See +``mcp.shared.experimental.ai_catalog.types`` for the model definitions. + +* Servers generate and serve a catalog with ``mcp.server.experimental.ai_catalog``. +* Clients ingest one with ``mcp.client.experimental.ai_catalog``. +""" + +from mcp.shared.experimental.ai_catalog.types import ( + AI_CATALOG_MEDIA_TYPE, + AI_CATALOG_URN_PREFIX, + AI_CATALOG_WELL_KNOWN_PATH, + MCP_CATALOG_WELL_KNOWN_PATH, + MCP_SERVER_CARD_MEDIA_TYPE, + AICatalog, + Attestation, + CatalogEntry, + HostInfo, + ProvenanceLink, + Publisher, + TrustManifest, + TrustSchema, +) + +__all__ = [ + "AI_CATALOG_MEDIA_TYPE", + "AI_CATALOG_URN_PREFIX", + "AI_CATALOG_WELL_KNOWN_PATH", + "MCP_CATALOG_WELL_KNOWN_PATH", + "MCP_SERVER_CARD_MEDIA_TYPE", + "AICatalog", + "Attestation", + "CatalogEntry", + "HostInfo", + "ProvenanceLink", + "Publisher", + "TrustManifest", + "TrustSchema", +] diff --git a/src/mcp/shared/experimental/ai_catalog/types.py b/src/mcp/shared/experimental/ai_catalog/types.py new file mode 100644 index 000000000..9c63ab227 --- /dev/null +++ b/src/mcp/shared/experimental/ai_catalog/types.py @@ -0,0 +1,257 @@ +"""Pydantic models for AI Catalogs. + +WARNING: These APIs are experimental and may change without notice. + +An AI Catalog is a typed, nestable JSON container for discovering +heterogeneous AI artifacts (MCP servers, A2A agents, skills, nested +catalogs, ...). Each entry declares its artifact type via a media type and +either references the artifact by URL or embeds it inline. Hosts advertise a +catalog at ``/.well-known/ai-catalog.json`` so clients can discover artifacts +— for MCP, the Server Cards in ``mcp.shared.experimental.server_card`` — +without prior configuration. + +The models mirror the normative CDDL schema of the AI Catalog specification, +including the optional Trust Manifest extension. The MCP Catalog defined by +the MCP discovery extension is a structural subset of an AI Catalog, so these +models ingest both document flavours. + +See https://github.com/Agent-Card/ai-catalog and +https://github.com/modelcontextprotocol/experimental-ext-server-card/blob/main/docs/discovery.md. +""" + +from __future__ import annotations + +from datetime import datetime +from typing import Annotated, Any + +from pydantic import Field, model_validator + +from mcp.types._types import MCPModel + +#: Media type identifying an AI Catalog document. +AI_CATALOG_MEDIA_TYPE = "application/ai-catalog+json" +#: Media type identifying an MCP Server Card artifact in a catalog entry, +#: per the MCP discovery extension. +MCP_SERVER_CARD_MEDIA_TYPE = "application/mcp-server-card+json" +#: Well-known path an AI Catalog is published at, relative to the host root. +AI_CATALOG_WELL_KNOWN_PATH = "/.well-known/ai-catalog.json" +#: Well-known path of the MCP-scoped catalog defined by the MCP discovery +#: extension. A structural subset of an AI Catalog, so it parses with these models. +MCP_CATALOG_WELL_KNOWN_PATH = "/.well-known/mcp/catalog.json" +#: URN prefix for AI Catalog entry identifiers. MCP server entries use +#: ``urn:air:{publisher}:{name}`` where ``publisher`` is the forward-DNS form of +#: the card name's namespace (``com.example/weather`` -> ``urn:air:example.com:weather``). +AI_CATALOG_URN_PREFIX = "urn:air:" + + +class TrustSchema(MCPModel): + """The trust framework applied to an artifact.""" + + identifier: str + """Identifier of the trust schema.""" + + version: str + """Version of the trust schema.""" + + governance_uri: str | None = None + """URI of the governance policy document.""" + + verification_methods: list[str] | None = None + """Supported verification methods (e.g. ``"did"``, ``"x509"``, ``"dns-01"``).""" + + +class Attestation(MCPModel): + """A verifiable proof of a claim about an artifact.""" + + type: str + """Attestation type (e.g. ``"publisher-identity"``, ``"SOC2-Type2"``).""" + + uri: str + """Location of the attestation document (HTTPS URL or Data URI).""" + + media_type: str + """Format of the attestation document (e.g. ``"application/jwt"``).""" + + digest: str | None = None + """Cryptographic hash for integrity verification (``algorithm:hex-value``).""" + + size: Annotated[int, Field(ge=0)] | None = None + """Size of the attestation document in bytes.""" + + description: str | None = None + """Human-readable label.""" + + +class ProvenanceLink(MCPModel): + """Lineage information for an artifact.""" + + relation: str + """The relationship (e.g. ``"publishedFrom"``, ``"derivedFrom"``).""" + + source_id: str + """Identifier of the source artifact or data.""" + + source_digest: str | None = None + """Digest of the source.""" + + registry_uri: str | None = None + """URI of the registry holding the source.""" + + statement_uri: str | None = None + """URI of a provenance statement document.""" + + signature_ref: str | None = None + """Reference to the key used to sign the provenance statement.""" + + +class TrustManifest(MCPModel): + """Verifiable identity, attestation and provenance metadata for an artifact. + + An optional companion to catalog entries and hosts; it sits alongside the + artifact without wrapping or modifying its native format. + """ + + identity: str + """Globally unique URI serving as the subject identifier (DID, SPIFFE ID, URL).""" + + identity_type: str | None = None + """Type hint for the identity URI (e.g. ``"did"``, ``"spiffe"``, ``"dns"``).""" + + trust_schema: TrustSchema | None = None + """The trust framework applied to the artifact.""" + + attestations: list[Attestation] | None = None + """Verifiable claims (publisher identity, compliance certifications, ...).""" + + provenance: list[ProvenanceLink] | None = None + """Lineage of the artifact.""" + + privacy_policy_url: str | None = None + """URL to the privacy policy governing the artifact.""" + + terms_of_service_url: str | None = None + """URL to the terms of service.""" + + signature: str | None = None + """Detached JWS signature computed over the Trust Manifest content.""" + + metadata: dict[str, Any] | None = None + """Open map for custom or non-standard trust metadata.""" + + +class Publisher(MCPModel): + """The entity responsible for publishing an artifact.""" + + identifier: str + """Verifiable identifier for the publisher organization.""" + + display_name: str + """Human-readable name of the publisher.""" + + identity_type: str | None = None + """Type hint for the publisher identifier (e.g. ``"did"``, ``"dns"``).""" + + +class HostInfo(MCPModel): + """The operator of a catalog.""" + + display_name: str + """Human-readable name of the host (e.g. the organization name).""" + + identifier: str | None = None + """Verifiable identifier for the host (e.g. a DID or domain name).""" + + documentation_url: str | None = None + """URL to the host's documentation.""" + + logo_url: str | None = None + """URL to the host's logo.""" + + trust_manifest: TrustManifest | None = None + """Trust metadata for the host itself.""" + + +class CatalogEntry(MCPModel): + """A single AI artifact in a catalog. + + Exactly one of ``url`` (artifact by reference) or ``data`` (artifact + inline) must be provided. + """ + + identifier: str + """Identifier for the artifact; SHOULD be a URN or URI. + + MCP server entries use ``urn:air:{publisher}:{name}``, where ``publisher`` is + the forward-DNS form of the referenced Server Card's namespace and ``name`` + is its name suffix. + """ + + display_name: str + """Human-readable name for the artifact.""" + + media_type: str + """Media type identifying the artifact type (e.g. ``"application/mcp-server-card+json"``).""" + + url: str | None = None + """URL where the full artifact document can be retrieved.""" + + data: Any = None + """The complete artifact document inline; its structure is determined by ``media_type``.""" + + version: str | None = None + """Version of the artifact. Semantic versioning is recommended.""" + + description: str | None = None + """Short description of the artifact.""" + + tags: list[str] | None = None + """Keywords for filtering and discovery.""" + + publisher: Publisher | None = None + """The entity that publishes this artifact.""" + + trust_manifest: TrustManifest | None = None + """Trust metadata for this artifact; its ``identity`` must equal ``identifier``.""" + + updated_at: datetime | None = None + """When this entry was last modified.""" + + metadata: dict[str, Any] | None = None + """Open map for custom or non-standard metadata.""" + + @model_validator(mode="after") + def _check_content_and_trust(self) -> CatalogEntry: + if (self.url is None) == (self.data is None): + raise ValueError("a catalog entry must provide exactly one of 'url' or 'data'") + # The spec requires consumers to reject a Trust Manifest whose identity + # does not match the containing entry's identifier. + if self.trust_manifest is not None and self.trust_manifest.identity != self.identifier: + raise ValueError( + f"trust manifest identity {self.trust_manifest.identity!r} " + f"does not match entry identifier {self.identifier!r}" + ) + return self + + +class AICatalog(MCPModel): + """A catalog of AI artifacts, served as ``application/ai-catalog+json``. + + A minimal catalog is just ``entries`` — names, media types and URLs. A + catalog may be served from any URL; hosts that want automated discovery + publish one at ``/.well-known/ai-catalog.json``. + """ + + spec_version: str = "1.0" + """The AI Catalog specification version, in ``"Major.Minor"`` format. + + Required by the specification; defaulted here for documents that omit it. + """ + + entries: list[CatalogEntry] + """The cataloged artifacts. May be empty.""" + + host: HostInfo | None = None + """The operator of this catalog.""" + + metadata: dict[str, Any] | None = None + """Open map for custom or non-standard metadata.""" diff --git a/src/mcp/shared/experimental/server_card/__init__.py b/src/mcp/shared/experimental/server_card/__init__.py new file mode 100644 index 000000000..7041027a7 --- /dev/null +++ b/src/mcp/shared/experimental/server_card/__init__.py @@ -0,0 +1,31 @@ +"""MCP Server Cards (SEP-2127) — shared types. + +WARNING: These APIs are experimental and may change without notice. + +A Server Card is a static metadata document describing a remote MCP server, +suitable for pre-connection discovery. See +``mcp.shared.experimental.server_card.types`` for the model definitions. + +* Servers generate and serve a card with ``mcp.server.experimental.server_card``. +* Clients ingest one with ``mcp.client.experimental.server_card``. +""" + +from mcp.shared.experimental.server_card.types import ( + SERVER_CARD_SCHEMA_URL, + Icon, + Input, + KeyValueInput, + Remote, + Repository, + ServerCard, +) + +__all__ = [ + "SERVER_CARD_SCHEMA_URL", + "Icon", + "Input", + "KeyValueInput", + "Remote", + "Repository", + "ServerCard", +] diff --git a/src/mcp/shared/experimental/server_card/types.py b/src/mcp/shared/experimental/server_card/types.py new file mode 100644 index 000000000..f68c8f67a --- /dev/null +++ b/src/mcp/shared/experimental/server_card/types.py @@ -0,0 +1,165 @@ +"""Pydantic models for MCP Server Cards (SEP-2127). + +WARNING: These APIs are experimental and may change without notice. + +A Server Card is a static metadata document describing a remote MCP server — +its identity, transport endpoints, and supported protocol versions — that a +client can fetch before initialization. Cards are published at any URL and +advertised through an AI Catalog entry (see +``mcp.shared.experimental.ai_catalog``). + +A Server Card describes remote connectivity only: it does not list primitives +(tools/resources/prompts), which remain subject to runtime listing. +""" + +from __future__ import annotations + +import re +from typing import Annotated, Any, Literal + +from pydantic import Field, field_validator + +from mcp.types import Icon +from mcp.types._types import MCPModel + +#: Canonical ``$schema`` value for a Server Card document. +SERVER_CARD_SCHEMA_URL = "https://static.modelcontextprotocol.io/schemas/v1/server-card.schema.json" + +# Pinned to the Server Card schema name: a card referencing the registry +# ``server.schema.json`` is rejected. +_SCHEMA_URL_PATTERN = r"^https://static\.modelcontextprotocol\.io/schemas/v1/server-card\.schema\.json$" +_NAME_PATTERN = r"^[a-zA-Z0-9.-]+/[a-zA-Z0-9._-]+$" +_URL_TEMPLATE_PATTERN = r"^(https?://[^\s]+|\{[a-zA-Z_][a-zA-Z0-9_]*\}[^\s]*)$" + +# Reject version ranges/wildcards. Range operators (incl. ``||`` unions and the +# whitespace of hyphen ranges like ``1.0.0 - 2.0.0``) match anywhere; wildcard +# segments (``1.x``, ``1.*``) only count in the release part, so prereleases +# like ``1.0.0-x`` stay valid. +_VERSION_RANGE_OPERATOR_RE = re.compile(r"[\^~|]|[<>]=?|\s") +_VERSION_WILDCARD_SEGMENT_RE = re.compile(r"(?:^|\.)[xX*](?:\.|$)") + + +class Input(MCPModel): + """A user-supplied or pre-set input value (header value or URL variable).""" + + description: str | None = None + """Human-readable explanation of the input.""" + + is_required: bool | None = None + """Whether the input must be supplied for the connection to succeed.""" + + is_secret: bool | None = None + """Whether the input is a secret value (password, token, ...).""" + + format: Literal["string", "number", "boolean", "filepath"] | None = None + """Input format. ``"filepath"`` is a path on the user's filesystem.""" + + default: str | None = None + """Default value for the input.""" + + placeholder: str | None = None + """Placeholder shown during configuration.""" + + value: str | None = None + """Pre-set value. ``{curly_braces}`` identifiers are replaced from ``variables``.""" + + choices: list[str] | None = None + """Allowed values. If provided, the user must select one.""" + + +class KeyValueInput(Input): + """A named input — used for HTTP headers — whose ``value`` may reference variables.""" + + name: str + """Name of the header.""" + + variables: dict[str, Input] | None = None + """Variables referenced by ``{curly_braces}`` identifiers in ``value``.""" + + +class Repository(MCPModel): + """Repository metadata for the MCP server source code.""" + + url: str + """Repository URL for browsing source and ``git clone``.""" + + source: str + """Hosting service identifier (e.g. ``"github"``).""" + + subfolder: str | None = None + """Relative path from repo root to the server in a monorepo.""" + + id: str | None = None + """Stable repository identifier from the hosting service.""" + + +class Remote(MCPModel): + """Metadata for connecting to a remote (HTTP-based) MCP server endpoint.""" + + type: Literal["streamable-http", "sse"] + """The transport type for this remote endpoint.""" + + url: Annotated[str, Field(pattern=_URL_TEMPLATE_PATTERN)] + """URL template. ``{curly_braces}`` variables are substituted before connecting.""" + + headers: list[KeyValueInput] | None = None + """HTTP headers required or accepted when connecting.""" + + variables: dict[str, Input] | None = None + """Variables referenceable as ``{curly_braces}`` in ``url`` and header values.""" + + supported_protocol_versions: list[str] | None = None + """MCP protocol versions actively supported by this endpoint.""" + + +class ServerCard(MCPModel): + """A static metadata document describing a remote MCP server. + + Published at any URL and advertised through an AI Catalog for + pre-connection discovery. Describes only identity, transport and protocol + versions — never the primitive listings (tools/resources/prompts), which + remain subject to runtime listing. + """ + + schema_uri: Annotated[str, Field(alias="$schema", pattern=_SCHEMA_URL_PATTERN)] = SERVER_CARD_SCHEMA_URL + """The Server Card JSON Schema URI this document conforms to (the ``$schema`` key). + + Required by the schema, but ingestion is intentionally lenient: a document + that omits ``$schema`` is accepted and defaulted to the current ``v1`` URL + rather than rejected. When present it must match the ``v1`` Server Card schema. + """ + + name: Annotated[str, Field(min_length=3, max_length=200, pattern=_NAME_PATTERN)] + """Server name in reverse-DNS ``namespace/name`` format.""" + + version: Annotated[str, Field(max_length=255)] + """Server version. SHOULD follow semantic versioning; ranges are rejected.""" + + description: Annotated[str, Field(min_length=1, max_length=100)] + """Clear human-readable explanation of server functionality.""" + + title: Annotated[str, Field(min_length=1, max_length=100)] | None = None + """Optional human-readable display name.""" + + website_url: str | None = None + """Optional URL to the server's homepage / documentation.""" + + repository: Repository | None = None + """Optional repository metadata for source inspection.""" + + icons: list[Icon] | None = None + """Optional set of sized icons for display in a UI.""" + + remotes: list[Remote] | None = None + """Metadata for making HTTP-based connections to this server.""" + + meta: dict[str, Any] | None = Field(alias="_meta", default=None) + """Extension metadata using reverse-DNS namespacing (the ``_meta`` key).""" + + @field_validator("version") + @classmethod + def _reject_version_ranges(cls, value: str) -> str: + release = value.split("-", 1)[0] + if _VERSION_RANGE_OPERATOR_RE.search(value) or _VERSION_WILDCARD_SEGMENT_RE.search(release): + raise ValueError(f"version must be an exact version, not a range/wildcard: {value!r}") + return value diff --git a/tests/experimental/ai_catalog/__init__.py b/tests/experimental/ai_catalog/__init__.py new file mode 100644 index 000000000..694593126 --- /dev/null +++ b/tests/experimental/ai_catalog/__init__.py @@ -0,0 +1 @@ +"""Tests for AI Catalog support.""" diff --git a/tests/experimental/ai_catalog/test_client.py b/tests/experimental/ai_catalog/test_client.py new file mode 100644 index 000000000..dfc04d3f9 --- /dev/null +++ b/tests/experimental/ai_catalog/test_client.py @@ -0,0 +1,88 @@ +"""Tests for client-side AI Catalog ingestion.""" + +from __future__ import annotations + +import functools + +import httpx +import pytest +from pydantic import ValidationError +from starlette.applications import Starlette +from starlette.responses import JSONResponse +from starlette.routing import Route + +import mcp.client.experimental.ai_catalog as client_module +from mcp.client.experimental.ai_catalog import fetch_ai_catalog, well_known_ai_catalog_url +from mcp.server.experimental.ai_catalog import ai_catalog_route +from mcp.shared.experimental.ai_catalog import MCP_CATALOG_WELL_KNOWN_PATH, AICatalog + +pytestmark = pytest.mark.anyio + +CATALOG = AICatalog(entries=[]) + + +def test_well_known_ai_catalog_url_from_origin() -> None: + assert well_known_ai_catalog_url("https://example.com") == "https://example.com/.well-known/ai-catalog.json" + + +def test_well_known_ai_catalog_url_from_endpoint_url() -> None: + assert well_known_ai_catalog_url("https://example.com:8443/mcp?x=1") == ( + "https://example.com:8443/.well-known/ai-catalog.json" + ) + + +def test_well_known_ai_catalog_url_custom_path() -> None: + assert well_known_ai_catalog_url("https://example.com", well_known_path=MCP_CATALOG_WELL_KNOWN_PATH) == ( + "https://example.com/.well-known/mcp/catalog.json" + ) + + +def test_well_known_ai_catalog_url_rejects_relative() -> None: + with pytest.raises(ValueError, match="absolute"): + well_known_ai_catalog_url("example.com/mcp") + + +def test_well_known_ai_catalog_url_rejects_non_http_scheme() -> None: + with pytest.raises(ValueError, match="http"): + well_known_ai_catalog_url("ftp://example.com") + + +async def test_fetch_with_provided_client() -> None: + app = Starlette(routes=[ai_catalog_route(CATALOG)]) + transport = httpx.ASGITransport(app=app) + async with httpx.AsyncClient(transport=transport) as client: + catalog = await fetch_ai_catalog("https://example.com/.well-known/ai-catalog.json", http_client=client) + assert catalog == CATALOG + + +async def test_fetch_with_default_client(monkeypatch: pytest.MonkeyPatch) -> None: + # Cover the branch that creates its own client, without touching the + # network: bind the module's client factory to an in-memory ASGI transport. + app = Starlette(routes=[ai_catalog_route(CATALOG)]) + transport = httpx.ASGITransport(app=app) + monkeypatch.setattr( + client_module, + "create_mcp_http_client", + functools.partial(httpx.AsyncClient, transport=transport, follow_redirects=True), + ) + catalog = await fetch_ai_catalog("https://example.com/.well-known/ai-catalog.json") + assert catalog == CATALOG + + +async def test_fetch_invalid_catalog_raises_validation_error() -> None: + async def bad(_request: object) -> JSONResponse: + return JSONResponse({"specVersion": "1.0"}) # entries missing + + app = Starlette(routes=[Route("/.well-known/ai-catalog.json", bad, methods=["GET"])]) + transport = httpx.ASGITransport(app=app) + async with httpx.AsyncClient(transport=transport) as client: + with pytest.raises(ValidationError): + await fetch_ai_catalog("https://example.com/.well-known/ai-catalog.json", http_client=client) + + +async def test_fetch_raises_for_http_error() -> None: + app = Starlette(routes=[]) # nothing at the well-known path -> 404 + transport = httpx.ASGITransport(app=app) + async with httpx.AsyncClient(transport=transport) as client: + with pytest.raises(httpx.HTTPStatusError): + await fetch_ai_catalog("https://example.com/.well-known/ai-catalog.json", http_client=client) diff --git a/tests/experimental/ai_catalog/test_server.py b/tests/experimental/ai_catalog/test_server.py new file mode 100644 index 000000000..6602e22b4 --- /dev/null +++ b/tests/experimental/ai_catalog/test_server.py @@ -0,0 +1,74 @@ +"""Tests for server-side AI Catalog generation and serving.""" + +from __future__ import annotations + +import httpx +import pytest +from starlette.applications import Starlette + +from mcp.server.experimental.ai_catalog import ai_catalog_route, mount_ai_catalog, server_card_entry +from mcp.shared.experimental.ai_catalog import AICatalog +from mcp.shared.experimental.server_card import ServerCard + +pytestmark = pytest.mark.anyio + +CARD_URL = "https://dice.example.com/server-card.json" + + +def make_card(title: str | None = None) -> ServerCard: + return ServerCard(name="example/dice", version="1.0.0", description="Rolls dice.", title=title) + + +def test_server_card_entry_derives_identifier_and_metadata_from_card() -> None: + entry = server_card_entry(make_card(title="Dice Roller"), CARD_URL) + assert entry.identifier == "urn:air:example:dice" + assert entry.display_name == "Dice Roller" + assert entry.media_type == "application/mcp-server-card+json" + assert entry.url == CARD_URL + assert entry.description == "Rolls dice." + assert entry.version == "1.0.0" + + +def test_server_card_entry_reverses_namespace_to_publisher_domain() -> None: + """The identifier is anchored on the publisher's forward-DNS domain.""" + card = ServerCard(name="com.example/weather", version="1.0.0", description="Weather.") + assert server_card_entry(card, CARD_URL).identifier == "urn:air:example.com:weather" + + +def test_server_card_entry_falls_back_to_card_name_without_title() -> None: + assert server_card_entry(make_card(), CARD_URL).display_name == "example/dice" + + +async def _get(app: Starlette, path: str) -> httpx.Response: + transport = httpx.ASGITransport(app=app) + async with httpx.AsyncClient(transport=transport, base_url="https://dice.example.com") as client: + return await client.get(path) + + +async def test_ai_catalog_route_serves_catalog_with_discovery_headers() -> None: + catalog = AICatalog(entries=[server_card_entry(make_card(), CARD_URL)]) + app = Starlette(routes=[ai_catalog_route(catalog)]) + response = await _get(app, "/.well-known/ai-catalog.json") + assert response.status_code == 200 + assert response.headers["content-type"] == "application/ai-catalog+json" + # Discovery requires CORS headers (MUST) and caching headers (SHOULD). + assert response.headers["access-control-allow-origin"] == "*" + assert response.headers["access-control-allow-methods"] == "GET" + assert response.headers["access-control-allow-headers"] == "Content-Type" + assert response.headers["cache-control"] == "public, max-age=3600" + assert response.text == catalog.model_dump_json(by_alias=True, exclude_none=True) + + +async def test_mount_ai_catalog_on_existing_app() -> None: + app = Starlette() + mount_ai_catalog(app, AICatalog(entries=[])) + response = await _get(app, "/.well-known/ai-catalog.json") + assert response.status_code == 200 + assert AICatalog.model_validate(response.json()) == AICatalog(entries=[]) + + +async def test_mount_ai_catalog_custom_path() -> None: + app = Starlette() + mount_ai_catalog(app, AICatalog(entries=[]), path="/.well-known/mcp/catalog.json") + response = await _get(app, "/.well-known/mcp/catalog.json") + assert response.status_code == 200 diff --git a/tests/experimental/ai_catalog/test_types.py b/tests/experimental/ai_catalog/test_types.py new file mode 100644 index 000000000..c797dc066 --- /dev/null +++ b/tests/experimental/ai_catalog/test_types.py @@ -0,0 +1,141 @@ +"""Tests for AI Catalog models.""" + +from __future__ import annotations + +from typing import Any + +import pytest +from pydantic import ValidationError + +from mcp.shared.experimental.ai_catalog import ( + AICatalog, + CatalogEntry, +) + +MINIMAL_ENTRY = { + "identifier": "urn:air:example.com:weather", + "displayName": "Weather Service", + "mediaType": "application/mcp-server-card+json", + "url": "https://example.com/server-card.json", +} + +# Trimmed from the AI Catalog specification's multi-artifact example. +FULL_CATALOG = { + "specVersion": "1.0", + "host": { + "displayName": "Acme Enterprise AI", + "identifier": "did:web:acme-corp.com", + "documentationUrl": "https://docs.acme-corp.com/ai", + }, + "entries": [ + { + "identifier": "urn:acme:agent:finance", + "displayName": "Acme Finance Agent", + "version": "2.1.0", + "mediaType": "application/a2a-agent-card+json", + "url": "https://api.acme-corp.com/agents/finance/v2.1.json", + "updatedAt": "2026-03-15T10:00:00Z", + "tags": ["finance", "agent"], + "publisher": { + "identifier": "did:web:acme-corp.com", + "displayName": "Acme Financial Corp", + "identityType": "did", + }, + "trustManifest": { + "identity": "urn:acme:agent:finance", + "identityType": "did", + "trustSchema": { + "identifier": "urn:trust:acme-enterprise-v1", + "version": "1.0", + "governanceUri": "https://acme-corp.com/trust/governance.pdf", + "verificationMethods": ["did", "x509"], + }, + "attestations": [ + { + "type": "SOC2-Type2", + "uri": "https://trust.acme-corp.com/reports/soc2.pdf", + "mediaType": "application/pdf", + "digest": "sha256:" + "a" * 64, + "size": 123456, + "description": "Annual SOC 2 report", + } + ], + "provenance": [ + { + "relation": "publishedFrom", + "sourceId": "https://github.com/acme-corp/finance-agent", + "sourceDigest": "sha256:" + "b" * 64, + "registryUri": "oci://registry.acme-corp.com/agents/finance", + "statementUri": "https://trust.acme-corp.com/provenance/finance-agent.json", + "signatureRef": "did:web:acme-corp.com#key-1", + } + ], + "privacyPolicyUrl": "https://acme-corp.com/legal/privacy", + "termsOfServiceUrl": "https://acme-corp.com/legal/terms", + "signature": "eyJhbGciOiJFUzI1NiJ9..detached-jws-signature", + "metadata": {"com.acme.reviewCycle": "annual"}, + }, + "metadata": {"com.acme.deploymentRegion": "eu-west-1"}, + }, + { + "identifier": "urn:air:acme.com:weather", + "displayName": "Weather Service", + "mediaType": "application/mcp-server-card+json", + "data": {"name": "com.acme/weather", "version": "1.0.0", "description": "Weather lookups."}, + }, + ], + "metadata": {"com.acme.catalogOwner": "platform-team"}, +} + +# The MCP Catalog from the MCP discovery extension is a structural subset of +# an AI Catalog and must parse with the same models. +MCP_CATALOG = { + "specVersion": "draft", + "entries": [ + { + "identifier": "urn:air:example.com:weather", + "displayName": "Weather Service", + "mediaType": "application/mcp-server-card+json", + "url": "https://example.com/.well-known/mcp-server-card", + } + ], +} + + +@pytest.mark.parametrize("doc", [FULL_CATALOG, MCP_CATALOG]) +def test_catalog_round_trips(doc: dict[str, Any]) -> None: + """A catalog document survives validate -> dump unchanged.""" + catalog = AICatalog.model_validate(doc) + assert catalog.model_dump(mode="json", by_alias=True, exclude_none=True) == doc + + +def test_spec_version_defaults_when_omitted() -> None: + """Ingestion is lenient: a catalog without specVersion gets the current default.""" + catalog = AICatalog.model_validate({"entries": []}) + assert catalog.spec_version == "1.0" + + +def test_entry_requires_url_or_data() -> None: + doc = {k: v for k, v in MINIMAL_ENTRY.items() if k != "url"} + with pytest.raises(ValidationError) as excinfo: + CatalogEntry.model_validate(doc) + assert "exactly one of 'url' or 'data'" in str(excinfo.value) + + +def test_entry_rejects_url_and_data_together() -> None: + with pytest.raises(ValidationError) as excinfo: + CatalogEntry.model_validate({**MINIMAL_ENTRY, "data": {"name": "com.example/weather"}}) + assert "exactly one of 'url' or 'data'" in str(excinfo.value) + + +def test_entry_rejects_mismatched_trust_manifest_identity() -> None: + """The spec requires rejecting trust manifests bound to a different identifier.""" + with pytest.raises(ValidationError) as excinfo: + CatalogEntry.model_validate({**MINIMAL_ENTRY, "trustManifest": {"identity": "urn:air:other.example:name"}}) + assert "does not match entry identifier" in str(excinfo.value) + + +def test_entry_accepts_matching_trust_manifest_identity() -> None: + entry = CatalogEntry.model_validate({**MINIMAL_ENTRY, "trustManifest": {"identity": MINIMAL_ENTRY["identifier"]}}) + assert entry.trust_manifest is not None + assert entry.trust_manifest.identity == entry.identifier diff --git a/tests/experimental/server_card/__init__.py b/tests/experimental/server_card/__init__.py new file mode 100644 index 000000000..d1694b9fc --- /dev/null +++ b/tests/experimental/server_card/__init__.py @@ -0,0 +1 @@ +"""Tests for MCP Server Card support.""" diff --git a/tests/experimental/server_card/test_client.py b/tests/experimental/server_card/test_client.py new file mode 100644 index 000000000..68db8b3cd --- /dev/null +++ b/tests/experimental/server_card/test_client.py @@ -0,0 +1,187 @@ +"""Tests for client-side Server Card ingestion and discovery.""" + +from __future__ import annotations + +import functools +import json +from pathlib import Path + +import httpx +import pytest +from pydantic import ValidationError +from starlette.applications import Starlette +from starlette.responses import JSONResponse, Response +from starlette.routing import Route + +import mcp.client.experimental.server_card as client_module +from mcp.client.experimental.server_card import discover_server_cards, fetch_server_card, load_server_card +from mcp.server.experimental.ai_catalog import ai_catalog_route, server_card_entry +from mcp.server.experimental.server_card import server_card_route +from mcp.shared.experimental.ai_catalog import MCP_CATALOG_WELL_KNOWN_PATH, AICatalog, CatalogEntry +from mcp.shared.experimental.server_card import ServerCard + +pytestmark = pytest.mark.anyio + +CARD = ServerCard(name="example/dice", version="1.0.0", description="Rolls dice.") +CARD_PATH = "/server-card.json" +CARD_URL = f"https://example.com{CARD_PATH}" + + +def make_discovery_app(*entries: CatalogEntry, catalog_path: str | None = None) -> Starlette: + """An app serving an AI Catalog with ``entries`` plus the card itself.""" + catalog = AICatalog(entries=list(entries) if entries else [server_card_entry(CARD, CARD_URL)]) + routes = [server_card_route(CARD, path=CARD_PATH)] + if catalog_path is None: + routes.append(ai_catalog_route(catalog)) + else: + routes.append(ai_catalog_route(catalog, path=catalog_path)) + return Starlette(routes=routes) + + +async def test_fetch_server_card_from_url() -> None: + transport = httpx.ASGITransport(app=make_discovery_app()) + async with httpx.AsyncClient(transport=transport) as client: + card = await fetch_server_card(CARD_URL, http_client=client) + assert card == CARD + + +async def test_fetch_server_card_with_default_client(monkeypatch: pytest.MonkeyPatch) -> None: + # Cover the branch that creates its own client, without touching the + # network: bind the module's client factory to an in-memory ASGI transport. + transport = httpx.ASGITransport(app=make_discovery_app()) + monkeypatch.setattr( + client_module, + "create_mcp_http_client", + functools.partial(httpx.AsyncClient, transport=transport, follow_redirects=True), + ) + assert await fetch_server_card(CARD_URL) == CARD + + +async def test_fetch_invalid_card_raises_validation_error() -> None: + async def bad(_request: object) -> JSONResponse: + return JSONResponse({"name": "missing-required-fields"}) + + app = Starlette(routes=[Route(CARD_PATH, bad, methods=["GET"])]) + transport = httpx.ASGITransport(app=app) + async with httpx.AsyncClient(transport=transport) as client: + with pytest.raises(ValidationError): + await fetch_server_card(CARD_URL, http_client=client) + + +async def test_fetch_raises_for_http_error() -> None: + app = Starlette(routes=[]) # nothing at the card URL -> 404 + transport = httpx.ASGITransport(app=app) + async with httpx.AsyncClient(transport=transport) as client: + with pytest.raises(httpx.HTTPStatusError): + await fetch_server_card(CARD_URL, http_client=client) + + +async def test_discover_server_cards_via_well_known_catalog() -> None: + transport = httpx.ASGITransport(app=make_discovery_app()) + async with httpx.AsyncClient(transport=transport) as client: + cards = await discover_server_cards("https://example.com", http_client=client) + assert cards == [CARD] + + +async def test_discover_server_cards_with_default_client(monkeypatch: pytest.MonkeyPatch) -> None: + # Cover the branch that creates its own client, without touching the + # network: bind the module's client factory to an in-memory ASGI transport. + transport = httpx.ASGITransport(app=make_discovery_app()) + monkeypatch.setattr( + client_module, + "create_mcp_http_client", + functools.partial(httpx.AsyncClient, transport=transport, follow_redirects=True), + ) + assert await discover_server_cards("https://example.com") == [CARD] + + +async def test_discover_server_cards_resolves_relative_entry_url() -> None: + entry = server_card_entry(CARD, CARD_PATH) # relative to the catalog location + transport = httpx.ASGITransport(app=make_discovery_app(entry)) + async with httpx.AsyncClient(transport=transport) as client: + cards = await discover_server_cards("https://example.com/mcp", http_client=client) + assert cards == [CARD] + + +async def test_discover_server_cards_reads_inline_data_entries() -> None: + entry = CatalogEntry( + identifier="urn:air:example:dice", + display_name="Dice", + media_type="application/mcp-server-card+json", + data=CARD.model_dump(mode="json", by_alias=True, exclude_none=True), + ) + transport = httpx.ASGITransport(app=make_discovery_app(entry)) + async with httpx.AsyncClient(transport=transport) as client: + cards = await discover_server_cards("https://example.com", http_client=client) + assert cards == [CARD] + + +async def test_discover_server_cards_ignores_non_card_entries() -> None: + """Catalog entries that are not Server Cards are skipped.""" + other = CatalogEntry( + identifier="urn:air:example.com:agent", + display_name="Some Agent", + media_type="application/a2a-agent-card+json", + url="https://example.com/agent.json", + ) + app = make_discovery_app(server_card_entry(CARD, CARD_URL), other) + transport = httpx.ASGITransport(app=app) + async with httpx.AsyncClient(transport=transport) as client: + cards = await discover_server_cards("https://example.com", http_client=client) + assert cards == [CARD] + + +async def test_discover_server_cards_rejects_non_http_card_url() -> None: + """A hostile catalog must not steer the client to non-http(s) schemes.""" + entry = server_card_entry(CARD, CARD_URL).model_copy(update={"url": "file:///etc/passwd"}) + transport = httpx.ASGITransport(app=make_discovery_app(entry)) + async with httpx.AsyncClient(transport=transport) as client: + with pytest.raises(ValueError, match="non-http"): + await discover_server_cards("https://example.com", http_client=client) + + +async def test_discover_server_cards_ignores_non_mcp_entries() -> None: + agent_entry = CatalogEntry( + identifier="urn:example:a2a:research", + display_name="Research Assistant", + media_type="application/a2a-agent-card+json", + url="https://agents.example.com/researchAssistant", + ) + transport = httpx.ASGITransport(app=make_discovery_app(agent_entry, server_card_entry(CARD, CARD_URL))) + async with httpx.AsyncClient(transport=transport) as client: + cards = await discover_server_cards("https://example.com", http_client=client) + assert cards == [CARD] + + +async def test_discover_server_cards_falls_back_to_mcp_catalog_path() -> None: + app = make_discovery_app(catalog_path=MCP_CATALOG_WELL_KNOWN_PATH) # no /.well-known/ai-catalog.json + transport = httpx.ASGITransport(app=app) + async with httpx.AsyncClient(transport=transport) as client: + cards = await discover_server_cards("https://example.com", http_client=client) + assert cards == [CARD] + + +async def test_discover_server_cards_raises_when_no_catalog_exists() -> None: + app = Starlette(routes=[]) # 404 on both well-known paths + transport = httpx.ASGITransport(app=app) + async with httpx.AsyncClient(transport=transport) as client: + with pytest.raises(httpx.HTTPStatusError): + await discover_server_cards("https://example.com", http_client=client) + + +async def test_discover_server_cards_propagates_non_404_catalog_errors() -> None: + async def error(_request: object) -> Response: + return Response(status_code=500) + + app = Starlette(routes=[Route("/.well-known/ai-catalog.json", error, methods=["GET"])]) + transport = httpx.ASGITransport(app=app) + async with httpx.AsyncClient(transport=transport) as client: + with pytest.raises(httpx.HTTPStatusError) as excinfo: + await discover_server_cards("https://example.com", http_client=client) + assert excinfo.value.response.status_code == 500 + + +def test_load_server_card_from_file(tmp_path: Path) -> None: + path = tmp_path / "server-card.json" + path.write_text(json.dumps(CARD.model_dump(mode="json", by_alias=True, exclude_none=True)), encoding="utf-8") + assert load_server_card(path) == CARD diff --git a/tests/experimental/server_card/test_server.py b/tests/experimental/server_card/test_server.py new file mode 100644 index 000000000..d54fb85d6 --- /dev/null +++ b/tests/experimental/server_card/test_server.py @@ -0,0 +1,95 @@ +"""Tests for server-side Server Card generation and serving.""" + +from __future__ import annotations + +import httpx +import pytest +from starlette.applications import Starlette + +from mcp.client.experimental.server_card import fetch_server_card +from mcp.server.experimental.server_card import ( + build_server_card, + mount_server_card, + server_card_route, +) +from mcp.server.lowlevel import Server +from mcp.shared.experimental.server_card import Remote, Repository, ServerCard + +pytestmark = pytest.mark.anyio + +CARD_PATH = "/server-card.json" + + +def make_server() -> Server: + return Server( + "dice-roller", + version="1.0.0", + title="Dice Roller", + description="Rolls dice for tabletop games.", + website_url="https://example.com/dice", + ) + + +def test_build_server_card_from_server_identity() -> None: + card = build_server_card( + make_server(), + name="com.example/dice-roller", + remotes=[Remote(type="streamable-http", url="https://dice.example.com/mcp")], + repository=Repository(url="https://github.com/example/dice", source="github"), + meta={"com.example/x": 1}, + ) + assert card.name == "com.example/dice-roller" + assert card.version == "1.0.0" + assert card.title == "Dice Roller" + assert card.description == "Rolls dice for tabletop games." + assert card.website_url == "https://example.com/dice" + assert card.remotes is not None and card.remotes[0].url == "https://dice.example.com/mcp" + assert card.meta == {"com.example/x": 1} + + +def test_build_server_card_requires_version() -> None: + server = Server("no-version", description="desc") # version defaults to None + with pytest.raises(ValueError, match="version"): + build_server_card(server, name="example/no-version") + + +def test_build_server_card_requires_description() -> None: + server = Server("no-desc", version="1.0.0") # description defaults to None + with pytest.raises(ValueError, match="description"): + build_server_card(server, name="example/no-desc") + + +async def _get(app: Starlette, path: str) -> httpx.Response: + transport = httpx.ASGITransport(app=app) + async with httpx.AsyncClient(transport=transport, base_url="https://dice.example.com") as client: + return await client.get(path) + + +async def test_server_card_route_serves_card_with_discovery_headers() -> None: + card = build_server_card(make_server(), name="example/dice") + app = Starlette(routes=[server_card_route(card, path=CARD_PATH)]) + response = await _get(app, CARD_PATH) + assert response.status_code == 200 + assert response.headers["content-type"] == "application/mcp-server-card+json" + # Discovery requires CORS headers (MUST) and caching headers (SHOULD). + assert response.headers["access-control-allow-origin"] == "*" + assert response.headers["access-control-allow-methods"] == "GET" + assert response.headers["access-control-allow-headers"] == "Content-Type" + assert response.headers["cache-control"] == "public, max-age=3600" + assert response.text == card.model_dump_json(by_alias=True, exclude_none=True) + assert ServerCard.model_validate(response.json()) == card + + +async def test_mount_server_card_on_existing_app_and_client_fetch() -> None: + card = build_server_card( + make_server(), + name="example/dice", + remotes=[Remote(type="streamable-http", url="https://dice.example.com/mcp")], + ) + app = Starlette() + mount_server_card(app, card, path=CARD_PATH) + + transport = httpx.ASGITransport(app=app) + async with httpx.AsyncClient(transport=transport) as client: + fetched = await fetch_server_card(f"https://dice.example.com{CARD_PATH}", http_client=client) + assert fetched == card diff --git a/tests/experimental/server_card/test_types.py b/tests/experimental/server_card/test_types.py new file mode 100644 index 000000000..fbc57becd --- /dev/null +++ b/tests/experimental/server_card/test_types.py @@ -0,0 +1,103 @@ +"""Tests for Server Card models.""" + +from __future__ import annotations + +from typing import Any + +import pytest +from pydantic import ValidationError + +from mcp.shared.experimental.server_card import ( + SERVER_CARD_SCHEMA_URL, + KeyValueInput, + ServerCard, +) + +MINIMAL = { + "$schema": SERVER_CARD_SCHEMA_URL, + "name": "example-org/minimal", + "version": "1.0.0", + "description": "Smallest valid Server Card.", +} + +TEMPLATED_REMOTE = { + "$schema": SERVER_CARD_SCHEMA_URL, + "name": "example-org/with-remote", + "version": "2.1.0", + "description": "Server Card with a templated remote endpoint and headers.", + "title": "Example Remote Server", + "websiteUrl": "https://example.com", + "remotes": [ + { + "type": "streamable-http", + "url": "https://{tenant}.example.com/mcp", + "headers": [ + { + "name": "Authorization", + "description": "Bearer token for the remote endpoint.", + "isRequired": True, + "isSecret": True, + "value": "Bearer {token}", + "variables": {"token": {"isRequired": True, "isSecret": True}}, + } + ], + "variables": {"tenant": {"isRequired": True, "default": "default"}}, + "supportedProtocolVersions": ["2025-06-18", "2025-11-25"], + } + ], + "_meta": {"com.example/internal": {"tier": "gold"}}, +} + + +@pytest.mark.parametrize("doc", [MINIMAL, TEMPLATED_REMOTE]) +def test_server_card_round_trips(doc: dict[str, Any]) -> None: + card = ServerCard.model_validate(doc) + assert card.model_dump(mode="json", by_alias=True, exclude_none=True) == doc + + +def test_default_schema_url() -> None: + assert ServerCard(name="a/b", version="1.0.0", description="d").schema_uri == SERVER_CARD_SCHEMA_URL + + +def test_fields_settable_by_python_name_and_serialize_camelcase() -> None: + header = KeyValueInput(name="Authorization", is_required=True, value="Bearer {t}") + assert header.model_dump(by_alias=True, exclude_none=True) == { + "name": "Authorization", + "isRequired": True, + "value": "Bearer {t}", + } + + +@pytest.mark.parametrize( + "version", ["^1.2.3", "~1.2.3", ">=1.2.3", "1.x", "1.2.X", "1.*", "x", "*", "1.0.0 - 2.0.0", "1.0.0 || 2.0.0"] +) +def test_version_ranges_rejected(version: str) -> None: + with pytest.raises(ValidationError, match="exact version"): + ServerCard(name="a/b", version=version, description="d") + + +@pytest.mark.parametrize("version", ["1.0.0", "1.0.0-x", "1.0.0-X.1", "1.0.0-rc.x", "2024-01-05"]) +def test_exact_versions_accepted(version: str) -> None: + """Semver prereleases like 1.0.0-x are exact versions, not wildcards.""" + assert ServerCard(name="a/b", version=version, description="d").version == version + + +@pytest.mark.parametrize( + "doc, field", + [ + ({**MINIMAL, "name": "no-slash"}, "name"), + ( + {**MINIMAL, "$schema": "https://static.modelcontextprotocol.io/schemas/2025-11-25/server-card.schema.json"}, + "$schema", + ), + ( + {**MINIMAL, "$schema": "https://static.modelcontextprotocol.io/schemas/v1/server.schema.json"}, + "$schema", + ), + ({**MINIMAL, "description": ""}, "description"), + ], +) +def test_invalid_cards_rejected(doc: dict[str, Any], field: str) -> None: + with pytest.raises(ValidationError) as excinfo: + ServerCard.model_validate(doc) + assert field in str(excinfo.value)