diff --git a/docs/advanced/server-cards.md b/docs/advanced/server-cards.md new file mode 100644 index 0000000000..a158280728 --- /dev/null +++ b/docs/advanced/server-cards.md @@ -0,0 +1,130 @@ +# Server Cards + +!!! warning "Experimental" + Server Cards are an experimental MCP extension tracking + [SEP-2127](https://github.com/modelcontextprotocol/modelcontextprotocol/pull/2127). + Everything on this page lives under `experimental` modules and may change or be + removed in any release without a deprecation cycle. + +A **Server Card** is a static JSON document that describes a single remote MCP server +well enough for a client to discover it and connect, before any protocol exchange: +identity (`name`, `version`, `description`, `title`, `icons`, `repository`, +`websiteUrl`), the transport endpoints (`remotes[]` with URL templates, header inputs +and supported protocol versions), and namespaced `_meta` for anything else. It is served +as `application/mcp-server-card+json`. + +A card deliberately omits two things. Tools, resources and prompts stay behind the +runtime `list` operations. Local install metadata (packages, registries, arguments, +environment) belongs to the MCP Registry's `server.json`. If you need install hints, +put them in `_meta`. + +## Serving a card + +The spec reserves `GET /server-card` as the default location, +and domain-level discovery reads an **AI Catalog** at `/.well-known/ai-catalog.json`. +`mount_discovery` sets up both on a single-domain deployment: + +```python title="serve_card.py" hl_lines="12 19" +--8<-- "docs_src/server_cards/tutorial001.py" +``` + +`build_server_card` derives `title`, `description`, `version`, `websiteUrl` and `icons` +from the server object, so the card stays consistent with what `serverInfo` reports at +runtime. Explicit keyword arguments override the derived values. The namespaced card +`name` and the public `remotes` URLs are yours to supply, since the server object cannot +know them. + +With the app above, `GET /mcp/server-card` and `GET /.well-known/ai-catalog.json` both +answer with the spec's required headers: the correct `Content-Type`, the CORS headers +the spec mandates on card endpoints, a `Cache-Control: public, max-age=3600` default, +and a strong `ETag` that turns a matching `If-None-Match` into an empty `304`. + +The card routes are appended to the app after `streamable_http_app()` builds it. Mount +them outside any auth middleware. Discovery is unauthenticated by design, and a card or +catalog must never contain credentials, internal topology or private endpoints. + +`public_url` is the externally visible base URL at which the app's root is served, +typically just the origin. Only include a path prefix when a reverse proxy really serves +the app under it, since the catalog entry advertises `public_url` plus the card path. + +If you need only one of the two endpoints, or different paths, use the smaller pieces: +`create_server_card_routes` / `mount_server_card` for the card and +`create_ai_catalog_routes` / `mount_ai_catalog` for the catalog. For fully custom +hosting (say a FastAPI route), `discovery_response` is the compliance chokepoint that +produces a correct response from a request and the document bytes. + +## Publishing on your brand domain + +The catalog belongs on the domain users associate with your service, which is often +different from the API host (think `github.com` versus `api.githubcopilot.com`). In that +case skip `mount_discovery` and build the entry yourself with `server_card_entry`, then +publish it in the catalog your brand domain serves. Entries can point at the hosted card +by URL or inline the whole card as `data`. + +A card is just JSON, so static publishing needs no server at all: + +```python title="publish_static.py" +--8<-- "docs_src/server_cards/tutorial004.py" +``` + +Upload the result to your CDN and reference it from your catalog. + +## Discovering and connecting + +`discover_server_cards` is one probe: it fetches the well-known catalog of the URL's +origin, follows Server Card entries (by URL or inline `data`) and nested catalogs, and +returns every card it found. It never runs implicitly. Your host decides when to probe. + +```python title="discover_and_connect.py" hl_lines="10 16" +--8<-- "docs_src/server_cards/tutorial002.py" +``` + +`resolve_remote` substitutes the card's `{curly_brace}` variables using declared +defaults and your values, and raises a `ValueError` naming every missing required input, +so you can prompt for all of them at once. `Remote.required_variables` lists them up +front. + +The probe collects per-entry problems in `DiscoveryResult.failures` instead of raising, +so one hostile or broken entry never hides the rest. Each `CardListing` carries the +listing chain for your consent UI: `listing_domain` is where the catalog listed the +card, `hosting_domain` is where the card itself lives, and they legitimately differ. + +For local development the hardened defaults get in the way (plain http and loopback +targets are blocked). Opt in explicitly with +`DiscoveryPolicy(allow_private_addresses=True)`. + +## Caching and revalidation + +Clients should honor `Cache-Control` and avoid polling. The SDK deliberately ships no +cache storage. Instead, the stateless request/parse pairs let your host revalidate with +its own store: + +```python title="revalidate.py" hl_lines="16 18" +--8<-- "docs_src/server_cards/tutorial003.py" +``` + +Store the `ETag` alongside the card and send it back as `If-None-Match`. An unchanged +card costs a 304 and no body. + +## Security model + +Cards are unverified, advisory input. The rules that keep discovery safe: + +- **Runtime wins.** A card's claims should match the live server, and + `reconcile_server_card` reports any drift, but clients must never treat card contents + as authoritative for security or access-control decisions. +- **De-duplicate on endpoints.** `name` and the catalog `identifier` are self-asserted + and spoofable. `ServerCard.endpoint_urls()` gives you the dedup key: the card's actual + `remotes[]` URLs. +- **Never auto-install.** Consent, its persistence ("not now", "this session", + "always"), decline memory and enterprise allowlists are host application policy. The + SDK exposes the data and stays out of the decision. +- **Hardened fetching.** Every fetch, redirect hop and nested catalog follow is checked + under `DiscoveryPolicy`: https only (plain http needs `allow_private_addresses=True`), + no userinfo in URLs, an SSRF guard that rejects private, loopback, link-local and + metadata addresses (checked again after DNS resolution), bounded redirects, response + size and entry count caps, a whole-probe entry budget with already-visited catalogs + never refetched, and no cookies or ambient credentials. If you pass your own + `http_client`, keep it credential-free. The guard resolves DNS before the request + while the client re-resolves on connect, so a DNS rebinding race remains possible + between the two. diff --git a/docs_src/server_cards/__init__.py b/docs_src/server_cards/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/docs_src/server_cards/tutorial001.py b/docs_src/server_cards/tutorial001.py new file mode 100644 index 0000000000..33c1056853 --- /dev/null +++ b/docs_src/server_cards/tutorial001.py @@ -0,0 +1,19 @@ +from mcp.server import MCPServer +from mcp.server.experimental.server_card import build_server_card, mount_discovery +from mcp.shared.experimental.server_card import Remote + +mcp = MCPServer( + name="weather", + version="1.4.0", + description="Hourly forecasts.", + website_url="https://example.com", +) + +card = build_server_card( + mcp, + name="com.example/weather", + remotes=[Remote(type="streamable-http", url="https://mcp.example.com/mcp")], +) + +app = mcp.streamable_http_app() +mount_discovery(app, card, public_url="https://mcp.example.com") diff --git a/docs_src/server_cards/tutorial002.py b/docs_src/server_cards/tutorial002.py new file mode 100644 index 0000000000..f45799b994 --- /dev/null +++ b/docs_src/server_cards/tutorial002.py @@ -0,0 +1,22 @@ +import httpx2 + +from mcp import Client +from mcp.client.experimental.server_card import discover_server_cards, reconcile_server_card +from mcp.client.streamable_http import streamable_http_client +from mcp.shared.experimental.server_card import resolve_remote + + +async def main() -> None: + result = await discover_server_cards("https://example.com/docs") + for listing in result.listings: + print(listing.entry.identifier, "listed on", listing.listing_domain, "hosted at", listing.hosting_domain) + + chosen = result.listings[0] # your host app: consent UI, dedup on chosen.card.endpoint_urls() + assert chosen.card.remotes is not None + resolved = resolve_remote(chosen.card.remotes[0], {"token": "..."}) # ValueError names missing inputs + + async with httpx2.AsyncClient(headers=resolved.headers, follow_redirects=True) as http_client: + transport = streamable_http_client(resolved.url, http_client=http_client) + async with Client(transport) as client: + for mismatch in reconcile_server_card(chosen.card, client.server_info): # advisory: runtime wins + print("card mismatch:", mismatch.field, mismatch.card_value, mismatch.runtime_value) diff --git a/docs_src/server_cards/tutorial003.py b/docs_src/server_cards/tutorial003.py new file mode 100644 index 0000000000..29d119e772 --- /dev/null +++ b/docs_src/server_cards/tutorial003.py @@ -0,0 +1,22 @@ +import httpx2 + +from mcp.client.experimental.server_card import create_server_card_request, parse_server_card_response +from mcp.shared.experimental.server_card import ServerCard + + +class CardStore: + """Your host's cache. Anything that keeps a card and its ETag per URL works.""" + + def __init__(self) -> None: + self.cards: dict[str, ServerCard] = {} + self.etags: dict[str, str] = {} + + +async def refresh(store: CardStore, http: httpx2.AsyncClient, url: str) -> ServerCard: + request = create_server_card_request(url, if_none_match=store.etags.get(url)) + response = await http.send(request) + if response.status_code != 304: # an unchanged card costs a 304 + store.cards[url] = parse_server_card_response(response) + if "etag" in response.headers: + store.etags[url] = response.headers["etag"] + return store.cards[url] diff --git a/docs_src/server_cards/tutorial004.py b/docs_src/server_cards/tutorial004.py new file mode 100644 index 0000000000..0582837eb8 --- /dev/null +++ b/docs_src/server_cards/tutorial004.py @@ -0,0 +1,16 @@ +from pathlib import Path + +from mcp.shared.experimental.server_card import Remote, ServerCard + +card = ServerCard( + name="com.example/weather", + version="1.4.0", + description="Hourly forecasts.", + title="Weather", + website_url="https://example.com", + remotes=[Remote(type="streamable-http", url="https://mcp.example.com/mcp")], +) + + +def publish(path: Path) -> None: + path.write_text(card.model_dump_json(by_alias=True, exclude_none=True)) diff --git a/mkdocs.yml b/mkdocs.yml index 4c0cd06ad7..ef26cd8a20 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -67,6 +67,7 @@ nav: - Middleware: advanced/middleware.md - Extensions: advanced/extensions.md - MCP Apps: advanced/apps.md + - Server Cards: advanced/server-cards.md - Troubleshooting: troubleshooting.md - Migration Guide: migration.md - API Reference: api/ diff --git a/src/mcp/client/experimental/__init__.py b/src/mcp/client/experimental/__init__.py new file mode 100644 index 0000000000..b0a07321ca --- /dev/null +++ b/src/mcp/client/experimental/__init__.py @@ -0,0 +1,6 @@ +"""Experimental: tracks SEP-2127 (the Server Card extension). + +Nothing under `mcp.client.experimental` carries a stability guarantee. Any +symbol here may change or be removed in any release without a deprecation +cycle. +""" diff --git a/src/mcp/client/experimental/_discovery_http.py b/src/mcp/client/experimental/_discovery_http.py new file mode 100644 index 0000000000..f80c59e847 --- /dev/null +++ b/src/mcp/client/experimental/_discovery_http.py @@ -0,0 +1,247 @@ +"""Hardened HTTP core for discovery fetches. Private module. + +Every network fetch (including every redirect hop and every nested catalog +follow) is re-admitted under the same rules: http(s) scheme only, no userinfo, +plain http only under `allow_private_addresses`, an SSRF address guard checked +before the request, bounded redirects, a streamed response size cap, and a +media type check. +""" + +import ipaddress +from dataclasses import dataclass +from typing import Literal +from urllib.parse import urljoin, urlsplit + +import anyio +import httpx2 + +from mcp.shared._httpx_utils import create_mcp_http_client +from mcp.shared.experimental.ai_catalog import MAX_CATALOG_NESTING_DEPTH + +DiscoveryErrorReason = Literal[ + "status", + "media_type", + "response_too_large", + "too_many_redirects", + "blocked_address", + "insecure_transport", + "invalid_entry", + "catalog_depth", + "probe_budget", +] + +_REDIRECT_STATUSES = frozenset({301, 302, 303, 307, 308}) +_CGNAT_NETWORK = ipaddress.ip_network("100.64.0.0/10") + + +@dataclass(frozen=True, slots=True, kw_only=True) +class DiscoveryPolicy: + """Limits applied to every discovery fetch. + + The defaults are the hardened production posture. Set + `allow_private_addresses=True` for local development, which permits plain + http and skips the address guard entirely. + + The address guard resolves DNS before the request and re-checks every + redirect hop, but the HTTP client re-resolves when it connects, so a + DNS rebinding race remains possible between the check and the connect. + + `max_catalog_entries` caps one catalog document and `max_catalog_depth` + caps nesting, but neither bounds their product, so `max_probe_entries` is + the aggregate budget: the total card and nested-catalog entries one + discovery probe will process across every catalog it walks. A probe that + exhausts it records a single `probe_budget` failure and returns what it + has, which also bounds the probe's total fetches and worst-case duration + (`max_probe_entries * timeout_seconds`). + """ + + max_response_bytes: int = 1_048_576 + max_redirects: int = 3 + max_catalog_entries: int = 100 + max_catalog_depth: int = MAX_CATALOG_NESTING_DEPTH + max_probe_entries: int = 500 + allow_private_addresses: bool = False + timeout_seconds: float = 10.0 + + +class DiscoveryError(Exception): + """A transport or policy failure during discovery. + + `reason` states which rule failed and `url` names the offending target. + Document shape failures raise `pydantic.ValidationError` instead, and raw + connection failures stay `httpx2.HTTPError`. + """ + + url: str + reason: DiscoveryErrorReason + + def __init__(self, message: str, *, url: str, reason: DiscoveryErrorReason) -> None: + super().__init__(message) + self.url = url + self.reason = reason + + +def accept_header(media_type: str) -> str: + """The Accept header for one discovery media type. + + The canonical type is preferred, with plain `application/json` at a lower + quality because static hosts and CDNs commonly serve it. This is the one + place the lenience is written down; `check_media_type` mirrors it. + """ + return f"{media_type}, application/json;q=0.5" + + +def check_status(response: httpx2.Response, url: str) -> None: + """Reject any non-2xx response. + + Raises: + DiscoveryError: With `reason="status"`, chaining the underlying + `httpx2.HTTPStatusError`. + """ + try: + response.raise_for_status() + except httpx2.HTTPStatusError as exc: + raise DiscoveryError(f"unexpected status {response.status_code} from {url}", url=url, reason="status") from exc + + +def check_media_type(response: httpx2.Response, url: str, expected: str) -> None: + """Reject a response whose media type is neither `expected` nor JSON. + + Plain `application/json` is accepted because static hosts and CDNs + commonly serve it. + + Raises: + DiscoveryError: With `reason="media_type"`. + """ + content_type = response.headers.get("content-type", "") + media_type = content_type.split(";", 1)[0].strip().lower() + if media_type not in (expected, "application/json"): + raise DiscoveryError(f"unexpected media type {content_type!r} from {url}", url=url, reason="media_type") + + +async def _host_addresses(host: str) -> list[ipaddress.IPv4Address | ipaddress.IPv6Address]: + """The IP addresses `host` names: the literal itself, or its DNS resolution.""" + try: + return [ipaddress.ip_address(host)] + except ValueError: + infos = await anyio.getaddrinfo(host, None) + # sockaddr[0] is the address; IPv6 link-local entries carry a %scope suffix. + return [ipaddress.ip_address(info[4][0].split("%", 1)[0]) for info in infos] + + +def _is_blocked_address(address: ipaddress.IPv4Address | ipaddress.IPv6Address) -> bool: + """Whether `address` is off-limits for discovery: anything not publicly routable.""" + if isinstance(address, ipaddress.IPv6Address) and address.ipv4_mapped is not None: + # Judge `::ffff:a.b.c.d` by its embedded IPv4 rules. Relying on the v6 + # properties would miss CGNAT everywhere and, before the gh-113171 + # ipaddress fix, the private v4 ranges too. + address = address.ipv4_mapped + if address.version == 4 and address in _CGNAT_NETWORK: + return True + # is_private covers loopback, RFC 1918, ULA and the unspecified address. + return address.is_private or address.is_link_local or address.is_multicast or address.is_reserved + + +async def _admit_url(url: str, policy: DiscoveryPolicy) -> None: + """Apply the scheme rule and the SSRF address guard to one fetch target. + + Raises: + DiscoveryError: `reason="insecure_transport"` for a non-http(s) URL, a + URL carrying userinfo, or plain http under the hardened policy, + `reason="blocked_address"` when the host is (or resolves to) a + non-public address. + """ + parts = urlsplit(url) + host = parts.hostname + if parts.scheme not in ("http", "https") or not host: + raise DiscoveryError( + f"discovery requires an absolute http(s) URL, got {url!r}", url=url, reason="insecure_transport" + ) + if "@" in parts.netloc: + # Discovery never sends credentials, and `user@host` URLs exist mainly + # to make a hostile target read as a trusted brand in consent UI. + raise DiscoveryError( + f"discovery URLs must not carry userinfo, got {url!r}", url=url, reason="insecure_transport" + ) + if policy.allow_private_addresses: + return + if parts.scheme == "http": + # Under the hardened policy loopback targets are blocked by the + # address guard anyway, so plain http (loopback included) never + # survives it; local development opts in via allow_private_addresses. + raise DiscoveryError( + f"plain http is only allowed with allow_private_addresses=True, got {url!r}", + url=url, + reason="insecure_transport", + ) + for address in await _host_addresses(host): + if _is_blocked_address(address): + raise DiscoveryError(f"{url!r} points at blocked address {address}", url=url, reason="blocked_address") + + +async def _read_limited(response: httpx2.Response, url: str, policy: DiscoveryPolicy) -> bytes: + """Stream the body, failing as soon as it exceeds the size cap.""" + chunks: list[bytes] = [] + total = 0 + async for chunk in response.aiter_bytes(): + total += len(chunk) + if total > policy.max_response_bytes: + raise DiscoveryError( + f"response from {url} exceeds {policy.max_response_bytes} bytes", url=url, reason="response_too_large" + ) + chunks.append(chunk) + return b"".join(chunks) + + +async def _fetch(client: httpx2.AsyncClient, url: str, media_type: str, policy: DiscoveryPolicy) -> bytes: + """One admitted fetch, walking redirects manually so each hop is re-checked.""" + with anyio.fail_after(policy.timeout_seconds): + current = url + for _ in range(policy.max_redirects + 1): + await _admit_url(current, policy) + headers = {"Accept": accept_header(media_type)} + request = client.build_request("GET", current, headers=headers) + response = await client.send(request, stream=True, follow_redirects=False) + try: + if response.status_code in _REDIRECT_STATUSES: + location = response.headers.get("location") + if location is None: + raise DiscoveryError( + f"redirect from {current} carries no Location", url=current, reason="status" + ) + current = urljoin(current, location) + continue + check_status(response, current) + check_media_type(response, current, media_type) + return await _read_limited(response, current, policy) + finally: + await response.aclose() + raise DiscoveryError( + f"more than {policy.max_redirects} redirects while fetching {url}", url=current, reason="too_many_redirects" + ) + + +async def fetch_discovery_document( + url: str, + media_type: str, + *, + http_client: httpx2.AsyncClient | None = None, + policy: DiscoveryPolicy | None = None, +) -> bytes: + """Fetch one discovery document under `policy`, returning its raw bytes. + + With `http_client=None` a fresh client is created per call, so no cookies + or ambient credentials ever accompany a discovery request. A caller-owned + client still gets every URL, redirect, size and media type check. + + Raises: + DiscoveryError: When any policy rule fails. + httpx2.HTTPError: For raw connection failures. + OSError: When DNS resolution itself fails. + TimeoutError: When the fetch exceeds `policy.timeout_seconds`. + """ + resolved_policy = policy if policy is not None else DiscoveryPolicy() + if http_client is None: + async with create_mcp_http_client() as own_client: + return await _fetch(own_client, url, media_type, resolved_policy) + return await _fetch(http_client, url, media_type, resolved_policy) diff --git a/src/mcp/client/experimental/server_card.py b/src/mcp/client/experimental/server_card.py new file mode 100644 index 0000000000..e9adf0f6e6 --- /dev/null +++ b/src/mcp/client/experimental/server_card.py @@ -0,0 +1,472 @@ +"""Discover, fetch and reconcile Server Cards (experimental, tracks SEP-2127). + +Discovery is host-invoked only. Nothing here runs implicitly, connects to a +discovered server, persists anything or asks for consent. Those are host +application decisions. `CardListing` carries the listing chain (listing +domain versus hosting domain) for consent UI, and +`ServerCard.endpoint_urls()` is the dedup key. + +Card contents are advisory. Runtime values win, and cards MUST NOT drive +security or access-control decisions. +""" + +import os +from dataclasses import dataclass +from pathlib import Path +from urllib.parse import urlsplit + +import httpx2 +import pydantic +from mcp_types import Implementation + +from mcp.client.experimental._discovery_http import ( + DiscoveryError, + DiscoveryErrorReason, + DiscoveryPolicy, + accept_header, + check_media_type, + check_status, + fetch_discovery_document, +) +from mcp.shared._httpx_utils import create_mcp_http_client +from mcp.shared.experimental.ai_catalog import ( + AI_CATALOG_MEDIA_TYPE, + AI_CATALOG_WELL_KNOWN_PATH, + AICatalog, + CatalogEntry, +) +from mcp.shared.experimental.server_card import RESERVED_SERVER_CARD_SUFFIX, SERVER_CARD_MEDIA_TYPE, ServerCard + +__all__ = [ + "DiscoveryPolicy", + "DiscoveryError", + "DiscoveryErrorReason", + "CardListing", + "DiscoveryFailure", + "DiscoveryResult", + "CardMismatch", + "fetch_server_card", + "fetch_ai_catalog", + "discover_server_cards", + "load_server_card", + "well_known_ai_catalog_url", + "server_card_url", + "create_server_card_request", + "parse_server_card_response", + "create_ai_catalog_request", + "parse_ai_catalog_response", + "reconcile_server_card", +] + +_SERVER_CARD_ACCEPT = accept_header(SERVER_CARD_MEDIA_TYPE) +_AI_CATALOG_ACCEPT = accept_header(AI_CATALOG_MEDIA_TYPE) + + +def _display_host(url: str) -> str: + """The host (plus `:port` when present) of `url`, never its userinfo. + + `netloc` would include a `user@` prefix, which a hostile catalog can use + to make `github.com@evil.example` read as a trusted brand in consent UI. + """ + parts = urlsplit(url) + host = parts.hostname or "" + if ":" in host: # bare IPv6 from .hostname; re-bracket so the port reads unambiguously + host = f"[{host}]" + return f"{host}:{parts.port}" if parts.port is not None else host + + +@dataclass(frozen=True, slots=True) +class CardListing: + """One discovered card together with where it was listed and hosted. + + A listing is an unverified assertion. Show both domains in consent UI: + they differ whenever a catalog lists a card hosted elsewhere. + """ + + card: ServerCard + entry: CatalogEntry + catalog_url: str + card_url: str | None + + @property + def listing_domain(self) -> str: + """Host (plus `:port` when present) of the catalog that listed the card.""" + return _display_host(self.catalog_url) + + @property + def hosting_domain(self) -> str | None: + """Host the card was fetched from, or None for an inline `data` entry.""" + return _display_host(self.card_url) if self.card_url is not None else None + + +@dataclass(frozen=True, slots=True) +class DiscoveryFailure: + """One catalog entry (or nested catalog) that could not be turned into a listing.""" + + url: str | None + entry_identifier: str | None + error: Exception + + +@dataclass(frozen=True, slots=True) +class DiscoveryResult: + """Everything one discovery probe produced. + + A bad entry never kills the probe. It lands in `failures` while the + other entries still produce `listings`. + """ + + listings: list[CardListing] + failures: list[DiscoveryFailure] + + +@dataclass(frozen=True, slots=True) +class CardMismatch: + """One advisory discrepancy between a card claim and a runtime value.""" + + field: str + card_value: str | None + runtime_value: str | None + + +def well_known_ai_catalog_url(url: str) -> str: + """The well-known catalog URL for the origin of any http(s) URL. + + Raises: + ValueError: If `url` is not absolute http(s). + """ + 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 f"{parts.scheme}://{parts.netloc}{AI_CATALOG_WELL_KNOWN_PATH}" + + +def server_card_url(streamable_http_url: str) -> str: + """The spec-reserved card URL for a streamable HTTP transport URL. + + The suffix is appended to the transport URL, not the domain root: + `https://host/mcp` becomes `https://host/mcp/server-card`. + + Raises: + ValueError: If `streamable_http_url` is not absolute http(s). + """ + parts = urlsplit(streamable_http_url) + if parts.scheme not in ("http", "https") or not parts.netloc: + raise ValueError(f"expected an absolute http(s) URL, got {streamable_http_url!r}") + return f"{parts.scheme}://{parts.netloc}{parts.path.rstrip('/')}{RESERVED_SERVER_CARD_SUFFIX}" + + +def load_server_card(path: str | os.PathLike[str]) -> ServerCard: + """Parse a Server Card from a local file. No network is involved. + + Raises: + OSError: If the file cannot be read. + pydantic.ValidationError: If the document is not a valid card. + """ + return ServerCard.model_validate_json(Path(path).read_bytes()) + + +async def fetch_server_card( + url: str, + *, + http_client: httpx2.AsyncClient | None = None, + policy: DiscoveryPolicy | None = None, +) -> ServerCard: + """Fetch and parse a Server Card from `url` under `policy`. + + A missing `$schema` is defaulted on ingestion. A wrong one is rejected. + Avoid passing an `http_client` that carries cookies or ambient + credentials. Discovery requests must never send any. + + Raises: + DiscoveryError: When a policy rule fails (status, media type, size, + redirects, blocked address, insecure transport). + pydantic.ValidationError: If the document is not a valid card. + httpx2.HTTPError: For raw connection failures. + OSError: When DNS resolution fails (an unresolvable host raises + `socket.gaierror` from the address guard, before any request). + TimeoutError: When the fetch exceeds `policy.timeout_seconds`. + """ + body = await fetch_discovery_document(url, SERVER_CARD_MEDIA_TYPE, http_client=http_client, policy=policy) + return ServerCard.model_validate_json(body) + + +async def fetch_ai_catalog( + url: str, + *, + http_client: httpx2.AsyncClient | None = None, + policy: DiscoveryPolicy | None = None, +) -> AICatalog: + """Fetch and parse an AI Catalog from `url` under `policy`. + + Raises: + DiscoveryError: When a policy rule fails. + pydantic.ValidationError: If the document is not a valid catalog. + httpx2.HTTPError: For raw connection failures. + OSError: When DNS resolution fails (an unresolvable host raises + `socket.gaierror` from the address guard, before any request). + TimeoutError: When the fetch exceeds `policy.timeout_seconds`. + """ + body = await fetch_discovery_document(url, AI_CATALOG_MEDIA_TYPE, http_client=http_client, policy=policy) + return AICatalog.model_validate_json(body) + + +@dataclass(slots=True) +class _ProbeState: + """Mutable per-probe budget shared by every catalog one probe walks. + + `max_catalog_entries` and `max_catalog_depth` cap one document and the + nesting, but their product is multiplicative, so a hostile catalog tree + could otherwise amplify one probe into ~`entries**depth` fetches. The + aggregate entry budget and the visited set bound the whole walk. + """ + + entries_remaining: int + visited_catalog_urls: set[str] + exhausted: bool = False + + def take_entry(self, catalog_url: str, policy: DiscoveryPolicy, failures: list[DiscoveryFailure]) -> bool: + """Consume one unit of budget, recording a single failure when it runs out.""" + if self.entries_remaining > 0: + self.entries_remaining -= 1 + return True + self.exhausted = True + error = DiscoveryError( + f"probe exceeded the budget of {policy.max_probe_entries} catalog entries", + url=catalog_url, + reason="probe_budget", + ) + failures.append(DiscoveryFailure(url=catalog_url, entry_identifier=None, error=error)) + return False + + +# The per-entry safety net: one hostile or broken entry becomes a failure, never +# a probe-killer. OSError subsumes socket.gaierror (the address guard resolves +# DNS itself) and TimeoutError (the per-fetch deadline). +_PER_ENTRY_ERRORS = (DiscoveryError, httpx2.HTTPError, OSError, pydantic.ValidationError) + + +async def _collect_card( + client: httpx2.AsyncClient, + entry: CatalogEntry, + catalog_url: str, + policy: DiscoveryPolicy, + listings: list[CardListing], + failures: list[DiscoveryFailure], +) -> None: + """Turn one card entry into a listing, or record why it could not be.""" + try: + if entry.url is not None: + card = await fetch_server_card(entry.url, http_client=client, policy=policy) + card_url = entry.url + else: + card = ServerCard.model_validate(entry.data) + card_url = None + except _PER_ENTRY_ERRORS as error: + failures.append(DiscoveryFailure(url=entry.url, entry_identifier=entry.identifier, error=error)) + return + listings.append(CardListing(card=card, entry=entry, catalog_url=catalog_url, card_url=card_url)) + + +async def _collect_nested_catalog( + client: httpx2.AsyncClient, + entry: CatalogEntry, + catalog_url: str, + depth: int, + policy: DiscoveryPolicy, + state: _ProbeState, + listings: list[CardListing], + failures: list[DiscoveryFailure], +) -> None: + """Follow one nested catalog entry, or record why it could not be.""" + if entry.url is not None and entry.url in state.visited_catalog_urls: + return # already walked: a legitimate duplicate adds nothing, a cycle never terminates + if depth > policy.max_catalog_depth: + error = DiscoveryError( + f"nested catalog exceeds the depth cap of {policy.max_catalog_depth}", + url=entry.url if entry.url is not None else catalog_url, + reason="catalog_depth", + ) + failures.append(DiscoveryFailure(url=entry.url, entry_identifier=entry.identifier, error=error)) + return + try: + if entry.url is not None: + state.visited_catalog_urls.add(entry.url) + nested = await fetch_ai_catalog(entry.url, http_client=client, policy=policy) + nested_url = entry.url + else: + nested = AICatalog.model_validate(entry.data) + nested_url = catalog_url + except _PER_ENTRY_ERRORS as error: + failures.append(DiscoveryFailure(url=entry.url, entry_identifier=entry.identifier, error=error)) + return + await _walk_catalog(client, nested, nested_url, depth, policy, state, listings, failures) + + +async def _walk_catalog( + client: httpx2.AsyncClient, + catalog: AICatalog, + catalog_url: str, + depth: int, + policy: DiscoveryPolicy, + state: _ProbeState, + listings: list[CardListing], + failures: list[DiscoveryFailure], +) -> None: + """Collect card listings from one catalog document, recursing into nested ones.""" + entries = catalog.entries + if len(entries) > policy.max_catalog_entries: + error = DiscoveryError( + f"catalog lists {len(entries)} entries; only the first {policy.max_catalog_entries} were processed", + url=catalog_url, + reason="invalid_entry", + ) + failures.append(DiscoveryFailure(url=catalog_url, entry_identifier=None, error=error)) + entries = entries[: policy.max_catalog_entries] + for entry in entries: + if entry.type not in (SERVER_CARD_MEDIA_TYPE, AI_CATALOG_MEDIA_TYPE): + continue # Other artifact types are not failures. Catalogs legitimately advertise them. + if state.exhausted: + return # a nested walk already recorded the budget failure; add no more noise + if not state.take_entry(catalog_url, policy, failures): + return + if entry.type == SERVER_CARD_MEDIA_TYPE: + await _collect_card(client, entry, catalog_url, policy, listings, failures) + else: + await _collect_nested_catalog(client, entry, catalog_url, depth + 1, policy, state, listings, failures) + + +async def _probe( + client: httpx2.AsyncClient, + catalog_url: str, + policy: DiscoveryPolicy, +) -> DiscoveryResult: + """Fetch the top-level catalog and walk it under one shared budget.""" + catalog = await fetch_ai_catalog(catalog_url, http_client=client, policy=policy) + state = _ProbeState(entries_remaining=policy.max_probe_entries, visited_catalog_urls={catalog_url}) + listings: list[CardListing] = [] + failures: list[DiscoveryFailure] = [] + await _walk_catalog(client, catalog, catalog_url, 1, policy, state, listings, failures) + return DiscoveryResult(listings=listings, failures=failures) + + +async def discover_server_cards( + url: str, + *, + http_client: httpx2.AsyncClient | None = None, + policy: DiscoveryPolicy | None = None, +) -> DiscoveryResult: + """One discovery probe: the well-known catalog of `url`'s origin, then its cards. + + Any user-entered URL works. Only its origin is used. The probe fetches + the catalog, follows card entries (by URL or inline `data`) and nested + catalogs up to `policy.max_catalog_depth`, and collects per-entry + failures instead of raising them. Already-visited catalog URLs are never + refetched, and `policy.max_probe_entries` bounds the whole walk, so a + hostile catalog tree cannot amplify one probe into unbounded fetches. + Nothing is deduplicated across listings, connected to or persisted. + Enterprise controls (disabling or allowlisting discovery) stay trivial + because the probe only runs when the host calls it. + + Raises: + ValueError: If `url` is not absolute http(s). + DiscoveryError: When fetching the top-level catalog itself fails a + policy rule. Per-entry failures are collected, never raised. + pydantic.ValidationError: If the top-level catalog is malformed. + httpx2.HTTPError: For raw connection failures on the top-level fetch. + OSError: When DNS resolution fails for the top-level catalog host + (an unresolvable host raises `socket.gaierror`). + TimeoutError: When the top-level fetch exceeds `policy.timeout_seconds`. + """ + resolved_policy = policy if policy is not None else DiscoveryPolicy() + catalog_url = well_known_ai_catalog_url(url) + if http_client is None: + # One fresh credential-free client for the whole walk, so a large + # catalog reuses connections instead of a TLS handshake per entry. + async with create_mcp_http_client() as own_client: + return await _probe(own_client, catalog_url, resolved_policy) + return await _probe(http_client, catalog_url, resolved_policy) + + +def create_server_card_request(url: str, *, if_none_match: str | None = None) -> httpx2.Request: + """A GET request for a card, with the Accept header and optional `If-None-Match`. + + Together with `parse_server_card_response` this is the revalidation + toolkit for hosts that keep their own cache: send the stored ETag, and an + unchanged card costs a 304. The SDK deliberately ships no cache storage. + """ + headers = {"Accept": _SERVER_CARD_ACCEPT} + if if_none_match is not None: + headers["If-None-Match"] = if_none_match + return httpx2.Request("GET", url, headers=headers) + + +def parse_server_card_response(response: httpx2.Response) -> ServerCard: + """Parse a card from a response the caller transported. + + Only status, media type and document shape are checked here. The caller + owns the transport, so no size or address rules apply. A 304 raises. + Branch on `response.status_code == 304` before parsing. + + Raises: + DiscoveryError: For a non-2xx status (including 304) or a wrong media type. + pydantic.ValidationError: If the document is not a valid card. + """ + url = str(response.request.url) + check_status(response, url) + check_media_type(response, url, SERVER_CARD_MEDIA_TYPE) + return ServerCard.model_validate_json(response.content) + + +def create_ai_catalog_request(url: str, *, if_none_match: str | None = None) -> httpx2.Request: + """A GET request for a catalog, with the Accept header and optional `If-None-Match`.""" + headers = {"Accept": _AI_CATALOG_ACCEPT} + if if_none_match is not None: + headers["If-None-Match"] = if_none_match + return httpx2.Request("GET", url, headers=headers) + + +def parse_ai_catalog_response(response: httpx2.Response) -> AICatalog: + """Parse a catalog from a response the caller transported. + + Raises: + DiscoveryError: For a non-2xx status (including 304) or a wrong media type. + pydantic.ValidationError: If the document is not a valid catalog. + """ + url = str(response.request.url) + check_status(response, url) + check_media_type(response, url, AI_CATALOG_MEDIA_TYPE) + return AICatalog.model_validate_json(response.content) + + +def reconcile_server_card( + card: ServerCard, + server_info: Implementation, + *, + protocol_version: str | None = None, +) -> list[CardMismatch]: + """Compare card claims to live values. Advisory only, never raises. + + Call it with `Client.server_info` after connecting. It returns + discrepancies for logging or UI. Runtime values MUST win, and cards MUST + NOT drive security or access-control decisions. The card name matches + when `server_info.name` equals either the full namespaced name or its + post-slash local part. + """ + mismatches: list[CardMismatch] = [] + local_name = card.name.split("/", 1)[1] + if server_info.name not in (card.name, local_name): + mismatches.append(CardMismatch(field="name", card_value=card.name, runtime_value=server_info.name)) + if server_info.version != card.version: + mismatches.append(CardMismatch(field="version", card_value=card.version, runtime_value=server_info.version)) + if protocol_version is not None: + declared = {version for remote in card.remotes or [] for version in remote.supported_protocol_versions or []} + if declared and protocol_version not in declared: + mismatches.append( + CardMismatch( + field="protocol_versions", + card_value=", ".join(sorted(declared)), + runtime_value=protocol_version, + ) + ) + return mismatches diff --git a/src/mcp/server/experimental/__init__.py b/src/mcp/server/experimental/__init__.py new file mode 100644 index 0000000000..314bc21298 --- /dev/null +++ b/src/mcp/server/experimental/__init__.py @@ -0,0 +1,6 @@ +"""Experimental: tracks SEP-2127 (the Server Card extension). + +Nothing under `mcp.server.experimental` carries a stability guarantee. Any +symbol here may change or be removed in any release without a deprecation +cycle. +""" diff --git a/src/mcp/server/experimental/server_card.py b/src/mcp/server/experimental/server_card.py new file mode 100644 index 0000000000..ae9c0e4796 --- /dev/null +++ b/src/mcp/server/experimental/server_card.py @@ -0,0 +1,328 @@ +"""Serve a Server Card and an AI Catalog over HTTP (experimental, tracks SEP-2127). + +The route builders return plain Starlette routes, so they compose with +`streamable_http_app()` or any ASGI application. Every response goes through +`discovery_response`, the single enforcement point for the discovery spec's +HTTP requirements: media type, CORS, caching headers and conditional requests. + +Mount these routes outside any auth middleware. Discovery is unauthenticated +by design, and a card or catalog MUST NOT contain credentials, internal +network topology or private endpoints. +""" + +import hashlib +import re +from collections.abc import Awaitable, Callable, Sequence +from typing import Any, Protocol +from urllib.parse import urlsplit + +from mcp_types import Icon +from starlette.applications import Starlette +from starlette.middleware.cors import CORSMiddleware +from starlette.requests import Request +from starlette.responses import Response +from starlette.routing import Route, request_response +from starlette.types import ASGIApp + +from mcp.shared.experimental._base import SERVER_CARD_NAME_PATTERN, is_loopback_host +from mcp.shared.experimental.ai_catalog import ( + AI_CATALOG_MEDIA_TYPE, + AI_CATALOG_WELL_KNOWN_PATH, + AICatalog, + CatalogEntry, +) +from mcp.shared.experimental.server_card import ( + RESERVED_SERVER_CARD_SUFFIX, + SERVER_CARD_MEDIA_TYPE, + Remote, + Repository, + ServerCard, +) + +__all__ = [ + "build_server_card", + "create_server_card_routes", + "mount_server_card", + "create_ai_catalog_routes", + "mount_ai_catalog", + "mount_discovery", + "server_card_entry", + "catalog_identifier", + "discovery_response", +] + +_CORS_HEADERS = { + "Access-Control-Allow-Origin": "*", + "Access-Control-Allow-Methods": "GET", + "Access-Control-Allow-Headers": "Content-Type", +} + + +class _ServerIdentity(Protocol): + """The identity surface `build_server_card` reads. + + Both `MCPServer` (properties) and the lowlevel `Server` (plain attributes) + satisfy this structurally. + """ + + @property + def name(self) -> str: ... + @property + def title(self) -> str | None: ... + @property + def version(self) -> str | None: ... + @property + def description(self) -> str | None: ... + @property + def website_url(self) -> str | None: ... + @property + def icons(self) -> list[Icon] | None: ... + + +def build_server_card( + server: _ServerIdentity, + *, + name: str, + remotes: Sequence[Remote] | None = None, + repository: Repository | None = None, + description: str | None = None, + title: str | None = None, + version: str | None = None, + website_url: str | None = None, + icons: Sequence[Icon] | None = None, + meta: dict[str, Any] | None = None, +) -> ServerCard: + """Build a `ServerCard` from a server's identity fields. + + Title, description, version, website URL and icons come from the server + object. Explicit keyword arguments override the derived values, which + keeps the card consistent with what `serverInfo` reports at runtime. The + namespaced `name` and the public `remotes` URLs are never derivable, so + the caller supplies them. + + Raises: + pydantic.ValidationError: If the result violates a card constraint, + for example a server description over 100 characters or a version + that is unset and not overridden. + """ + resolved_icons = list(icons) if icons is not None else server.icons + fields: dict[str, Any] = { + "name": name, + "version": version if version is not None else server.version, + "description": description if description is not None else server.description, + "title": title if title is not None else server.title, + "website_url": website_url if website_url is not None else server.website_url, + "icons": resolved_icons, + "repository": repository, + "remotes": list(remotes) if remotes is not None else None, + "meta": meta, + } + return ServerCard.model_validate(fields) + + +def discovery_response( + request: Request, + body: bytes, + media_type: str, + *, + cache_control: str = "public, max-age=3600", +) -> Response: + """Answer one discovery request (card or catalog) per the extension spec. + + This is the compliance chokepoint the built-in routes use. It is public so + custom hosting (FastAPI apps, non-default paths, multi-card hosts) can + stay compliant. It emits the CORS headers the spec requires on every + response, a `Cache-Control` header (pass `cache_control=""` to omit it), a + strong SHA-256 `ETag`, a `304 Not Modified` for a matching + `If-None-Match`, and an empty 200 for `OPTIONS` preflight. + """ + headers = dict(_CORS_HEADERS) + if request.method == "OPTIONS": + return Response(status_code=200, headers=headers) + if cache_control: + headers["Cache-Control"] = cache_control + etag = '"' + hashlib.sha256(body).hexdigest() + '"' + headers["ETag"] = etag + if _if_none_match_matches(request.headers.get("if-none-match"), etag): + return Response(status_code=304, headers=headers) + return Response(content=body, media_type=media_type, headers=headers) + + +def _if_none_match_matches(header_value: str | None, etag: str) -> bool: + """Whether an `If-None-Match` header matches a strong ETag. + + Accepts the exact strong tag, its `W/`-prefixed weak form, `*`, or any + member of a comma-separated list. + """ + if header_value is None: + return False + for candidate in header_value.split(","): + stripped = candidate.strip() + if stripped == "*" or stripped.removeprefix("W/") == etag: + return True + return False + + +def _cors_endpoint(handler: Callable[[Request], Awaitable[Response]]) -> ASGIApp: + """Wrap a handler with `CORSMiddleware`, belt and braces over the explicit headers. + + The middleware short-circuits real browser preflights (OPTIONS with an + `Origin` and a requested method), answering with `GET` as the only allowed + method; Starlette also advertises the CORS-safelisted request headers + beside `Content-Type` there, a valid superset of the spec's example. A + bare OPTIONS still reaches `discovery_response`. + """ + return CORSMiddleware( + app=request_response(handler), + allow_origins=["*"], + allow_methods=["GET"], + allow_headers=["Content-Type"], + ) + + +def _document_routes(path: str, body: bytes, media_type: str) -> list[Route]: + """GET and OPTIONS routes serving one static discovery document.""" + + async def handler(request: Request) -> Response: + return discovery_response(request, body, media_type) + + return [Route(path, endpoint=_cors_endpoint(handler), methods=["GET", "OPTIONS"])] + + +def create_server_card_routes( + card: ServerCard, + *, + streamable_http_path: str = "/mcp", + path: str | None = None, +) -> list[Route]: + """Routes serving `card` at the spec-reserved path. + + The default path is the streamable HTTP path plus `/server-card`, for + example `/mcp/server-card`. Pass `path` to host the card at any other + unreserved URI instead. + """ + resolved_path = path if path is not None else streamable_http_path.rstrip("/") + RESERVED_SERVER_CARD_SUFFIX + body = card.model_dump_json(by_alias=True, exclude_none=True).encode("utf-8") + return _document_routes(resolved_path, body, SERVER_CARD_MEDIA_TYPE) + + +def create_ai_catalog_routes(catalog: AICatalog, *, path: str = AI_CATALOG_WELL_KNOWN_PATH) -> list[Route]: + """Routes serving `catalog`, by default at the well-known discovery path.""" + body = catalog.model_dump_json(by_alias=True, exclude_none=True).encode("utf-8") + return _document_routes(path, body, AI_CATALOG_MEDIA_TYPE) + + +def mount_server_card( + app: Starlette, + card: ServerCard, + *, + streamable_http_path: str = "/mcp", + path: str | None = None, +) -> None: + """Append the card routes to an already-built app. + + Works on the result of `streamable_http_app()`. Mount outside any auth + middleware. Discovery is unauthenticated by design. + """ + app.router.routes.extend(create_server_card_routes(card, streamable_http_path=streamable_http_path, path=path)) + + +def mount_ai_catalog(app: Starlette, catalog: AICatalog, *, path: str = AI_CATALOG_WELL_KNOWN_PATH) -> None: + """Append the catalog routes to an already-built app.""" + app.router.routes.extend(create_ai_catalog_routes(catalog, path=path)) + + +def _require_public_http_url(url: str) -> str: + """Validate an absolute http(s) URL and return its host. + + Raises: + ValueError: If `url` is not absolute http(s), or uses plain http to a + host that is not loopback. The spec requires HTTPS in production. + """ + parts = urlsplit(url) + if parts.scheme not in ("http", "https") or not parts.hostname: + raise ValueError(f"expected an absolute http(s) URL, got {url!r}") + if parts.scheme == "http" and not is_loopback_host(parts.hostname): + raise ValueError(f"plain http is only allowed for loopback hosts, got {url!r}") + return parts.hostname + + +def catalog_identifier(card_name: str, *, publisher: str) -> str: + """The recommended catalog URN for a card name. + + `catalog_identifier("com.example/weather", publisher="example.com")` + returns `"urn:air:example.com:mcp:weather"`. The `mcp` namespace segment + comes from the discovery spec. + + Raises: + ValueError: If `card_name` does not match the `namespace/name` pattern. + """ + if re.fullmatch(SERVER_CARD_NAME_PATTERN, card_name) is None: + raise ValueError(f"card name must be namespace/name, got {card_name!r}") + local_name = card_name.split("/", 1)[1] + return f"urn:air:{publisher}:mcp:{local_name}" + + +def server_card_entry( + card: ServerCard, + *, + publisher: str, + url: str | None = None, + identifier: str | None = None, +) -> CatalogEntry: + """A `CatalogEntry` advertising `card`. + + With `url` the entry points at the hosted card. With `url=None` the full + card is inlined as the entry's `data`. The identifier defaults to + `catalog_identifier(card.name, publisher=publisher)`. Display fields are + not duplicated from the card beyond `displayName`. Clients read title, + description and version from the card itself. + + Raises: + ValueError: If `url` is relative, or plain http to a non-loopback host. + """ + resolved_identifier = identifier if identifier is not None else catalog_identifier(card.name, publisher=publisher) + if url is None: + data = card.model_dump(by_alias=True, exclude_none=True, mode="json") + return CatalogEntry( + identifier=resolved_identifier, type=SERVER_CARD_MEDIA_TYPE, data=data, display_name=card.title + ) + _require_public_http_url(url) + return CatalogEntry(identifier=resolved_identifier, type=SERVER_CARD_MEDIA_TYPE, url=url, display_name=card.title) + + +def mount_discovery( + app: Starlette, + card: ServerCard, + *, + public_url: str, + streamable_http_path: str = "/mcp", +) -> None: + """Mount both discovery endpoints on a single-domain deployment. + + Serves the card at the reserved path under `streamable_http_path` and a + single-entry AI Catalog at the well-known path. The catalog entry's URL is + absolute (catalogs are fetched cross-domain) and the entry's URN publisher + is the host of `public_url`. + + `public_url` is the externally visible base URL at which `app`'s root is + served, typically just the origin (`https://mcp.example.com`). Include a + path only when a reverse proxy really serves the app under that prefix, + because the catalog entry's URL is `public_url` plus the card path while + the routes themselves are mounted at the app root; a `public_url` carrying + a path that the proxy does not strip advertises a card URL that 404s. + + Per the best-practices guidance, the catalog belongs on the domain users + associate with the service, which may not be the API app. In that case use + `server_card_entry` to build an entry for the catalog on your brand + domain instead. + + Raises: + ValueError: If `public_url` is not absolute http(s), or is plain http + to a non-loopback host. + """ + publisher = _require_public_http_url(public_url) + card_path = streamable_http_path.rstrip("/") + RESERVED_SERVER_CARD_SUFFIX + entry = server_card_entry(card, publisher=publisher, url=public_url.rstrip("/") + card_path) + mount_server_card(app, card, streamable_http_path=streamable_http_path) + mount_ai_catalog(app, AICatalog(spec_version="1.0", entries=[entry])) diff --git a/src/mcp/shared/experimental/__init__.py b/src/mcp/shared/experimental/__init__.py new file mode 100644 index 0000000000..b902dbd710 --- /dev/null +++ b/src/mcp/shared/experimental/__init__.py @@ -0,0 +1,6 @@ +"""Experimental: tracks SEP-2127 (the Server Card extension). + +Nothing under `mcp.shared.experimental` carries a stability guarantee. Any +symbol here may change or be removed in any release without a deprecation +cycle. +""" diff --git a/src/mcp/shared/experimental/_base.py b/src/mcp/shared/experimental/_base.py new file mode 100644 index 0000000000..5a12073f51 --- /dev/null +++ b/src/mcp/shared/experimental/_base.py @@ -0,0 +1,29 @@ +"""Internals shared by the experimental Server Card and AI Catalog modules.""" + +import ipaddress + +from pydantic import BaseModel, ConfigDict +from pydantic.alias_generators import to_camel + +SERVER_CARD_NAME_PATTERN = r"^[a-zA-Z0-9.-]+/[a-zA-Z0-9._-]+$" +"""Reverse-DNS namespace, one slash, then the server name (`com.example/weather`).""" + + +class CardModel(BaseModel): + """Base for all card and catalog models. + + The wire format is camelCase and the schema objects are open, so extra + (vendor) fields must survive a parse and re-serialize round trip. + """ + + model_config = ConfigDict(alias_generator=to_camel, populate_by_name=True, extra="allow") + + +def is_loopback_host(host: str) -> bool: + """Whether `host` is `localhost` or a loopback IP literal (`127.0.0.0/8`, `::1`).""" + if host == "localhost": + return True + try: + return ipaddress.ip_address(host.strip("[]")).is_loopback + except ValueError: + return False diff --git a/src/mcp/shared/experimental/ai_catalog.py b/src/mcp/shared/experimental/ai_catalog.py new file mode 100644 index 0000000000..9ddb20b15e --- /dev/null +++ b/src/mcp/shared/experimental/ai_catalog.py @@ -0,0 +1,67 @@ +"""AI Catalog models (experimental, minimal typed subset). + +An AI Catalog is a JSON document advertising AI artifacts, including Server +Cards. The format is owned by the AI Catalog working group +(https://github.com/Agent-Card/ai-catalog). Only the fields the SDK helpers +consume are typed here. Trust and publisher shapes pass through as plain +dictionaries for the host's consent UI. +""" + +from typing import Any, Final + +from pydantic import model_validator + +from mcp.shared.experimental._base import CardModel as _CardModel + +__all__ = [ + "AI_CATALOG_MEDIA_TYPE", + "AI_CATALOG_WELL_KNOWN_PATH", + "MAX_CATALOG_NESTING_DEPTH", + "CatalogEntry", + "AICatalog", +] + +AI_CATALOG_MEDIA_TYPE: Final = "application/ai-catalog+json" +"""Media type for AI Catalog documents.""" + +AI_CATALOG_WELL_KNOWN_PATH: Final = "/.well-known/ai-catalog.json" +"""Well-known path for domain-level catalog discovery.""" + +MAX_CATALOG_NESTING_DEPTH: Final = 4 +"""The AI Catalog spec's cap on nested catalog depth.""" + + +class CatalogEntry(_CardModel): + """One advertised artifact: a Server Card, a nested catalog, or anything else. + + Exactly one of `url` and `data` must be set. Violations raise + `pydantic.ValidationError`. + """ + + identifier: str + type: str + url: str | None = None + data: dict[str, Any] | None = None + display_name: str | None = None + description: str | None = None + tags: list[str] | None = None + version: str | None = None + updated_at: str | None = None + publisher: dict[str, Any] | None = None + trust_manifest: dict[str, Any] | None = None + metadata: dict[str, Any] | None = None + + @model_validator(mode="after") + def _exactly_one_of_url_or_data(self) -> "CatalogEntry": + if (self.url is None) == (self.data is None): + raise ValueError("a catalog entry carries exactly one of url or data") + return self + + +class AICatalog(_CardModel): + """An AI Catalog document (`application/ai-catalog+json`).""" + + spec_version: str + entries: list[CatalogEntry] + host: dict[str, Any] | None = None + metadata: dict[str, Any] | None = None diff --git a/src/mcp/shared/experimental/server_card.py b/src/mcp/shared/experimental/server_card.py new file mode 100644 index 0000000000..63a93e0c22 --- /dev/null +++ b/src/mcp/shared/experimental/server_card.py @@ -0,0 +1,262 @@ +"""Server Card models and pure helpers (experimental, tracks SEP-2127). + +A Server Card is a static JSON document that describes a single remote MCP +server well enough for a client to discover and connect to it before any +protocol exchange. See +https://github.com/modelcontextprotocol/experimental-ext-server-card for the +authoritative schema. + +Card contents are unverified and advisory. Clients MUST NOT treat them as +authoritative for security or access-control decisions, and SHOULD prefer +runtime values (`serverInfo`) whenever the two disagree. +""" + +import re +from collections.abc import Mapping +from dataclasses import dataclass +from typing import Any, Final, Literal + +from mcp_types import Icon +from pydantic import Field, field_validator + +from mcp.shared.experimental._base import SERVER_CARD_NAME_PATTERN +from mcp.shared.experimental._base import CardModel as _CardModel + +__all__ = [ + "SERVER_CARD_SCHEMA_URL", + "SERVER_CARD_MEDIA_TYPE", + "RESERVED_SERVER_CARD_SUFFIX", + "Input", + "KeyValueInput", + "Repository", + "Remote", + "ServerCard", + "ResolvedRemote", + "resolve_remote", +] + +SERVER_CARD_SCHEMA_URL: Final = "https://static.modelcontextprotocol.io/schemas/v1/server-card.schema.json" +"""The only `$schema` value a v1 Server Card may carry.""" + +SERVER_CARD_MEDIA_TYPE: Final = "application/mcp-server-card+json" +"""Media type for Server Card documents.""" + +RESERVED_SERVER_CARD_SUFFIX: Final = "/server-card" +"""Path suffix the spec reserves under the streamable HTTP URL.""" + +_REMOTE_URL_PATTERN = r"^(https?://[^\s]+|\{[a-zA-Z_][a-zA-Z0-9_]*\}[^\s]*)$" +_TEMPLATE_VARIABLE = re.compile(r"\{([a-zA-Z_][a-zA-Z0-9_]*)\}") +_VERSION_RANGE_PREFIXES = ("^", "~", ">=", "<=", ">", "<") +_VERSION_WILDCARD_SEGMENTS = frozenset({"x", "X", "*"}) + + +class Input(_CardModel): + """A user-suppliable value referenced by a `Remote` URL or header template.""" + + description: str | None = None + is_required: bool | None = None + is_secret: bool | None = None + format: Literal["string", "number", "boolean", "filepath"] | None = None + default: str | None = None + placeholder: str | None = None + value: str | None = None + choices: list[str] | None = None + + +class KeyValueInput(Input): + """A named input, used for HTTP headers on a `Remote`.""" + + name: str + variables: dict[str, Input] | None = None + + +class Repository(_CardModel): + """Source repository metadata for the server implementation.""" + + url: str + source: str + subfolder: str | None = None + id: str | None = None + + +class Remote(_CardModel): + """Connection metadata for one remote transport endpoint.""" + + type: Literal["streamable-http", "sse"] + url: str = Field(pattern=_REMOTE_URL_PATTERN) + headers: list[KeyValueInput] | None = None + variables: dict[str, Input] | None = None + supported_protocol_versions: list[str] | None = None + + @property + def required_variables(self) -> frozenset[str]: + """The input names a host must prompt the user for. + + Covers the URL variables, each header's nested variables, and required + headers themselves (keyed by header name, exactly as `resolve_remote` + accepts values for them), keeping the ones declared `isRequired` with + no `default` and no pre-set `value`. + """ + names: set[str] = set() + header_variables = [header.variables for header in self.headers or []] + for variables in [self.variables, *header_variables]: + for name, spec in (variables or {}).items(): + if spec.is_required and spec.default is None and spec.value is None: + names.add(name) + for header in self.headers or []: + if header.is_required and header.default is None and header.value is None: + names.add(header.name) + return frozenset(names) + + +class ServerCard(_CardModel): + """A Server Card document (`application/mcp-server-card+json`). + + Malformed documents raise `pydantic.ValidationError` at construction or + `model_validate*` time. + """ + + schema_: str = Field(default=SERVER_CARD_SCHEMA_URL, alias="$schema") + name: str = Field(min_length=3, max_length=200, pattern=SERVER_CARD_NAME_PATTERN) + version: str = Field(max_length=255) + description: str = Field(min_length=1, max_length=100) + title: str | None = Field(default=None, min_length=1, max_length=100) + website_url: str | None = None + repository: Repository | None = None + icons: list[Icon] | None = None + remotes: list[Remote] | None = None + meta: dict[str, Any] | None = Field(default=None, alias="_meta") + + @field_validator("schema_") + @classmethod + def _schema_url_is_the_v1_url(cls, value: str) -> str: + if value != SERVER_CARD_SCHEMA_URL: + raise ValueError(f"$schema must be exactly {SERVER_CARD_SCHEMA_URL!r}") + return value + + @field_validator("version") + @classmethod + def _version_is_not_a_range(cls, value: str) -> str: + # Range syntax is rejected at the prose level of the spec. It is not + # expressible in the published JSON Schema. + if value.startswith(_VERSION_RANGE_PREFIXES) or "||" in value: + raise ValueError("version must be a single version, never a range") + if any(segment in _VERSION_WILDCARD_SEGMENTS for segment in value.split(".")): + raise ValueError("version must not contain wildcard segments") + return value + + def endpoint_urls(self) -> frozenset[str]: + """The raw `remotes[].url` values, templates unresolved. + + This is the dedup key for discovered servers: hosts MUST de-duplicate + on endpoints, never on the self-asserted name or catalog identifier. + """ + return frozenset(remote.url for remote in self.remotes or []) + + +@dataclass(frozen=True, slots=True) +class ResolvedRemote: + """A `Remote` with every template variable substituted, ready to connect.""" + + type: Literal["streamable-http", "sse"] + url: str + headers: dict[str, str] + + +def _substitute(template: str, mapping: Mapping[str, str]) -> tuple[str, set[str]]: + """Replace `{name}` placeholders from `mapping`, collecting unresolved names.""" + unresolved: set[str] = set() + + def replace(match: re.Match[str]) -> str: + name = match.group(1) + if name in mapping: + return mapping[name] + unresolved.add(name) + return match.group(0) + + return _TEMPLATE_VARIABLE.sub(replace, template), unresolved + + +def _resolve_input(name: str, spec: Input, values: Mapping[str, str] | None) -> str | None: + """The effective value for one input, or None when nothing supplies it. + + A pre-set `value` is not end-user configurable, so it wins. Caller values + override declared defaults. + """ + if spec.value is not None: + return spec.value + if values is not None and name in values: + return values[name] + return spec.default + + +def _resolve_variables( + declared: Mapping[str, Input] | None, + values: Mapping[str, str] | None, + missing: set[str], + bad_choices: list[str], +) -> dict[str, str]: + """Build the substitution mapping for one template scope. + + Caller values apply to undeclared names too. Missing required names and + values outside a declared `choices` list are collected, never raised here. + """ + mapping: dict[str, str] = dict(values or {}) + for name, spec in (declared or {}).items(): + resolved = _resolve_input(name, spec, values) + if resolved is None: + if spec.is_required: + missing.add(name) + continue + if spec.choices is not None and resolved not in spec.choices: + bad_choices.append(f"{name}={resolved!r} (choices: {spec.choices})") + continue + mapping[name] = resolved + return mapping + + +def resolve_remote(remote: Remote, values: Mapping[str, str] | None = None) -> ResolvedRemote: + """Substitute every `{curly_brace}` variable in `remote`, purely in memory. + + Declared `default`s and pre-set `value`s apply automatically. Caller + `values` override defaults and also supply header values by header name. + No prompting, no persistence and no `isSecret` handling happens here. + Those belong to the host application. + + Raises: + ValueError: Naming every missing `isRequired` variable, any value + outside a declared `choices` list, any placeholder left + unresolved, or a resolved URL that is not http(s). + """ + missing: set[str] = set() + bad_choices: list[str] = [] + unresolved: set[str] = set() + + url_mapping = _resolve_variables(remote.variables, values, missing, bad_choices) + url, url_unresolved = _substitute(remote.url, url_mapping) + unresolved |= url_unresolved + + headers: dict[str, str] = {} + for header in remote.headers or []: + template = _resolve_input(header.name, header, values) + if template is None: + if header.is_required: + missing.add(header.name) + continue + if header.choices is not None and template not in header.choices: + bad_choices.append(f"{header.name}={template!r} (choices: {header.choices})") + continue + header_mapping = _resolve_variables(header.variables, values, missing, bad_choices) + resolved_value, value_unresolved = _substitute(template, header_mapping) + unresolved |= value_unresolved + headers[header.name] = resolved_value + + if missing: + raise ValueError(f"missing required variables: {', '.join(sorted(missing))}") + if bad_choices: + raise ValueError(f"values outside declared choices: {'; '.join(bad_choices)}") + if unresolved: + raise ValueError(f"unresolved template variables: {', '.join(sorted(unresolved))}") + if not url.startswith(("http://", "https://")): + raise ValueError(f"resolved remote URL must be http(s), got {url!r}") + return ResolvedRemote(type=remote.type, url=url, headers=headers) diff --git a/tests/client/experimental/__init__.py b/tests/client/experimental/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/tests/client/experimental/test_server_card.py b/tests/client/experimental/test_server_card.py new file mode 100644 index 0000000000..7191b355c9 --- /dev/null +++ b/tests/client/experimental/test_server_card.py @@ -0,0 +1,972 @@ +"""`mcp.client.experimental.server_card`: hardened fetch, discovery, revalidation, reconcile.""" + +import ipaddress +import json +import socket +from collections.abc import Callable, Coroutine +from pathlib import Path +from typing import Any + +import anyio +import httpx2 +import pytest +from inline_snapshot import snapshot +from mcp_types import Implementation +from pydantic import ValidationError +from starlette.applications import Starlette + +from mcp import Client +from mcp.client.experimental import _discovery_http +from mcp.client.experimental.server_card import ( + CardListing, + DiscoveryError, + DiscoveryErrorReason, + DiscoveryPolicy, + DiscoveryResult, + create_ai_catalog_request, + create_server_card_request, + discover_server_cards, + fetch_ai_catalog, + fetch_server_card, + load_server_card, + parse_ai_catalog_response, + parse_server_card_response, + reconcile_server_card, + server_card_url, + well_known_ai_catalog_url, +) +from mcp.server import MCPServer +from mcp.server.experimental.server_card import build_server_card, mount_discovery +from mcp.shared.experimental.ai_catalog import AICatalog, CatalogEntry +from mcp.shared.experimental.server_card import Remote, ServerCard + +pytestmark = pytest.mark.anyio + +CARD_MEDIA_TYPE = "application/mcp-server-card+json" +CATALOG_MEDIA_TYPE = "application/ai-catalog+json" + + +def _card(name: str = "com.example/weather") -> ServerCard: + return ServerCard( + name=name, + version="1.4.0", + description="Hourly forecasts.", + remotes=[Remote(type="streamable-http", url="https://mcp.example.com/mcp")], + ) + + +def _card_bytes(name: str = "com.example/weather") -> bytes: + return _card(name).model_dump_json(by_alias=True, exclude_none=True).encode() + + +def _catalog_bytes(entries: list[dict[str, Any]]) -> bytes: + return json.dumps({"specVersion": "1.0", "entries": entries}).encode() + + +def _mock_client( + handler: Callable[[httpx2.Request], httpx2.Response] + | Callable[[httpx2.Request], Coroutine[None, None, httpx2.Response]], +) -> httpx2.AsyncClient: + return httpx2.AsyncClient(transport=httpx2.MockTransport(handler)) + + +def _refuse_all(request: httpx2.Request) -> httpx2.Response: + raise NotImplementedError + + +@pytest.fixture +def public_dns(monkeypatch: pytest.MonkeyPatch) -> None: + """Pin DNS so default-policy tests stay deterministic and offline. + + The address guard resolves every hostname before the request. Stubbing the + private resolver to a public address is the only way to exercise the + default (hardened) policy against an in-memory transport without real DNS. + """ + + async def resolve(host: str) -> list[ipaddress.IPv4Address | ipaddress.IPv6Address]: + try: + return [ipaddress.ip_address(host)] # IP literals keep their real meaning + except ValueError: + return [ipaddress.ip_address("93.184.216.34")] + + monkeypatch.setattr(_discovery_http, "_host_addresses", resolve) + + +# -- fetch happy paths -------------------------------------------------------------------- + + +async def test_fetch_server_card_sends_the_accept_header_and_parses_the_card(public_dns: None) -> None: + """Spec-mandated: the client sends `Accept: application/mcp-server-card+json` and + parses the canonical media type.""" + seen: list[httpx2.Request] = [] + + def handler(request: httpx2.Request) -> httpx2.Response: + seen.append(request) + return httpx2.Response(200, content=_card_bytes(), headers={"Content-Type": CARD_MEDIA_TYPE}) + + async with _mock_client(handler) as client: + card = await fetch_server_card("https://example.com/mcp/server-card", http_client=client) + assert card == _card() + assert seen[0].headers["accept"] == "application/mcp-server-card+json, application/json;q=0.5" + + +async def test_fetch_accepts_media_type_parameters_and_plain_json(public_dns: None) -> None: + """SDK-defined lenience: `; charset=` parameters are ignored and plain + `application/json` passes, because static hosts and CDNs commonly serve it.""" + + def handler(request: httpx2.Request) -> httpx2.Response: + if request.url.path.endswith("server-card"): + return httpx2.Response( + 200, content=_card_bytes(), headers={"Content-Type": f"{CARD_MEDIA_TYPE}; charset=utf-8"} + ) + return httpx2.Response(200, content=_catalog_bytes([]), headers={"Content-Type": "application/json"}) + + async with _mock_client(handler) as client: + card = await fetch_server_card("https://example.com/mcp/server-card", http_client=client) + catalog = await fetch_ai_catalog("https://example.com/.well-known/ai-catalog.json", http_client=client) + assert card == _card() + assert catalog == AICatalog(spec_version="1.0", entries=[]) + + +async def test_fetch_rejects_a_malformed_card_document(public_dns: None) -> None: + """SDK-defined: document shape failures stay `pydantic.ValidationError`, distinct + from transport and policy failures.""" + + def handler(request: httpx2.Request) -> httpx2.Response: + return httpx2.Response(200, content=b'{"name": "no-namespace"}', headers={"Content-Type": CARD_MEDIA_TYPE}) + + async with _mock_client(handler) as client: + with pytest.raises(ValidationError): + await fetch_server_card("https://example.com/mcp/server-card", http_client=client) + + +# -- policy failures ------------------------------------------------------------------------ + + +async def test_non_2xx_status_raises_with_reason_status(public_dns: None) -> None: + """SDK-defined: a non-2xx answer becomes a `DiscoveryError` naming the URL, with the + `httpx2.HTTPStatusError` chained as the cause.""" + + def handler(request: httpx2.Request) -> httpx2.Response: + return httpx2.Response(404) + + async with _mock_client(handler) as client: + with pytest.raises(DiscoveryError) as exc_info: + await fetch_server_card("https://example.com/mcp/server-card", http_client=client) + assert exc_info.value.reason == "status" + assert exc_info.value.url == "https://example.com/mcp/server-card" + assert isinstance(exc_info.value.__cause__, httpx2.HTTPStatusError) + + +async def test_wrong_media_type_raises_with_reason_media_type(public_dns: None) -> None: + """Spec-mandated: hosts must respect the card media type, so `text/html` (a typical + captive error page) is rejected.""" + + def handler(request: httpx2.Request) -> httpx2.Response: + return httpx2.Response(200, content=b"", headers={"Content-Type": "text/html"}) + + async with _mock_client(handler) as client: + with pytest.raises(DiscoveryError) as exc_info: + await fetch_server_card("https://example.com/mcp/server-card", http_client=client) + assert exc_info.value.reason == "media_type" + + +async def test_a_body_over_the_size_cap_raises_mid_stream(public_dns: None) -> None: + """SDK-defined (best practices): response size is capped and enforced while + streaming, so an oversized document fails without being buffered whole.""" + + def handler(request: httpx2.Request) -> httpx2.Response: + return httpx2.Response(200, content=b"x" * 64, headers={"Content-Type": CARD_MEDIA_TYPE}) + + async with _mock_client(handler) as client: + with pytest.raises(DiscoveryError) as exc_info: + await fetch_server_card( + "https://example.com/mcp/server-card", + http_client=client, + policy=DiscoveryPolicy(max_response_bytes=16), + ) + assert exc_info.value.reason == "response_too_large" + + +async def test_redirects_are_followed_within_the_cap(public_dns: None) -> None: + """SDK-defined (best practices): redirects are walked manually, resolving relative + `Location` values, and the final document parses normally.""" + + def handler(request: httpx2.Request) -> httpx2.Response: + if request.url.path == "/old": + return httpx2.Response(301, headers={"Location": "/newer"}) + if request.url.path == "/newer": + return httpx2.Response(302, headers={"Location": "https://example.com/final"}) + assert request.url.path == "/final" + return httpx2.Response(200, content=_card_bytes(), headers={"Content-Type": CARD_MEDIA_TYPE}) + + async with _mock_client(handler) as client: + card = await fetch_server_card("https://example.com/old", http_client=client) + assert card == _card() + + +async def test_redirects_past_the_cap_raise(public_dns: None) -> None: + """SDK-defined (best practices): the redirect budget is bounded.""" + + def handler(request: httpx2.Request) -> httpx2.Response: + return httpx2.Response(302, headers={"Location": "https://example.com/loop"}) + + async with _mock_client(handler) as client: + with pytest.raises(DiscoveryError) as exc_info: + await fetch_server_card( + "https://example.com/loop", http_client=client, policy=DiscoveryPolicy(max_redirects=1) + ) + assert exc_info.value.reason == "too_many_redirects" + + +async def test_a_redirect_without_location_raises(public_dns: None) -> None: + """SDK-defined: a 3xx with no `Location` header cannot be followed.""" + + def handler(request: httpx2.Request) -> httpx2.Response: + return httpx2.Response(302) + + async with _mock_client(handler) as client: + with pytest.raises(DiscoveryError) as exc_info: + await fetch_server_card("https://example.com/mcp/server-card", http_client=client) + assert exc_info.value.reason == "status" + + +async def test_each_redirect_hop_is_revalidated(public_dns: None) -> None: + """SDK-defined (best practices): a public origin redirecting into a private range is + caught on the hop, not trusted because the first URL was fine.""" + + def handler(request: httpx2.Request) -> httpx2.Response: + return httpx2.Response(302, headers={"Location": "https://10.0.0.1/internal"}) + + async with _mock_client(handler) as client: + with pytest.raises(DiscoveryError) as exc_info: + await fetch_server_card("https://example.com/mcp/server-card", http_client=client) + assert exc_info.value.reason == "blocked_address" + assert exc_info.value.url == "https://10.0.0.1/internal" + + +@pytest.mark.parametrize( + "url", + [ + "https://10.0.0.1/mcp/server-card", + "https://172.16.0.5/mcp/server-card", + "https://192.168.1.1/mcp/server-card", + "https://169.254.169.254/latest/meta-data", + "https://100.64.0.1/mcp/server-card", + "https://224.0.0.1/mcp/server-card", + "https://240.0.0.1/mcp/server-card", + "https://0.0.0.0/mcp/server-card", + "https://127.0.0.1/mcp/server-card", + "https://[::1]/mcp/server-card", + "https://[fc00::1]/mcp/server-card", + "https://[fe80::1]/mcp/server-card", + "https://[::ffff:10.0.0.1]/mcp/server-card", + "https://[::ffff:100.64.0.1]/mcp/server-card", + ], +) +async def test_ip_literal_targets_off_the_public_internet_are_blocked(url: str) -> None: + """SDK-defined (best practices): loopback, link-local (including the cloud metadata + endpoint), private, CGNAT, multicast, reserved, unspecified and IPv4-mapped IPv6 + literals never get a request. The handler proves it by refusing to answer.""" + async with _mock_client(_refuse_all) as client: + with pytest.raises(DiscoveryError) as exc_info: + await fetch_server_card(url, http_client=client) + assert exc_info.value.reason == "blocked_address" + + +async def test_a_hostname_resolving_to_a_private_address_is_blocked(monkeypatch: pytest.MonkeyPatch) -> None: + """SDK-defined (best practices): the guard re-checks after DNS resolution, so a + public-looking hostname pointing into a private range is rejected.""" + + async def resolve(host: str) -> list[ipaddress.IPv4Address | ipaddress.IPv6Address]: + return [ipaddress.ip_address("10.9.8.7")] + + monkeypatch.setattr(_discovery_http, "_host_addresses", resolve) + async with _mock_client(_refuse_all) as client: + with pytest.raises(DiscoveryError) as exc_info: + await fetch_server_card("https://internal.example.com/mcp/server-card", http_client=client) + assert exc_info.value.reason == "blocked_address" + + +async def test_localhost_is_blocked_under_the_default_policy() -> None: + """SDK-defined: `https://localhost` resolves (via the real resolver) to loopback and + is blocked. Local development opts in with `allow_private_addresses=True`.""" + async with _mock_client(_refuse_all) as client: + with pytest.raises(DiscoveryError) as exc_info: + await fetch_server_card("https://localhost/mcp/server-card", http_client=client) + assert exc_info.value.reason == "blocked_address" + + +@pytest.mark.parametrize("url", ["http://example.com/mcp/server-card", "http://127.0.0.1/mcp/server-card"]) +async def test_plain_http_raises_insecure_transport_under_the_default_policy(url: str) -> None: + """Spec-mandated: HTTPS is a MUST in production. Under the hardened policy even a + loopback http target is refused up front (its address would be blocked anyway); + local development opts in with `allow_private_addresses=True`.""" + async with _mock_client(_refuse_all) as client: + with pytest.raises(DiscoveryError) as exc_info: + await fetch_server_card(url, http_client=client) + assert exc_info.value.reason == "insecure_transport" + + +async def test_a_url_carrying_userinfo_is_rejected_before_any_request() -> None: + """SDK-defined hardening: `https://github.com@evil.example/...` reads as a trusted + brand but names `evil.example`. Discovery never sends credentials, so userinfo URLs + never get a request. The handler proves it by refusing to answer.""" + async with _mock_client(_refuse_all) as client: + with pytest.raises(DiscoveryError) as exc_info: + await fetch_server_card("https://github.com@evil.example/mcp/server-card", http_client=client) + assert exc_info.value.reason == "insecure_transport" + + +@pytest.mark.parametrize("url", ["ftp://example.com/card", "/mcp/server-card"]) +async def test_non_http_and_relative_urls_raise_insecure_transport(url: str) -> None: + """SDK-defined: discovery only ever speaks absolute http(s).""" + async with _mock_client(_refuse_all) as client: + with pytest.raises(DiscoveryError) as exc_info: + await fetch_server_card(url, http_client=client) + assert exc_info.value.reason == "insecure_transport" + + +async def test_allow_private_addresses_permits_local_development() -> None: + """SDK-defined: the local-dev policy admits plain http to localhost and private + targets, skipping the scheme and address guards.""" + policy = DiscoveryPolicy(allow_private_addresses=True) + + def handler(request: httpx2.Request) -> httpx2.Response: + return httpx2.Response(200, content=_card_bytes(), headers={"Content-Type": CARD_MEDIA_TYPE}) + + async with _mock_client(handler) as client: + via_localhost = await fetch_server_card( + "http://localhost:8000/mcp/server-card", http_client=client, policy=policy + ) + via_private_ip = await fetch_server_card("https://10.0.0.1/mcp/server-card", http_client=client, policy=policy) + assert via_localhost == _card() + assert via_private_ip == _card() + + +async def test_the_default_client_path_applies_the_same_guards() -> None: + """SDK-defined: with no `http_client`, a fresh credential-free client is created and + the URL is still admitted first. A blocked target fails before any request.""" + with pytest.raises(DiscoveryError) as exc_info: + await fetch_server_card("https://10.0.0.1/mcp/server-card") + assert exc_info.value.reason == "blocked_address" + + +# -- discover_server_cards ------------------------------------------------------------------- + + +async def test_discover_probes_the_well_known_path_of_the_origin(public_dns: None) -> None: + """Spec-mandated: any user-entered URL probes exactly its origin's + `/.well-known/ai-catalog.json`, then follows card entries by URL.""" + seen_paths: list[str] = [] + + def handler(request: httpx2.Request) -> httpx2.Response: + seen_paths.append(request.url.path) + if request.url.path == "/.well-known/ai-catalog.json": + entry = { + "identifier": "urn:air:example.com:mcp:weather", + "type": CARD_MEDIA_TYPE, + "url": "https://example.com/mcp/server-card", + } + return httpx2.Response(200, content=_catalog_bytes([entry]), headers={"Content-Type": CATALOG_MEDIA_TYPE}) + return httpx2.Response(200, content=_card_bytes(), headers={"Content-Type": CARD_MEDIA_TYPE}) + + async with _mock_client(handler) as client: + result = await discover_server_cards("https://example.com/docs/page?q=1", http_client=client) + assert seen_paths[0] == "/.well-known/ai-catalog.json" + assert result.failures == [] + (listing,) = result.listings + assert listing.card == _card() + assert listing.catalog_url == "https://example.com/.well-known/ai-catalog.json" + assert listing.card_url == "https://example.com/mcp/server-card" + assert (listing.listing_domain, listing.hosting_domain) == ("example.com", "example.com") + + +async def test_discover_reads_inline_data_entries_without_a_fetch(public_dns: None) -> None: + """Spec-mandated: an entry may inline the card as `data`. No card URL exists, so the + hosting domain is None.""" + inline = { + "identifier": "urn:air:example.com:mcp:weather", + "type": CARD_MEDIA_TYPE, + "data": json.loads(_card_bytes()), + } + + def handler(request: httpx2.Request) -> httpx2.Response: + return httpx2.Response(200, content=_catalog_bytes([inline]), headers={"Content-Type": CATALOG_MEDIA_TYPE}) + + async with _mock_client(handler) as client: + result = await discover_server_cards("https://example.com", http_client=client) + (listing,) = result.listings + assert listing.card == _card() + assert listing.card_url is None + assert listing.hosting_domain is None + + +async def test_discover_follows_nested_catalogs_by_url_and_inline(public_dns: None) -> None: + """Spec-mandated: catalog entries may be catalogs themselves. Listings keep the + nested catalog's URL as their listing source; an inline nested catalog keeps the + document it was embedded in.""" + inline_card = { + "identifier": "urn:air:example.com:mcp:inline", + "type": CARD_MEDIA_TYPE, + "data": json.loads(_card_bytes("com.example/inline")), + } + nested_by_url = { + "identifier": "urn:air:example.com:catalog:more", + "type": CATALOG_MEDIA_TYPE, + "url": "https://example.com/more-catalog.json", + } + nested_inline = { + "identifier": "urn:air:example.com:catalog:embedded", + "type": CATALOG_MEDIA_TYPE, + "data": json.loads(_catalog_bytes([inline_card])), + } + + def handler(request: httpx2.Request) -> httpx2.Response: + if request.url.path == "/.well-known/ai-catalog.json": + body = _catalog_bytes([nested_by_url, nested_inline]) + else: + assert request.url.path == "/more-catalog.json" + card_entry = { + "identifier": "urn:air:example.com:mcp:weather", + "type": CARD_MEDIA_TYPE, + "data": json.loads(_card_bytes()), + } + body = _catalog_bytes([card_entry]) + return httpx2.Response(200, content=body, headers={"Content-Type": CATALOG_MEDIA_TYPE}) + + async with _mock_client(handler) as client: + result = await discover_server_cards("https://example.com", http_client=client) + assert result.failures == [] + assert [listing.catalog_url for listing in result.listings] == [ + "https://example.com/more-catalog.json", + "https://example.com/.well-known/ai-catalog.json", + ] + + +async def test_discover_records_a_depth_cap_failure_instead_of_recursing(public_dns: None) -> None: + """SDK-defined (AI Catalog spec cap): nesting past `max_catalog_depth` becomes one + failure with reason `catalog_depth`, never an unbounded walk.""" + nested = { + "identifier": "urn:air:example.com:catalog:deep", + "type": CATALOG_MEDIA_TYPE, + "url": "https://example.com/deep.json", + } + + def handler(request: httpx2.Request) -> httpx2.Response: + return httpx2.Response(200, content=_catalog_bytes([nested]), headers={"Content-Type": CATALOG_MEDIA_TYPE}) + + async with _mock_client(handler) as client: + result = await discover_server_cards( + "https://example.com", http_client=client, policy=DiscoveryPolicy(max_catalog_depth=1) + ) + assert result.listings == [] + (failure,) = result.failures + assert isinstance(failure.error, DiscoveryError) + assert failure.error.reason == "catalog_depth" + assert failure.entry_identifier == "urn:air:example.com:catalog:deep" + + +async def test_discover_ignores_entries_of_other_artifact_types(public_dns: None) -> None: + """Spec-mandated: catalogs legitimately advertise other artifacts. They are neither + listings nor failures.""" + other = { + "identifier": "urn:air:example.com:agent:helper", + "type": "application/agent-card+json", + "url": "https://example.com/agent.json", + } + + def handler(request: httpx2.Request) -> httpx2.Response: + return httpx2.Response(200, content=_catalog_bytes([other]), headers={"Content-Type": CATALOG_MEDIA_TYPE}) + + async with _mock_client(handler) as client: + result = await discover_server_cards("https://example.com", http_client=client) + assert (result.listings, result.failures) == ([], []) + + +async def test_discover_collects_per_entry_failures_and_keeps_good_listings(public_dns: None) -> None: + """SDK-defined (best practices): one hostile or broken entry never kills the probe. + Failures are collected with their entry identifiers while good entries still list.""" + entries = [ + {"identifier": "urn:air:example.com:mcp:good", "type": CARD_MEDIA_TYPE, "data": json.loads(_card_bytes())}, + {"identifier": "urn:air:example.com:mcp:gone", "type": CARD_MEDIA_TYPE, "url": "https://example.com/gone"}, + {"identifier": "urn:air:example.com:mcp:bad", "type": CARD_MEDIA_TYPE, "data": {"name": "not-a-card"}}, + ] + + def handler(request: httpx2.Request) -> httpx2.Response: + if request.url.path == "/.well-known/ai-catalog.json": + return httpx2.Response(200, content=_catalog_bytes(entries), headers={"Content-Type": CATALOG_MEDIA_TYPE}) + return httpx2.Response(404) + + async with _mock_client(handler) as client: + result = await discover_server_cards("https://example.com", http_client=client) + assert [listing.entry.identifier for listing in result.listings] == ["urn:air:example.com:mcp:good"] + assert [failure.entry_identifier for failure in result.failures] == [ + "urn:air:example.com:mcp:gone", + "urn:air:example.com:mcp:bad", + ] + assert isinstance(result.failures[0].error, DiscoveryError) + assert isinstance(result.failures[1].error, ValidationError) + + +async def test_discover_caps_the_entry_count_and_records_the_excess(public_dns: None) -> None: + """SDK-defined (best practices): entry count is capped. The excess is recorded as + one failure on the catalog itself, not silently dropped.""" + entries = [ + {"identifier": f"urn:air:example.com:mcp:c{i}", "type": CARD_MEDIA_TYPE, "data": json.loads(_card_bytes())} + for i in range(3) + ] + + def handler(request: httpx2.Request) -> httpx2.Response: + return httpx2.Response(200, content=_catalog_bytes(entries), headers={"Content-Type": CATALOG_MEDIA_TYPE}) + + async with _mock_client(handler) as client: + result = await discover_server_cards( + "https://example.com", http_client=client, policy=DiscoveryPolicy(max_catalog_entries=2) + ) + assert [listing.entry.identifier for listing in result.listings] == [ + "urn:air:example.com:mcp:c0", + "urn:air:example.com:mcp:c1", + ] + (failure,) = result.failures + assert failure.entry_identifier is None + assert isinstance(failure.error, DiscoveryError) + assert failure.error.reason == "invalid_entry" + + +async def test_discover_keeps_other_listings_when_an_entry_host_does_not_resolve( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """SDK-defined (best practices): the address guard resolves DNS itself, so an + unresolvable entry host raises `socket.gaierror`, the most likely hostile or broken + entry shape. It becomes a failure while the good entries still list.""" + + async def resolve(host: str) -> list[ipaddress.IPv4Address | ipaddress.IPv6Address]: + if host == "gone.invalid": + raise socket.gaierror(socket.EAI_NONAME, "Name or service not known") + return [ipaddress.ip_address("93.184.216.34")] + + monkeypatch.setattr(_discovery_http, "_host_addresses", resolve) + entries = [ + {"identifier": "urn:air:example.com:mcp:good", "type": CARD_MEDIA_TYPE, "data": json.loads(_card_bytes())}, + {"identifier": "urn:air:example.com:mcp:gone", "type": CARD_MEDIA_TYPE, "url": "https://gone.invalid/card"}, + ] + + def handler(request: httpx2.Request) -> httpx2.Response: + return httpx2.Response(200, content=_catalog_bytes(entries), headers={"Content-Type": CATALOG_MEDIA_TYPE}) + + async with _mock_client(handler) as client: + result = await discover_server_cards("https://example.com", http_client=client) + assert [listing.entry.identifier for listing in result.listings] == ["urn:air:example.com:mcp:good"] + (failure,) = result.failures + assert failure.entry_identifier == "urn:air:example.com:mcp:gone" + assert isinstance(failure.error, socket.gaierror) + + +async def test_discover_keeps_other_listings_when_an_entry_times_out(public_dns: None) -> None: + """SDK-defined (best practices): a tar-pit entry hits the per-fetch deadline + (`TimeoutError`) and becomes a failure while the good entries still list. The sleep + is the thing under test here (the deadline is a time-based feature), and the probe + runs in a child task because the deadline's cancellation stops coverage tracing in + every frame still suspended on the awaited call.""" + entries = [ + {"identifier": "urn:air:example.com:mcp:good", "type": CARD_MEDIA_TYPE, "data": json.loads(_card_bytes())}, + {"identifier": "urn:air:example.com:mcp:slow", "type": CARD_MEDIA_TYPE, "url": "https://example.com/tar-pit"}, + ] + + async def handler(request: httpx2.Request) -> httpx2.Response: + if request.url.path == "/.well-known/ai-catalog.json": + return httpx2.Response(200, content=_catalog_bytes(entries), headers={"Content-Type": CATALOG_MEDIA_TYPE}) + await anyio.sleep(3600) # cancelled by the per-fetch deadline + raise NotImplementedError + + results: list[DiscoveryResult] = [] + + async def probe() -> None: + async with _mock_client(handler) as client: + results.append( + await discover_server_cards( + "https://example.com", http_client=client, policy=DiscoveryPolicy(timeout_seconds=0.05) + ) + ) + + with anyio.fail_after(5): + async with anyio.create_task_group() as task_group: + task_group.start_soon(probe) + (result,) = results + assert [listing.entry.identifier for listing in result.listings] == ["urn:air:example.com:mcp:good"] + (failure,) = result.failures + assert failure.entry_identifier == "urn:air:example.com:mcp:slow" + assert isinstance(failure.error, TimeoutError) + + +async def test_discover_never_refetches_an_already_visited_catalog(public_dns: None) -> None: + """SDK-defined (best practices): a nested entry pointing back at an already-walked + catalog URL (here the well-known catalog itself) is skipped, not refetched, so a + self-referential catalog terminates after one fetch with no failure.""" + entries = [ + { + "identifier": "urn:air:example.com:catalog:self", + "type": CATALOG_MEDIA_TYPE, + "url": "https://example.com/.well-known/ai-catalog.json", + }, + {"identifier": "urn:air:example.com:mcp:good", "type": CARD_MEDIA_TYPE, "data": json.loads(_card_bytes())}, + ] + seen_paths: list[str] = [] + + def handler(request: httpx2.Request) -> httpx2.Response: + seen_paths.append(request.url.path) + return httpx2.Response(200, content=_catalog_bytes(entries), headers={"Content-Type": CATALOG_MEDIA_TYPE}) + + async with _mock_client(handler) as client: + result = await discover_server_cards("https://example.com", http_client=client) + assert seen_paths == ["/.well-known/ai-catalog.json"] + assert [listing.entry.identifier for listing in result.listings] == ["urn:air:example.com:mcp:good"] + assert result.failures == [] + + +async def test_discover_stops_at_the_probe_budget_with_a_single_failure(public_dns: None) -> None: + """SDK-defined (best practices): the per-catalog entry cap and the depth cap are + multiplicative, so a hostile catalog tree could amplify one probe into + `entries**depth` fetches. `max_probe_entries` bounds the aggregate walk: everything + past the budget is dropped after one `probe_budget` failure, never one per entry.""" + + def card_entry(number: int) -> dict[str, Any]: + return { + "identifier": f"urn:air:example.com:mcp:c{number}", + "type": CARD_MEDIA_TYPE, + "url": f"https://example.com/card/{number}", + } + + def nested_entry(number: int) -> dict[str, Any]: + return { + "identifier": f"urn:air:example.com:catalog:n{number}", + "type": CATALOG_MEDIA_TYPE, + "url": f"https://example.com/nested/{number}", + } + + seen_paths: list[str] = [] + + def handler(request: httpx2.Request) -> httpx2.Response: + seen_paths.append(request.url.path) + if request.url.path == "/.well-known/ai-catalog.json": + body = _catalog_bytes([card_entry(1), card_entry(2), nested_entry(1), nested_entry(2)]) + elif request.url.path.startswith("/nested/"): + body = _catalog_bytes([card_entry(3)]) + else: + return httpx2.Response(200, content=_card_bytes(), headers={"Content-Type": CARD_MEDIA_TYPE}) + return httpx2.Response(200, content=body, headers={"Content-Type": CATALOG_MEDIA_TYPE}) + + async with _mock_client(handler) as client: + result = await discover_server_cards( + "https://example.com", http_client=client, policy=DiscoveryPolicy(max_probe_entries=3) + ) + # Budget of 3: card 1, card 2, then nested catalog 1. Its own card entry finds the + # budget spent (one failure), and nested catalog 2 is dropped without another. + assert seen_paths == ["/.well-known/ai-catalog.json", "/card/1", "/card/2", "/nested/1"] + assert [listing.entry.identifier for listing in result.listings] == [ + "urn:air:example.com:mcp:c1", + "urn:air:example.com:mcp:c2", + ] + (failure,) = result.failures + assert failure.entry_identifier is None + assert isinstance(failure.error, DiscoveryError) + expected_reason: DiscoveryErrorReason = "probe_budget" + assert failure.error.reason == expected_reason + + +def test_listing_domains_are_host_and_port_never_userinfo() -> None: + """SDK-defined hardening: the consent-UI domain properties show host[:port] only, so + a `user@host` URL in a hand-built listing cannot lead with a trusted brand; IPv6 + hosts keep their brackets and a URL with no host shows as empty.""" + + def listing_for(url: str) -> CardListing: + entry = CatalogEntry(identifier="urn:air:example.com:mcp:weather", type=CARD_MEDIA_TYPE, url=url) + return CardListing(card=_card(), entry=entry, catalog_url=url, card_url=url) + + assert listing_for("https://github.com@evil.example/card").listing_domain == "evil.example" + assert listing_for("https://user:pass@example.com:8443/card").hosting_domain == "example.com:8443" + assert listing_for("https://[2001:db8::1]:8443/card").listing_domain == "[2001:db8::1]:8443" + assert listing_for("https:///card").listing_domain == "" + + +async def test_discover_with_the_default_client_applies_the_same_guards() -> None: + """SDK-defined: with no `http_client`, one fresh credential-free client serves the + whole probe, and a blocked target still fails before any request.""" + with pytest.raises(DiscoveryError) as exc_info: + await discover_server_cards("https://10.0.0.1/docs") + assert exc_info.value.reason == "blocked_address" + + +async def test_discover_raises_when_the_catalog_itself_fails(public_dns: None) -> None: + """SDK-defined: a probe that cannot read the top-level catalog found nothing usable, + so that failure raises instead of returning an empty result.""" + + def handler(request: httpx2.Request) -> httpx2.Response: + return httpx2.Response(404) + + async with _mock_client(handler) as client: + with pytest.raises(DiscoveryError) as exc_info: + await discover_server_cards("https://example.com", http_client=client) + assert exc_info.value.reason == "status" + + +async def test_discover_rejects_a_non_http_input_url() -> None: + """SDK-defined: only http(s) URLs have an origin to probe.""" + with pytest.raises(ValueError, match="absolute http"): + await discover_server_cards("mailto:ops@example.com") + + +# -- end to end against the real server routes ------------------------------------------------ + + +async def test_serve_discover_and_reconcile_round_trip(public_dns: None) -> None: + """The full loop, spec-mandated end to end. Steps: + + 1. A server publishes its card and catalog with `mount_discovery`. + 2. A host probes the domain with `discover_server_cards` under the default policy. + 3. The listing exposes the endpoint dedup key and the listing chain. + 4. The host connects (in memory) and `reconcile_server_card` finds no mismatch, + because `build_server_card` derived the card from the same identity. + """ + server = MCPServer(name="weather", version="1.4.0", description="Hourly forecasts.") + card = build_server_card( + server, + name="com.example/weather", + remotes=[Remote(type="streamable-http", url="https://mcp.example.com/mcp")], + ) + app = Starlette() + mount_discovery(app, card, public_url="https://mcp.example.com") + + http_client = httpx2.AsyncClient(transport=httpx2.ASGITransport(app=app), base_url="https://mcp.example.com") + async with http_client: + result = await discover_server_cards("https://mcp.example.com/docs", http_client=http_client) + assert result.failures == [] + (listing,) = result.listings + assert listing.card.endpoint_urls() == frozenset({"https://mcp.example.com/mcp"}) + assert (listing.listing_domain, listing.hosting_domain) == ("mcp.example.com", "mcp.example.com") + + async with Client(server) as client: + assert reconcile_server_card(listing.card, client.server_info) == [] + + +async def test_revalidation_round_trip_gets_a_304_from_the_served_card(public_dns: None) -> None: + """SDK-defined (spec issue #33) end to end: the request/parse pair against the real + routes turns a stored ETag into a 304, and a fresh fetch into a parsed card.""" + app = Starlette() + mount_discovery(app, _card(), public_url="https://mcp.example.com") + transport = httpx2.ASGITransport(app=app) + async with httpx2.AsyncClient(transport=transport, base_url="https://mcp.example.com") as http: + first = await http.send(create_server_card_request("https://mcp.example.com/mcp/server-card")) + card = parse_server_card_response(first) + etag = first.headers["etag"] + second = await http.send( + create_server_card_request("https://mcp.example.com/mcp/server-card", if_none_match=etag) + ) + assert card == _card() + assert second.status_code == 304 + + +# -- URL helpers ------------------------------------------------------------------------------- + + +@pytest.mark.parametrize( + ("url", "expected"), + [ + ("https://example.com/any/page?q=1", "https://example.com/.well-known/ai-catalog.json"), + ("https://example.com:8443", "https://example.com:8443/.well-known/ai-catalog.json"), + ("http://localhost:8000/mcp", "http://localhost:8000/.well-known/ai-catalog.json"), + ], +) +def test_well_known_ai_catalog_url_uses_only_the_origin(url: str, expected: str) -> None: + """Spec-mandated: domain-level discovery reads the origin's well-known path, + whatever page the input URL pointed at.""" + assert well_known_ai_catalog_url(url) == expected + + +@pytest.mark.parametrize("url", ["mailto:ops@example.com", "example.com/docs", ""]) +def test_well_known_ai_catalog_url_rejects_non_http_input(url: str) -> None: + """SDK-defined: a URL without an http(s) origin has no well-known path.""" + with pytest.raises(ValueError, match="absolute http"): + well_known_ai_catalog_url(url) + + +@pytest.mark.parametrize( + ("url", "expected"), + [ + ("https://example.com/mcp", "https://example.com/mcp/server-card"), + ("https://example.com/mcp/", "https://example.com/mcp/server-card"), + ("https://example.com:8443/api/mcp", "https://example.com:8443/api/mcp/server-card"), + ("https://example.com", "https://example.com/server-card"), + ], +) +def test_server_card_url_appends_the_suffix_to_the_transport_url(url: str, expected: str) -> None: + """Spec-mandated: the reserved suffix anchors to the streamable HTTP URL, not the + domain root, with any trailing slash stripped first.""" + assert server_card_url(url) == expected + + +@pytest.mark.parametrize("url", ["ftp://example.com/mcp", "/mcp"]) +def test_server_card_url_rejects_non_http_input(url: str) -> None: + """SDK-defined: the transport URL must be absolute http(s).""" + with pytest.raises(ValueError, match="absolute http"): + server_card_url(url) + + +# -- load_server_card --------------------------------------------------------------------------- + + +def test_load_server_card_reads_a_local_file(tmp_path: Path) -> None: + """SDK-defined: a card can be loaded from disk with no network involved.""" + path = tmp_path / "server-card.json" + path.write_bytes(_card_bytes()) + assert load_server_card(path) == _card() + + +def test_load_server_card_missing_file_raises_oserror(tmp_path: Path) -> None: + """SDK-defined: file problems surface as `OSError`, not as validation noise.""" + with pytest.raises(OSError): + load_server_card(tmp_path / "absent.json") + + +def test_load_server_card_invalid_document_raises_validation_error(tmp_path: Path) -> None: + """SDK-defined: a readable file that is not a card fails validation.""" + path = tmp_path / "server-card.json" + path.write_text('{"version": "1.0.0"}') + with pytest.raises(ValidationError): + load_server_card(path) + + +# -- request/parse pairs -------------------------------------------------------------------------- + + +def test_create_requests_carry_accept_and_optional_if_none_match() -> None: + """SDK-defined: the request builders set the media-type Accept header and attach + `If-None-Match` only when a stored ETag is passed.""" + plain = create_server_card_request("https://example.com/mcp/server-card") + assert plain.method == "GET" + assert plain.headers["accept"] == "application/mcp-server-card+json, application/json;q=0.5" + assert "if-none-match" not in plain.headers + + conditional = create_server_card_request("https://example.com/mcp/server-card", if_none_match='"abc"') + assert conditional.headers["if-none-match"] == '"abc"' + + catalog_plain = create_ai_catalog_request("https://example.com/.well-known/ai-catalog.json") + assert catalog_plain.headers["accept"] == "application/ai-catalog+json, application/json;q=0.5" + assert "if-none-match" not in catalog_plain.headers + + catalog_conditional = create_ai_catalog_request( + "https://example.com/.well-known/ai-catalog.json", if_none_match='"abc"' + ) + assert catalog_conditional.headers["if-none-match"] == '"abc"' + + +def _response(status: int, content: bytes, media_type: str, url: str) -> httpx2.Response: + return httpx2.Response( + status, + content=content, + headers={"Content-Type": media_type}, + request=httpx2.Request("GET", url), + ) + + +def test_parse_responses_return_the_documents() -> None: + """SDK-defined: the parse half applies status and media type checks, then validates.""" + card = parse_server_card_response(_response(200, _card_bytes(), CARD_MEDIA_TYPE, "https://example.com/c")) + assert card == _card() + catalog = parse_ai_catalog_response(_response(200, _catalog_bytes([]), CATALOG_MEDIA_TYPE, "https://example.com/w")) + assert catalog == AICatalog(spec_version="1.0", entries=[]) + + +def test_parse_raises_on_a_304_so_callers_branch_first() -> None: + """SDK-defined: a 304 has no body to parse. Callers check the status code before + calling parse, as the revalidation recipe shows.""" + with pytest.raises(DiscoveryError) as exc_info: + parse_server_card_response(_response(304, b"", CARD_MEDIA_TYPE, "https://example.com/c")) + assert exc_info.value.reason == "status" + + +def test_parse_rejects_a_wrong_media_type() -> None: + """SDK-defined: the media type discipline applies even on caller-owned transports.""" + with pytest.raises(DiscoveryError) as exc_info: + parse_ai_catalog_response(_response(200, _catalog_bytes([]), "text/html", "https://example.com/w")) + assert exc_info.value.reason == "media_type" + + +# -- reconcile_server_card -------------------------------------------------------------------------- + + +def test_reconcile_accepts_the_local_name_part_and_exact_version() -> None: + """Spec-mandated consistency: `serverInfo.name` matching the card name's post-slash + local part counts as consistent, and equal versions produce no mismatch.""" + server_info = Implementation(name="weather", version="1.4.0") + assert reconcile_server_card(_card(), server_info) == [] + + +def test_reconcile_accepts_the_full_namespaced_name() -> None: + """SDK-defined: a server reporting the full namespaced name is also consistent.""" + server_info = Implementation(name="com.example/weather", version="1.4.0") + assert reconcile_server_card(_card(), server_info) == [] + + +def test_reconcile_reports_name_and_version_mismatches() -> None: + """Spec-mandated consistency: disagreements come back as data for logging or UI, + never as an exception. Runtime values win.""" + server_info = Implementation(name="calendar", version="2.0.0") + mismatches = reconcile_server_card(_card(), server_info) + assert [(m.field, m.card_value, m.runtime_value) for m in mismatches] == snapshot( + [ + ("name", "com.example/weather", "calendar"), + ("version", "1.4.0", "2.0.0"), + ] + ) + + +def test_reconcile_checks_the_protocol_version_against_declared_unions() -> None: + """SDK-defined: with `protocol_version=` given, the union of every remote's declared + `supportedProtocolVersions` is consulted. A member passes, a stranger mismatches.""" + card = ServerCard( + name="com.example/weather", + version="1.4.0", + description="Hourly forecasts.", + remotes=[ + Remote(type="streamable-http", url="https://a.example.com/mcp", supported_protocol_versions=["2025-06-18"]), + Remote(type="sse", url="https://b.example.com/sse", supported_protocol_versions=["2025-11-25"]), + ], + ) + server_info = Implementation(name="weather", version="1.4.0") + assert reconcile_server_card(card, server_info, protocol_version="2025-11-25") == [] + mismatches = reconcile_server_card(card, server_info, protocol_version="2026-07-28") + assert [(m.field, m.card_value, m.runtime_value) for m in mismatches] == snapshot( + [("protocol_versions", "2025-06-18, 2025-11-25", "2026-07-28")] + ) + + +def test_reconcile_skips_the_protocol_check_without_declared_versions() -> None: + """SDK-defined: a card whose remotes declare no protocol versions makes no claim, so + any negotiated version is consistent.""" + server_info = Implementation(name="weather", version="1.4.0") + assert reconcile_server_card(_card(), server_info, protocol_version="2026-07-28") == [] + + +async def test_discover_records_a_failure_for_an_unreachable_nested_catalog(public_dns: None) -> None: + """SDK-defined (best practices): a nested catalog that cannot be fetched becomes a + failure entry, and the probe still returns.""" + nested = { + "identifier": "urn:air:example.com:catalog:gone", + "type": CATALOG_MEDIA_TYPE, + "url": "https://example.com/gone.json", + } + + def handler(request: httpx2.Request) -> httpx2.Response: + if request.url.path == "/.well-known/ai-catalog.json": + return httpx2.Response(200, content=_catalog_bytes([nested]), headers={"Content-Type": CATALOG_MEDIA_TYPE}) + return httpx2.Response(404) + + async with _mock_client(handler) as client: + result = await discover_server_cards("https://example.com", http_client=client) + assert result.listings == [] + (failure,) = result.failures + assert failure.entry_identifier == "urn:air:example.com:catalog:gone" + assert isinstance(failure.error, DiscoveryError) + assert failure.error.reason == "status" diff --git a/tests/docs_src/test_server_cards.py b/tests/docs_src/test_server_cards.py new file mode 100644 index 0000000000..a658bb948a --- /dev/null +++ b/tests/docs_src/test_server_cards.py @@ -0,0 +1,126 @@ +"""`docs/advanced/server-cards.md`: every claim the page makes, proved against the real SDK.""" + +import ipaddress +from pathlib import Path + +import httpx2 +import pytest + +from docs_src.server_cards import tutorial001, tutorial002, tutorial003, tutorial004 +from mcp import Client +from mcp.client.experimental import _discovery_http +from mcp.client.experimental.server_card import discover_server_cards, load_server_card, reconcile_server_card +from mcp.shared.experimental.server_card import Input, Remote, resolve_remote + +# See test_index.py for why this is a per-module mark and not a conftest hook. +pytestmark = [pytest.mark.anyio, pytest.mark.filterwarnings("error::mcp.MCPDeprecationWarning")] + +INITIALIZE = { + "jsonrpc": "2.0", + "id": 1, + "method": "initialize", + "params": {"protocolVersion": "2025-06-18", "capabilities": {}, "clientInfo": {"name": "d", "version": "1"}}, +} +MCP_HEADERS = {"Accept": "application/json, text/event-stream", "Content-Type": "application/json"} + + +@pytest.fixture +def public_dns(monkeypatch: pytest.MonkeyPatch) -> None: + """Pin DNS to a public address so the default discovery policy runs offline. + + The SSRF guard resolves every hostname before the request, and the page's + example domains do not resolve in CI. + """ + + async def resolve(host: str) -> list[ipaddress.IPv4Address | ipaddress.IPv6Address]: + return [ipaddress.ip_address("93.184.216.34")] + + monkeypatch.setattr(_discovery_http, "_host_addresses", resolve) + + +async def test_mount_discovery_serves_both_endpoints_with_the_spec_headers() -> None: + """tutorial001: the two GET endpoints exist beside the live transport, with the + page's promised Content-Type, CORS, Cache-Control and ETag/304 behavior.""" + transport = httpx2.ASGITransport(app=tutorial001.app) + async with tutorial001.mcp.session_manager.run(): + async with httpx2.AsyncClient(transport=transport, base_url="http://localhost:8000") as http: + initialize = await http.post("/mcp", json=INITIALIZE, headers=MCP_HEADERS) + card_response = await http.get("/mcp/server-card") + catalog_response = await http.get("/.well-known/ai-catalog.json") + revalidated = await http.get("/mcp/server-card", headers={"If-None-Match": card_response.headers["etag"]}) + assert initialize.status_code == 200 + assert card_response.headers["content-type"] == "application/mcp-server-card+json" + assert card_response.headers["access-control-allow-origin"] == "*" + assert card_response.headers["access-control-allow-methods"] == "GET" + assert card_response.headers["access-control-allow-headers"] == "Content-Type" + assert card_response.headers["cache-control"] == "public, max-age=3600" + assert catalog_response.headers["content-type"] == "application/ai-catalog+json" + assert (revalidated.status_code, revalidated.content) == (304, b"") + + +async def test_build_server_card_keeps_the_card_consistent_with_server_info() -> None: + """tutorial001 + the page's derivation claim: the card carries the server's version, + description and websiteUrl, so `reconcile_server_card` finds no drift after connect.""" + assert tutorial001.card.version == "1.4.0" + assert tutorial001.card.description == "Hourly forecasts." + assert tutorial001.card.website_url == "https://example.com" + async with Client(tutorial001.mcp) as client: + assert reconcile_server_card(tutorial001.card, client.server_info) == [] + + +async def test_the_discovery_probe_finds_the_served_card(public_dns: None) -> None: + """tutorial002: `discover_server_cards` on any page of the origin finds the card + tutorial001 mounted, and the listing exposes the consent-UI domains and the endpoint + dedup key. The tutorial's `main()` itself needs a live network, so the flow is + proved here against the in-memory app.""" + http_client = httpx2.AsyncClient( + transport=httpx2.ASGITransport(app=tutorial001.app), base_url="https://mcp.example.com" + ) + async with http_client: + result = await discover_server_cards("https://mcp.example.com/docs", http_client=http_client) + assert result.failures == [] + (listing,) = result.listings + assert listing.entry.identifier == "urn:air:mcp.example.com:mcp:weather" + assert (listing.listing_domain, listing.hosting_domain) == ("mcp.example.com", "mcp.example.com") + assert listing.card.endpoint_urls() == frozenset({"https://mcp.example.com/mcp"}) + assert listing.card == tutorial001.card + + +def test_resolve_remote_names_every_missing_required_input() -> None: + """tutorial002's comment: `resolve_remote` raises a ValueError naming the missing + required inputs so a host can prompt for all of them at once.""" + remote = Remote( + type="streamable-http", + url="https://{tenant}.example.com/mcp", + variables={"tenant": Input(is_required=True)}, + ) + assert remote.required_variables == frozenset({"tenant"}) + with pytest.raises(ValueError, match="tenant"): + resolve_remote(remote) + assert resolve_remote(remote, {"tenant": "acme"}).url == "https://acme.example.com/mcp" + + +async def test_the_revalidation_recipe_costs_a_304_when_unchanged() -> None: + """tutorial003: the first `refresh` parses and stores the card, the second sends the + stored ETag and reuses the cache on the 304.""" + store = tutorial003.CardStore() + transport = httpx2.ASGITransport(app=tutorial001.app) + async with httpx2.AsyncClient(transport=transport, base_url="https://mcp.example.com") as http: + first = await tutorial003.refresh(store, http, "https://mcp.example.com/mcp/server-card") + second = await tutorial003.refresh(store, http, "https://mcp.example.com/mcp/server-card") + assert first == tutorial001.card + assert second is first + assert store.etags["https://mcp.example.com/mcp/server-card"].startswith('"') + + +def test_static_publishing_writes_a_loadable_card(tmp_path: Path) -> None: + """tutorial004: the written file is a valid card document that loads back.""" + target = tmp_path / "server-card.json" + tutorial004.publish(target) + assert load_server_card(target) == tutorial004.card + + +def test_the_connect_tutorial_exposes_its_client_program() -> None: + """tutorial002's `main()` needs a live network, so its coverage here is the import + plus the in-memory proof of the same flow above.""" + assert callable(tutorial002.main) diff --git a/tests/server/experimental/__init__.py b/tests/server/experimental/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/tests/server/experimental/test_server_card.py b/tests/server/experimental/test_server_card.py new file mode 100644 index 0000000000..70237caf58 --- /dev/null +++ b/tests/server/experimental/test_server_card.py @@ -0,0 +1,404 @@ +"""`mcp.server.experimental.server_card`: card building, discovery routes and mounts.""" + +from collections.abc import Callable +from typing import Any + +import httpx2 +import pytest +from inline_snapshot import snapshot +from mcp_types import Icon +from pydantic import ValidationError +from starlette.applications import Starlette +from starlette.requests import Request + +from mcp.server import MCPServer, Server +from mcp.server.experimental.server_card import ( + build_server_card, + catalog_identifier, + create_ai_catalog_routes, + create_server_card_routes, + discovery_response, + mount_ai_catalog, + mount_discovery, + mount_server_card, + server_card_entry, +) +from mcp.shared.experimental.ai_catalog import AICatalog +from mcp.shared.experimental.server_card import Remote, Repository, ServerCard + +pytestmark = pytest.mark.anyio + +INITIALIZE = { + "jsonrpc": "2.0", + "id": 1, + "method": "initialize", + "params": {"protocolVersion": "2025-06-18", "capabilities": {}, "clientInfo": {"name": "t", "version": "1"}}, +} +MCP_HEADERS = {"Accept": "application/json, text/event-stream", "Content-Type": "application/json"} + + +def _card() -> ServerCard: + return ServerCard( + name="com.example/weather", + version="1.4.0", + description="Hourly forecasts.", + title="Weather", + remotes=[Remote(type="streamable-http", url="https://mcp.example.com/mcp")], + ) + + +def _client_for(routes_app: Starlette) -> httpx2.AsyncClient: + return httpx2.AsyncClient(transport=httpx2.ASGITransport(app=routes_app), base_url="https://mcp.example.com") + + +# -- build_server_card ------------------------------------------------------------------- + + +def test_build_server_card_derives_identity_from_an_mcpserver() -> None: + """SDK-defined: title, description, version, website URL and icons come from the + server object, keeping the card consistent with runtime `serverInfo`.""" + server = MCPServer( + name="Weather", + title="Weather", + version="1.4.0", + description="Hourly forecasts.", + website_url="https://example.com", + ) + card = build_server_card(server, name="com.example/weather") + assert card.model_dump(by_alias=True, exclude_none=True) == snapshot( + { + "$schema": "https://static.modelcontextprotocol.io/schemas/v1/server-card.schema.json", + "name": "com.example/weather", + "version": "1.4.0", + "description": "Hourly forecasts.", + "title": "Weather", + "websiteUrl": "https://example.com", + } + ) + + +def test_build_server_card_derives_identity_from_a_lowlevel_server() -> None: + """SDK-defined: the lowlevel `Server` satisfies the same identity surface as + `MCPServer`, via plain attributes instead of properties.""" + server = Server("weather", version="2.0.0", description="Forecasts.", title="Weather") + card = build_server_card(server, name="com.example/weather") + assert (card.version, card.description, card.title) == ("2.0.0", "Forecasts.", "Weather") + + +def test_build_server_card_explicit_kwargs_override_derived_values() -> None: + """SDK-defined: every explicit keyword argument beats the server-derived value, and + `repository`/`icons` (never derivable from a server object) land on the card.""" + server = MCPServer(name="Weather", version="1.4.0", description="Hourly forecasts.") + remotes = [Remote(type="streamable-http", url="https://mcp.example.com/mcp")] + repository = Repository(url="https://github.com/example/weather", source="github") + icons = [Icon(src="https://example.com/icon.png", mime_type="image/png", sizes=["48x48"])] + card = build_server_card( + server, + name="com.example/weather", + version="9.9.9", + description="Overridden.", + title="Custom", + website_url="https://override.example.com", + remotes=remotes, + repository=repository, + icons=icons, + meta={"com.example/build": 7}, + ) + assert card.version == "9.9.9" + assert card.description == "Overridden." + assert card.title == "Custom" + assert card.website_url == "https://override.example.com" + assert card.remotes == remotes + assert card.repository == repository + assert card.icons == icons + assert card.meta == {"com.example/build": 7} + + +def test_build_server_card_rejects_a_server_without_a_version() -> None: + """SDK-defined: a card requires a version, so a server that has none and no + `version=` override fails card validation.""" + server = MCPServer(name="Weather", description="Hourly forecasts.") + with pytest.raises(ValidationError): + build_server_card(server, name="com.example/weather") + + +def test_build_server_card_rejects_an_overlong_derived_description() -> None: + """SDK-defined: the card's 100 character description cap applies to derived values + too, surfacing as `pydantic.ValidationError`.""" + server = MCPServer(name="Weather", version="1.0.0", description="x" * 101) + with pytest.raises(ValidationError): + build_server_card(server, name="com.example/weather") + + +# -- serving the card -------------------------------------------------------------------- + + +async def test_card_is_served_with_its_media_type_at_the_reserved_path() -> None: + """Spec-mandated: the card lives at the streamable HTTP path plus `/server-card` + and is served as `application/mcp-server-card+json`.""" + card = _card() + async with _client_for(Starlette(routes=create_server_card_routes(card))) as client: + response = await client.get("/mcp/server-card") + assert response.status_code == 200 + assert response.headers["content-type"] == "application/mcp-server-card+json" + assert ServerCard.model_validate_json(response.content) == card + + +async def test_card_responses_carry_the_three_cors_headers() -> None: + """Spec-mandated MUST: `Access-Control-Allow-Origin: *`, `-Methods: GET` and + `-Headers: Content-Type` on the card response.""" + async with _client_for(Starlette(routes=create_server_card_routes(_card()))) as client: + response = await client.get("/mcp/server-card") + assert response.headers["access-control-allow-origin"] == "*" + assert response.headers["access-control-allow-methods"] == "GET" + assert response.headers["access-control-allow-headers"] == "Content-Type" + + +async def test_card_responses_carry_the_default_cache_control() -> None: + """Spec-mandated SHOULD: hosts send caching headers, defaulting to the spec's + example of one hour.""" + async with _client_for(Starlette(routes=create_server_card_routes(_card()))) as client: + response = await client.get("/mcp/server-card") + assert response.headers["cache-control"] == "public, max-age=3600" + + +async def test_options_preflight_returns_the_cors_headers_and_no_body() -> None: + """SDK-defined: a bare OPTIONS gets 200 with the CORS headers and an empty body.""" + async with _client_for(Starlette(routes=create_server_card_routes(_card()))) as client: + response = await client.options("/mcp/server-card") + assert response.status_code == 200 + assert response.content == b"" + assert response.headers["access-control-allow-origin"] == "*" + + +async def test_browser_preflight_through_the_cors_middleware_is_allowed() -> None: + """Spec-mandated: a real browser preflight (Origin plus requested method) succeeds + and advertises GET as the only allowed method, so web-based hosts can fetch the card + cross-origin. Starlette answers this one, not `discovery_response`.""" + async with _client_for(Starlette(routes=create_server_card_routes(_card()))) as client: + response = await client.options( + "/mcp/server-card", + headers={"Origin": "https://host.example.org", "Access-Control-Request-Method": "GET"}, + ) + assert response.status_code == 200 + assert response.headers["access-control-allow-origin"] == "*" + assert response.headers["access-control-allow-methods"] == "GET" + + +# -- ETag / If-None-Match ---------------------------------------------------------------- + + +async def test_etag_is_strong_and_stable_across_requests() -> None: + """SDK-defined (spec issue #33): every 200 carries a strong SHA-256 ETag that does + not change while the card does not.""" + async with _client_for(Starlette(routes=create_server_card_routes(_card()))) as client: + first = await client.get("/mcp/server-card") + second = await client.get("/mcp/server-card") + etag = first.headers["etag"] + assert etag.startswith('"') and etag.endswith('"') + assert second.headers["etag"] == etag + + +def _exact_tag(tag: str) -> str: + return tag + + +def _weak_tag(tag: str) -> str: + return f"W/{tag}" + + +def _star_tag(tag: str) -> str: + return "*" + + +def _list_member_tag(tag: str) -> str: + return f'"nope", {tag}' + + +@pytest.mark.parametrize( + "wrap", + [_exact_tag, _weak_tag, _star_tag, _list_member_tag], + ids=["exact", "weak", "star", "list-member"], +) +async def test_matching_if_none_match_returns_304_with_headers(wrap: Callable[[str], str]) -> None: + """SDK-defined (spec issue #33): the exact tag, its weak form, `*` and a list member + all revalidate to an empty 304 that keeps the ETag, CORS and caching headers.""" + async with _client_for(Starlette(routes=create_server_card_routes(_card()))) as client: + first = await client.get("/mcp/server-card") + etag = first.headers["etag"] + response = await client.get("/mcp/server-card", headers={"If-None-Match": wrap(etag)}) + assert response.status_code == 304 + assert response.content == b"" + assert response.headers["etag"] == etag + assert response.headers["access-control-allow-origin"] == "*" + assert response.headers["cache-control"] == "public, max-age=3600" + + +async def test_non_matching_if_none_match_returns_the_full_card() -> None: + """SDK-defined: a stale ETag misses and the full 200 comes back.""" + async with _client_for(Starlette(routes=create_server_card_routes(_card()))) as client: + response = await client.get("/mcp/server-card", headers={"If-None-Match": '"stale"'}) + assert response.status_code == 200 + assert response.content != b"" + + +# -- path resolution and cache control knobs ---------------------------------------------- + + +async def test_card_path_follows_the_streamable_http_path() -> None: + """Spec-mandated: the reserved suffix anchors to the transport path, not the domain + root, and a trailing slash on the transport path does not double up.""" + routes = create_server_card_routes(_card(), streamable_http_path="/api/mcp/") + assert [route.path for route in routes] == ["/api/mcp/server-card"] + + +async def test_explicit_path_overrides_the_reserved_location() -> None: + """Spec-mandated: a card MAY be hosted at any unreserved URI.""" + routes = create_server_card_routes(_card(), path="/cards/weather.json") + assert [route.path for route in routes] == ["/cards/weather.json"] + + +async def test_cache_control_override_and_empty_string_omission() -> None: + """SDK-defined: `discovery_response`'s `cache_control` is tunable through custom + handlers; the built-in routes always use the default. An empty string omits the + header entirely.""" + scope: dict[str, Any] = {"type": "http", "method": "GET", "headers": [], "query_string": b"", "path": "/c"} + tuned = discovery_response(Request(scope), b"{}", "application/json", cache_control="public, max-age=60") + assert tuned.headers["cache-control"] == "public, max-age=60" + omitted = discovery_response(Request(scope), b"{}", "application/json", cache_control="") + assert "cache-control" not in omitted.headers + + +# -- the catalog routes -------------------------------------------------------------------- + + +async def test_catalog_is_served_at_the_well_known_path_with_its_media_type() -> None: + """Spec-mandated: domain-level discovery reads `/.well-known/ai-catalog.json` served + as `application/ai-catalog+json`, with the same CORS and caching headers.""" + catalog = AICatalog(spec_version="1.0", entries=[server_card_entry(_card(), publisher="example.com")]) + async with _client_for(Starlette(routes=create_ai_catalog_routes(catalog))) as client: + response = await client.get("/.well-known/ai-catalog.json") + assert response.status_code == 200 + assert response.headers["content-type"] == "application/ai-catalog+json" + assert response.headers["access-control-allow-origin"] == "*" + assert AICatalog.model_validate_json(response.content) == catalog + + +# -- catalog_identifier and server_card_entry ---------------------------------------------- + + +def test_catalog_identifier_includes_the_mcp_namespace_segment() -> None: + """Spec-mandated: the URN is `urn:air:{publisher}:mcp:{name}` for card + `com.example/weather` published by example.com.""" + assert catalog_identifier("com.example/weather", publisher="example.com") == "urn:air:example.com:mcp:weather" + + +def test_catalog_identifier_rejects_a_name_without_a_namespace() -> None: + """SDK-defined: a card name outside the `namespace/name` pattern has no URN.""" + with pytest.raises(ValueError, match="namespace/name"): + catalog_identifier("weather", publisher="example.com") + + +def test_server_card_entry_with_url_points_at_the_hosted_card() -> None: + """Spec-mandated: a URL entry carries the card's media type and does not duplicate + the card's fields beyond `displayName`.""" + entry = server_card_entry(_card(), publisher="example.com", url="https://mcp.example.com/mcp/server-card") + assert entry.model_dump(by_alias=True, exclude_none=True) == snapshot( + { + "identifier": "urn:air:example.com:mcp:weather", + "type": "application/mcp-server-card+json", + "url": "https://mcp.example.com/mcp/server-card", + "displayName": "Weather", + } + ) + + +def test_server_card_entry_without_url_inlines_the_full_card() -> None: + """Spec-mandated: `url=None` inlines the card as the entry's `data`, and the inline + document parses back to the same card.""" + card = _card() + entry = server_card_entry(card, publisher="example.com") + assert entry.url is None + assert ServerCard.model_validate(entry.data) == card + + +def test_server_card_entry_honors_an_explicit_identifier() -> None: + """SDK-defined: a caller-chosen identifier wins over the derived URN.""" + entry = server_card_entry(_card(), publisher="example.com", identifier="urn:air:example.com:custom:weather") + assert entry.identifier == "urn:air:example.com:custom:weather" + + +@pytest.mark.parametrize("url", ["/mcp/server-card", "http://mcp.example.com/mcp/server-card"]) +def test_server_card_entry_rejects_relative_and_insecure_urls(url: str) -> None: + """Spec-mandated: catalogs are fetched cross-domain, so entry URLs must be absolute, + and production transport must be HTTPS.""" + with pytest.raises(ValueError, match="URL|loopback"): + server_card_entry(_card(), publisher="example.com", url=url) + + +def test_server_card_entry_allows_plain_http_to_loopback() -> None: + """SDK-defined: plain http stays available for local development.""" + entry = server_card_entry(_card(), publisher="localhost", url="http://localhost:8000/mcp/server-card") + assert entry.url == "http://localhost:8000/mcp/server-card" + + +# -- mounting on a built app ---------------------------------------------------------------- + + +async def test_mount_discovery_serves_both_endpoints_beside_a_live_transport() -> None: + """Spec-mandated end to end: after `mount_discovery` on `streamable_http_app()`, the + card, the catalog and the MCP transport all answer on one app, and the catalog entry + points at the card's absolute URL.""" + server = MCPServer(name="Weather", version="1.4.0", description="Hourly forecasts.") + app = server.streamable_http_app() + card = build_server_card( + server, + name="com.example/weather", + remotes=[Remote(type="streamable-http", url="https://mcp.example.com/mcp")], + ) + mount_discovery(app, card, public_url="https://mcp.example.com") + + transport = httpx2.ASGITransport(app=app) + async with server.session_manager.run(): + # The default transport security allows `localhost:*` hosts only, which is + # fine here: the discovery routes carry no host restriction. + async with httpx2.AsyncClient(transport=transport, base_url="http://localhost:8000") as client: + initialize = await client.post("/mcp", json=INITIALIZE, headers=MCP_HEADERS) + card_response = await client.get("/mcp/server-card") + catalog_response = await client.get("/.well-known/ai-catalog.json") + assert initialize.status_code == 200 + assert ServerCard.model_validate_json(card_response.content) == card + catalog = AICatalog.model_validate_json(catalog_response.content) + assert catalog.entries[0].url == "https://mcp.example.com/mcp/server-card" + assert catalog.entries[0].identifier == "urn:air:mcp.example.com:mcp:weather" + + +async def test_mount_server_card_and_mount_ai_catalog_append_routes_independently() -> None: + """SDK-defined: the single-purpose mounts add exactly their own endpoint to an + existing app.""" + app = Starlette() + mount_server_card(app, _card(), streamable_http_path="/api/mcp") + mount_ai_catalog(app, AICatalog(spec_version="1.0", entries=[]), path="/catalog.json") + async with _client_for(app) as client: + card_response = await client.get("/api/mcp/server-card") + catalog_response = await client.get("/catalog.json") + assert card_response.status_code == 200 + assert catalog_response.status_code == 200 + + +@pytest.mark.parametrize("public_url", ["http://mcp.example.com", "example.com/mcp", "ftp://mcp.example.com"]) +def test_mount_discovery_rejects_insecure_or_relative_public_urls(public_url: str) -> None: + """Spec-mandated: HTTPS is a MUST in production, so `public_url` must be absolute + http(s) and plain http is loopback-only.""" + with pytest.raises(ValueError, match="URL|loopback"): + mount_discovery(Starlette(), _card(), public_url=public_url) + + +async def test_mount_discovery_allows_plain_http_to_loopback() -> None: + """SDK-defined: local development mounts against `http://localhost` work.""" + app = Starlette() + mount_discovery(app, _card(), public_url="http://localhost:8000") + async with _client_for(app) as client: + response = await client.get("/.well-known/ai-catalog.json") + entry = AICatalog.model_validate_json(response.content).entries[0] + assert entry.url == "http://localhost:8000/mcp/server-card" diff --git a/tests/shared/experimental/__init__.py b/tests/shared/experimental/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/tests/shared/experimental/fixtures/server_card/invalid/bad-name-pattern.json b/tests/shared/experimental/fixtures/server_card/invalid/bad-name-pattern.json new file mode 100644 index 0000000000..5f9d7c6c9f --- /dev/null +++ b/tests/shared/experimental/fixtures/server_card/invalid/bad-name-pattern.json @@ -0,0 +1,6 @@ +{ + "$schema": "https://static.modelcontextprotocol.io/schemas/v1/server-card.schema.json", + "name": "no-slash-in-name", + "version": "1.0.0", + "description": "Name must contain exactly one slash separating namespace and server name." +} diff --git a/tests/shared/experimental/fixtures/server_card/invalid/date-versioned-schema.json b/tests/shared/experimental/fixtures/server_card/invalid/date-versioned-schema.json new file mode 100644 index 0000000000..53314cdf4c --- /dev/null +++ b/tests/shared/experimental/fixtures/server_card/invalid/date-versioned-schema.json @@ -0,0 +1,6 @@ +{ + "$schema": "https://static.modelcontextprotocol.io/schemas/2025-11-25/server-card.schema.json", + "name": "example-org/wrong-schema-url", + "version": "1.0.0", + "description": "Uses date-versioned schema URL; only /v1/ URLs are valid." +} diff --git a/tests/shared/experimental/fixtures/server_card/invalid/missing-name.json b/tests/shared/experimental/fixtures/server_card/invalid/missing-name.json new file mode 100644 index 0000000000..8269b209b3 --- /dev/null +++ b/tests/shared/experimental/fixtures/server_card/invalid/missing-name.json @@ -0,0 +1,5 @@ +{ + "$schema": "https://static.modelcontextprotocol.io/schemas/v1/server-card.schema.json", + "version": "1.0.0", + "description": "Missing required `name` field." +} diff --git a/tests/shared/experimental/fixtures/server_card/invalid/missing-schema.json b/tests/shared/experimental/fixtures/server_card/invalid/missing-schema.json new file mode 100644 index 0000000000..f541b4757b --- /dev/null +++ b/tests/shared/experimental/fixtures/server_card/invalid/missing-schema.json @@ -0,0 +1,5 @@ +{ + "name": "example-org/no-schema", + "version": "1.0.0", + "description": "Missing the required `$schema` field." +} diff --git a/tests/shared/experimental/fixtures/server_card/invalid/wrong-schema-name.json b/tests/shared/experimental/fixtures/server_card/invalid/wrong-schema-name.json new file mode 100644 index 0000000000..e4f822bfb1 --- /dev/null +++ b/tests/shared/experimental/fixtures/server_card/invalid/wrong-schema-name.json @@ -0,0 +1,6 @@ +{ + "$schema": "https://static.modelcontextprotocol.io/schemas/v1/server.schema.json", + "name": "example-org/wrong-schema-name", + "version": "1.0.0", + "description": "References the removed registry server.json schema; only server-card.schema.json is valid in v1." +} diff --git a/tests/shared/experimental/fixtures/server_card/valid/minimal.json b/tests/shared/experimental/fixtures/server_card/valid/minimal.json new file mode 100644 index 0000000000..a09d8edc15 --- /dev/null +++ b/tests/shared/experimental/fixtures/server_card/valid/minimal.json @@ -0,0 +1,6 @@ +{ + "$schema": "https://static.modelcontextprotocol.io/schemas/v1/server-card.schema.json", + "name": "example-org/minimal", + "version": "1.0.0", + "description": "Smallest valid Server Card." +} diff --git a/tests/shared/experimental/fixtures/server_card/valid/templated-remote.json b/tests/shared/experimental/fixtures/server_card/valid/templated-remote.json new file mode 100644 index 0000000000..c30bb1d909 --- /dev/null +++ b/tests/shared/experimental/fixtures/server_card/valid/templated-remote.json @@ -0,0 +1,38 @@ +{ + "$schema": "https://static.modelcontextprotocol.io/schemas/v1/server-card.schema.json", + "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": { + "description": "API token issued by Example Inc.", + "isRequired": true, + "isSecret": true + } + } + } + ], + "variables": { + "tenant": { + "description": "Tenant subdomain.", + "isRequired": true, + "default": "default" + } + }, + "supportedProtocolVersions": ["2025-06-18", "2025-11-25"] + } + ] +} diff --git a/tests/shared/experimental/test_ai_catalog.py b/tests/shared/experimental/test_ai_catalog.py new file mode 100644 index 0000000000..4570eb989d --- /dev/null +++ b/tests/shared/experimental/test_ai_catalog.py @@ -0,0 +1,70 @@ +"""`mcp.shared.experimental.ai_catalog`: catalog models and the url-xor-data rule.""" + +import json + +import pytest +from pydantic import ValidationError + +from mcp.shared.experimental.ai_catalog import AICatalog, CatalogEntry + + +def test_entry_with_url_only_is_valid() -> None: + """Spec-mandated: an entry may reference its artifact by URL.""" + entry = CatalogEntry( + identifier="urn:air:example.com:mcp:weather", + type="application/mcp-server-card+json", + url="https://example.com/mcp/server-card", + ) + assert entry.data is None + + +def test_entry_with_inline_data_only_is_valid() -> None: + """Spec-mandated: an entry may inline its artifact as `data` instead of a URL.""" + entry = CatalogEntry( + identifier="urn:air:example.com:mcp:weather", + type="application/mcp-server-card+json", + data={"name": "com.example/weather"}, + ) + assert entry.url is None + + +def test_entry_with_both_url_and_data_is_rejected() -> None: + """Spec-mandated: `url` and `data` are mutually exclusive.""" + with pytest.raises(ValidationError): + CatalogEntry( + identifier="urn:air:x:mcp:y", type="application/mcp-server-card+json", url="https://example.com/c", data={} + ) + + +def test_entry_with_neither_url_nor_data_is_rejected() -> None: + """Spec-mandated: an entry must carry its artifact one way or the other.""" + with pytest.raises(ValidationError): + CatalogEntry(identifier="urn:air:x:mcp:y", type="application/mcp-server-card+json") + + +def test_catalog_with_empty_entries_is_valid() -> None: + """Spec-mandated: `entries` is required but may be empty.""" + assert AICatalog(spec_version="1.0", entries=[]).entries == [] + + +def test_trust_and_publisher_shapes_pass_through_verbatim() -> None: + """SDK-defined: `publisher`, `trustManifest` and `host` are untyped passthrough for + the host's consent UI. The AI Catalog working group owns their shapes.""" + document = { + "specVersion": "1.0", + "host": {"displayName": "Example Inc.", "identifier": "example.com"}, + "entries": [ + { + "identifier": "urn:air:example.com:mcp:weather", + "type": "application/mcp-server-card+json", + "url": "https://example.com/mcp/server-card", + "displayName": "Weather", + "updatedAt": "2026-07-01T00:00:00Z", + "publisher": {"identifier": "example.com", "displayName": "Example Inc."}, + "trustManifest": {"identity": "did:web:example.com"}, + } + ], + } + catalog = AICatalog.model_validate(document) + assert catalog.entries[0].trust_manifest == {"identity": "did:web:example.com"} + assert json.loads(catalog.model_dump_json(by_alias=True, exclude_none=True)) == document diff --git a/tests/shared/experimental/test_server_card.py b/tests/shared/experimental/test_server_card.py new file mode 100644 index 0000000000..6ec221678a --- /dev/null +++ b/tests/shared/experimental/test_server_card.py @@ -0,0 +1,364 @@ +"""`mcp.shared.experimental.server_card`: models, validators and `resolve_remote`.""" + +import json +from pathlib import Path + +import pytest +from inline_snapshot import snapshot +from pydantic import ValidationError + +from mcp.shared.experimental.server_card import ( + SERVER_CARD_SCHEMA_URL, + Input, + KeyValueInput, + Remote, + ServerCard, + resolve_remote, +) + +FIXTURES = Path(__file__).parent / "fixtures" / "server_card" + +VALID_FIXTURES = sorted((FIXTURES / "valid").glob("*.json")) +INVALID_FIXTURES = sorted(path for path in (FIXTURES / "invalid").glob("*.json") if path.name != "missing-schema.json") + + +def _minimal_card_data(**overrides: object) -> dict[str, object]: + """The smallest valid card document, with per-test field overrides.""" + data: dict[str, object] = { + "$schema": SERVER_CARD_SCHEMA_URL, + "name": "com.example/weather", + "version": "1.0.0", + "description": "Hourly forecasts.", + } + data.update(overrides) + return data + + +# -- conformance fixtures (vendored from the extension repo's examples/ServerCard) ----- + + +@pytest.mark.parametrize("path", VALID_FIXTURES, ids=lambda p: p.name) +def test_valid_fixture_round_trips_byte_for_byte_equivalent(path: Path) -> None: + """Spec-mandated: every valid conformance fixture parses and re-serializes to the + same JSON document, so no field is dropped, renamed or reshaped in transit.""" + card = ServerCard.model_validate_json(path.read_bytes()) + assert json.loads(card.model_dump_json(by_alias=True, exclude_none=True)) == json.loads(path.read_text()) + + +@pytest.mark.parametrize("path", INVALID_FIXTURES, ids=lambda p: p.name) +def test_invalid_fixture_is_rejected(path: Path) -> None: + """Spec-mandated: the invalid conformance fixtures (bad name pattern, wrong or + date-versioned `$schema`, missing `name`) raise `pydantic.ValidationError`.""" + with pytest.raises(ValidationError): + ServerCard.model_validate_json(path.read_bytes()) + + +def test_missing_schema_is_defaulted_on_ingestion() -> None: + """SDK-defined divergence from the fixture set: the spec marks a missing `$schema` + invalid, but the SDK deliberately defaults it so cards from lenient publishers still + parse. Serialization always writes the canonical URL back.""" + card = ServerCard.model_validate_json((FIXTURES / "invalid" / "missing-schema.json").read_bytes()) + assert card.schema_ == SERVER_CARD_SCHEMA_URL + + +# -- field validators ------------------------------------------------------------------- + + +@pytest.mark.parametrize( + "schema_url", + [ + "https://static.modelcontextprotocol.io/schemas/2025-11-25/server-card.schema.json", + "https://static.modelcontextprotocol.io/schemas/v1/server.schema.json", + "https://example.com/server-card.schema.json", + ], +) +def test_any_schema_url_other_than_the_v1_url_is_rejected(schema_url: str) -> None: + """Spec-mandated: `$schema` must be exactly the v1 URL. Date-versioned URLs and the + removed registry `server.schema.json` are both invalid.""" + with pytest.raises(ValidationError): + ServerCard.model_validate(_minimal_card_data(**{"$schema": schema_url})) + + +@pytest.mark.parametrize("name", ["no-slash-in-name", "ab", "a/b/c", "com.example/", "x" * 200 + "/name"]) +def test_name_outside_the_namespace_slash_name_pattern_is_rejected(name: str) -> None: + """Spec-mandated: `name` is 3-200 chars of reverse-DNS namespace, exactly one slash, + then the server name.""" + with pytest.raises(ValidationError): + ServerCard.model_validate(_minimal_card_data(name=name)) + + +@pytest.mark.parametrize( + "version", ["^1.2.3", "~1.2.3", ">=1.2.3", "<=1.2.3", ">1.2.3", "<1.2.3", "1.x", "1.*", "1.0 || 2.0"] +) +def test_version_range_syntax_is_rejected(version: str) -> None: + """Spec-mandated (prose level, not expressible in the JSON Schema): `version` is a + single version, never a range or wildcard.""" + with pytest.raises(ValidationError): + ServerCard.model_validate(_minimal_card_data(version=version)) + + +@pytest.mark.parametrize("version", ["1.0.2", "2.1.0-alpha", "2025-06-18", "not semver"]) +def test_plain_version_strings_including_non_semver_are_accepted(version: str) -> None: + """Spec-mandated: semver is a SHOULD, so plain non-semver strings still pass.""" + assert ServerCard.model_validate(_minimal_card_data(version=version)).version == version + + +@pytest.mark.parametrize("description", ["", "x" * 101]) +def test_description_outside_1_to_100_chars_is_rejected(description: str) -> None: + """Spec-mandated: `description` is required with 1-100 characters.""" + with pytest.raises(ValidationError): + ServerCard.model_validate(_minimal_card_data(description=description)) + + +def test_title_of_101_chars_is_rejected() -> None: + """Spec-mandated: `title`, when present, is 1-100 characters.""" + with pytest.raises(ValidationError): + ServerCard.model_validate(_minimal_card_data(title="x" * 101)) + + +@pytest.mark.parametrize( + "url", ["https://mcp.example.com/mcp", "http://localhost:8000/mcp", "{tenant}.example.com/mcp"] +) +def test_remote_url_accepts_http_https_and_template_prefixes(url: str) -> None: + """Spec-mandated: `remotes[].url` starts with http://, https:// or a `{template}`.""" + assert Remote(type="streamable-http", url=url).url == url + + +@pytest.mark.parametrize("url", ["ftp://example.com/mcp", "mcp.example.com/mcp", ""]) +def test_remote_url_rejects_other_schemes_and_bare_hosts(url: str) -> None: + """Spec-mandated: anything not http(s) or `{template}`-prefixed fails the pattern.""" + with pytest.raises(ValidationError): + Remote(type="streamable-http", url=url) + + +# -- open objects and aliasing ---------------------------------------------------------- + + +def test_vendor_fields_and_meta_survive_a_round_trip() -> None: + """Spec-mandated: card objects are open, so unknown vendor fields and `_meta` must + survive parse and re-serialize.""" + data = _minimal_card_data(**{"com.example/flag": True, "_meta": {"com.example/build": 7}}) + card = ServerCard.model_validate(data) + assert json.loads(card.model_dump_json(by_alias=True, exclude_none=True)) == data + + +def test_icons_and_repository_round_trip_with_their_wire_names() -> None: + """Spec-mandated: `icons[]` (camelCase `mimeType`, `sizes`) and `repository` + survive parse and re-serialize byte-for-byte equivalent.""" + data = _minimal_card_data( + icons=[{"src": "https://example.com/icon.png", "mimeType": "image/png", "sizes": ["48x48", "96x96"]}], + repository={"url": "https://github.com/example/weather", "source": "github", "subfolder": "servers/weather"}, + ) + card = ServerCard.model_validate(data) + assert card.icons is not None + assert (card.icons[0].mime_type, card.icons[0].sizes) == ("image/png", ["48x48", "96x96"]) + assert card.repository is not None + assert (card.repository.url, card.repository.source) == ("https://github.com/example/weather", "github") + assert json.loads(card.model_dump_json(by_alias=True, exclude_none=True)) == data + + +@pytest.mark.parametrize("missing", ["url", "source"]) +def test_repository_missing_a_required_field_is_rejected(missing: str) -> None: + """Spec-mandated: `repository.url` and `repository.source` are both required.""" + repository = {"url": "https://github.com/example/weather", "source": "github"} + del repository[missing] + with pytest.raises(ValidationError): + ServerCard.model_validate(_minimal_card_data(repository=repository)) + + +def test_fields_populate_by_python_name_and_serialize_camel_case() -> None: + """SDK-defined: models accept snake_case constructor names and always write the + spec's camelCase wire names.""" + card = ServerCard( + name="com.example/weather", version="1.0.0", description="Forecasts.", website_url="https://example.com" + ) + dumped = json.loads(card.model_dump_json(by_alias=True, exclude_none=True)) + assert dumped["websiteUrl"] == "https://example.com" + assert ServerCard.model_validate(dumped).website_url == "https://example.com" + + +# -- endpoint_urls and required_variables ----------------------------------------------- + + +def test_endpoint_urls_returns_raw_remote_urls() -> None: + """SDK-defined: `endpoint_urls()` is the host's dedup key, so it returns the raw + (still templated) `remotes[].url` values.""" + card = ServerCard.model_validate( + _minimal_card_data( + remotes=[ + {"type": "streamable-http", "url": "https://{tenant}.example.com/mcp"}, + {"type": "sse", "url": "https://example.com/sse"}, + ] + ) + ) + assert card.endpoint_urls() == frozenset({"https://{tenant}.example.com/mcp", "https://example.com/sse"}) + + +def test_endpoint_urls_is_empty_without_remotes() -> None: + """SDK-defined: a card with no `remotes` has no endpoints to key on.""" + assert ServerCard.model_validate(_minimal_card_data()).endpoint_urls() == frozenset() + + +def test_required_variables_keeps_only_unresolvable_required_inputs() -> None: + """SDK-defined: a variable counts as required to prompt for only when it is + `isRequired` with no `default` and no pre-set `value`, across the URL variables and + each header's nested variables.""" + remote = Remote( + type="streamable-http", + url="https://{tenant}.example.com/{region}/mcp", + variables={ + "tenant": Input(is_required=True), + "region": Input(is_required=True, default="eu"), + "theme": Input(), + }, + headers=[ + KeyValueInput( + name="Authorization", + value="Bearer {token} {suffix}", + variables={"token": Input(is_required=True), "suffix": Input(is_required=True, value="v1")}, + ) + ], + ) + assert remote.required_variables == frozenset({"tenant", "token"}) + + +def test_required_variables_includes_required_headers_without_a_value() -> None: + """SDK-defined: a header that is itself `isRequired` with no pre-set `value` and no + `default` is an input the host must prompt for, keyed by its header name, so a bare + `resolve_remote` demands exactly what `required_variables` listed up front.""" + remote = Remote( + type="streamable-http", + url="https://example.com/mcp", + headers=[ + KeyValueInput(name="X-API-Key", is_required=True), + KeyValueInput(name="X-Region", is_required=True, default="eu"), + KeyValueInput(name="X-Trace"), + ], + ) + assert remote.required_variables == frozenset({"X-API-Key"}) + with pytest.raises(ValueError) as exc_info: + resolve_remote(remote) + assert str(exc_info.value) == snapshot("missing required variables: X-API-Key") + + +def test_required_variables_is_empty_without_declared_variables() -> None: + """SDK-defined: no `variables` and no `headers` means nothing to prompt for.""" + assert Remote(type="streamable-http", url="https://example.com/mcp").required_variables == frozenset() + + +# -- resolve_remote --------------------------------------------------------------------- + + +def _templated_remote() -> Remote: + """The vendored `templated-remote.json` fixture's remote: templated URL, defaulted + tenant, and an Authorization header with a required nested token.""" + card = ServerCard.model_validate_json((FIXTURES / "valid" / "templated-remote.json").read_bytes()) + assert card.remotes is not None + return card.remotes[0] + + +def test_resolve_remote_applies_defaults_and_nested_header_variables() -> None: + """SDK-defined: declared defaults fill the URL template and nested header variables + fill the header value, from the spec's own full example card.""" + resolved = resolve_remote(_templated_remote(), {"token": "abc123"}) + assert resolved.type == "streamable-http" + assert resolved.url == "https://default.example.com/mcp" + assert resolved.headers == {"Authorization": "Bearer abc123"} + + +def test_resolve_remote_caller_values_override_defaults() -> None: + """SDK-defined: a caller-supplied value beats the declared default.""" + resolved = resolve_remote(_templated_remote(), {"token": "abc123", "tenant": "acme"}) + assert resolved.url == "https://acme.example.com/mcp" + + +def test_resolve_remote_names_every_missing_required_variable() -> None: + """SDK-defined: one ValueError lists all missing `isRequired` variables, so a host + can prompt for everything at once.""" + remote = Remote( + type="streamable-http", + url="https://{tenant}.example.com/mcp", + variables={"tenant": Input(is_required=True)}, + headers=[KeyValueInput(name="X-Token", is_required=True)], + ) + with pytest.raises(ValueError) as exc_info: + resolve_remote(remote) + assert str(exc_info.value) == snapshot("missing required variables: X-Token, tenant") + + +def test_resolve_remote_supplies_header_values_by_header_name() -> None: + """SDK-defined: a header with no pre-set `value` is itself an input, keyed by the + header name in the caller's values.""" + remote = Remote(type="streamable-http", url="https://example.com/mcp", headers=[KeyValueInput(name="X-Token")]) + resolved = resolve_remote(remote, {"X-Token": "sekrit"}) + assert resolved.headers == {"X-Token": "sekrit"} + + +def test_resolve_remote_omits_optional_headers_nothing_supplies() -> None: + """SDK-defined: an optional header with no value, default or caller entry simply + does not appear in the resolved headers.""" + remote = Remote(type="streamable-http", url="https://example.com/mcp", headers=[KeyValueInput(name="X-Trace")]) + assert resolve_remote(remote).headers == {} + + +def test_resolve_remote_pre_set_value_beats_caller_value() -> None: + """SDK-defined: a pre-set `value` is not end-user configurable, so a caller entry + for the same name does not override it.""" + remote = Remote( + type="streamable-http", + url="https://{plan}.example.com/mcp", + variables={"plan": Input(value="enterprise")}, + ) + assert resolve_remote(remote, {"plan": "free"}).url == "https://enterprise.example.com/mcp" + + +def test_resolve_remote_rejects_a_value_outside_declared_choices() -> None: + """SDK-defined: when an input declares `choices`, any other value is an error.""" + remote = Remote( + type="streamable-http", + url="https://{region}.example.com/mcp", + variables={"region": Input(choices=["eu", "us"])}, + ) + with pytest.raises(ValueError) as exc_info: + resolve_remote(remote, {"region": "mars"}) + assert str(exc_info.value) == snapshot("values outside declared choices: region='mars' (choices: ['eu', 'us'])") + + +def test_resolve_remote_rejects_a_header_value_outside_declared_choices() -> None: + """SDK-defined: `choices` on the header input itself constrains the header value.""" + remote = Remote( + type="streamable-http", + url="https://example.com/mcp", + headers=[KeyValueInput(name="X-Mode", choices=["fast", "safe"])], + ) + with pytest.raises(ValueError) as exc_info: + resolve_remote(remote, {"X-Mode": "wild"}) + assert str(exc_info.value) == snapshot("values outside declared choices: X-Mode='wild' (choices: ['fast', 'safe'])") + + +def test_resolve_remote_rejects_a_still_templated_result() -> None: + """SDK-defined: a `{placeholder}` that no declared variable and no caller value + resolves is an error, never silently passed through.""" + remote = Remote(type="streamable-http", url="https://{tenant}.example.com/mcp") + with pytest.raises(ValueError) as exc_info: + resolve_remote(remote) + assert str(exc_info.value) == snapshot("unresolved template variables: tenant") + + +def test_resolve_remote_rejects_a_non_http_resolved_url() -> None: + """SDK-defined: template substitution must produce an http(s) URL to connect to.""" + remote = Remote( + type="streamable-http", + url="{scheme}example.com/mcp", + variables={"scheme": Input(default="ftp://")}, + ) + with pytest.raises(ValueError) as exc_info: + resolve_remote(remote) + assert str(exc_info.value) == snapshot("resolved remote URL must be http(s), got 'ftp://example.com/mcp'") + + +def test_resolve_remote_skips_optional_variables_nothing_supplies() -> None: + """SDK-defined: an optional declared variable with no value, default or caller entry + is simply left out of the substitution mapping.""" + remote = Remote(type="streamable-http", url="https://example.com/mcp", variables={"theme": Input()}) + assert resolve_remote(remote).url == "https://example.com/mcp"