Add ETag support to discovery responses

Compute strong SHA-256 ETags for server-card and AI Catalog response bodies, handle matching If-None-Match requests with 304 responses, and cover conditional request behavior in tests.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: SamMorrowDrums

src/mcp/server/experimental/ai_catalog.py

@@ -14,6 +14,8 @@
 
 from __future__ import annotations
 
+import hashlib
+
 from starlette.applications import Starlette
 from starlette.requests import Request
 from starlette.responses import Response
@@ -42,6 +44,37 @@
 }
 
 
+def _strong_etag(body: bytes) -> str:
+    return f'"{hashlib.sha256(body).hexdigest()}"'
+
+
+def _if_none_match_matches(if_none_match: str | None, etag: str) -> bool:
+    if if_none_match is None:
+        return False
+    for candidate in if_none_match.split(","):
+        candidate = candidate.strip()
+        if candidate == "*":
+            return True
+        if candidate.startswith(("W/", "w/")):
+            candidate = candidate[2:].strip()
+        if candidate == etag:
+            return True
+    return False
+
+
+def discovery_response(request: Request, body: bytes, media_type: str) -> Response:
+    etag = _strong_etag(body)
+    if _if_none_match_matches(request.headers.get("if-none-match"), etag):
+        return Response(
+            status_code=304,
+            headers={
+                "ETag": etag,
+                "Cache-Control": DISCOVERY_HEADERS["Cache-Control"],
+            },
+        )
+    return Response(body, media_type=media_type, headers={**DISCOVERY_HEADERS, "ETag": etag})
+
+
 def _air_identifier(card_name: str) -> str:
     """Derive an AI Catalog ``urn:air:`` identifier from a Server Card name.
 
@@ -82,8 +115,8 @@ def ai_catalog_route(catalog: AICatalog, *, path: str = AI_CATALOG_WELL_KNOWN_PA
     """
     body = catalog.model_dump_json(by_alias=True, exclude_none=True).encode()
 
-    async def endpoint(_request: Request) -> Response:
-        return Response(body, media_type=AI_CATALOG_MEDIA_TYPE, headers=DISCOVERY_HEADERS)
+    async def endpoint(request: Request) -> Response:
+        return discovery_response(request, body, AI_CATALOG_MEDIA_TYPE)
 
     return Route(path, endpoint=endpoint, methods=["GET"], name="ai_catalog")
 

src/mcp/server/experimental/server_card.py

@@ -28,7 +28,7 @@
 from starlette.responses import Response
 from starlette.routing import Route
 
-from mcp.server.experimental.ai_catalog import DISCOVERY_HEADERS
+from mcp.server.experimental.ai_catalog import discovery_response
 from mcp.shared.experimental.ai_catalog.types import MCP_SERVER_CARD_MEDIA_TYPE
 from mcp.shared.experimental.server_card.types import (
     Icon,
@@ -112,8 +112,8 @@ def server_card_route(card: ServerCard, *, path: str = "/server-card") -> Route:
     """
     body = card.model_dump_json(by_alias=True, exclude_none=True).encode()
 
-    async def endpoint(_request: Request) -> Response:
-        return Response(body, media_type=MCP_SERVER_CARD_MEDIA_TYPE, headers=DISCOVERY_HEADERS)
+    async def endpoint(request: Request) -> Response:
+        return discovery_response(request, body, MCP_SERVER_CARD_MEDIA_TYPE)
 
     return Route(path, endpoint=endpoint, methods=["GET"], name="mcp_server_card")
 

tests/experimental/ai_catalog/test_server.py

@@ -2,6 +2,8 @@
 
 from __future__ import annotations
 
+import re
+
 import httpx
 import pytest
 from starlette.applications import Starlette
@@ -39,10 +41,16 @@ def test_server_card_entry_falls_back_to_card_name_without_title() -> None:
     assert server_card_entry(make_card(), CARD_URL).display_name == "example/dice"
 
 
-async def _get(app: Starlette, path: str) -> httpx.Response:
+async def _get(app: Starlette, path: str, headers: dict[str, str] | None = None) -> httpx.Response:
+    transport = httpx.ASGITransport(app=app)
+    async with httpx.AsyncClient(transport=transport, base_url="https://dice.example.com") as client:
+        return await client.get(path, headers=headers)
+
+
+async def _head(app: Starlette, path: str) -> httpx.Response:
     transport = httpx.ASGITransport(app=app)
     async with httpx.AsyncClient(transport=transport, base_url="https://dice.example.com") as client:
-        return await client.get(path)
+        return await client.head(path)
 
 
 async def test_ai_catalog_route_serves_catalog_with_discovery_headers() -> None:
@@ -56,8 +64,32 @@ async def test_ai_catalog_route_serves_catalog_with_discovery_headers() -> None:
     assert response.headers["access-control-allow-methods"] == "GET"
     assert response.headers["access-control-allow-headers"] == "Content-Type"
     assert response.headers["cache-control"] == "public, max-age=3600"
+    etag = response.headers["etag"]
+    assert re.fullmatch(r'"[0-9a-f]{64}"', etag)
+    assert (await _get(app, "/.well-known/ai-catalog.json")).headers["etag"] == etag
+    assert (await _head(app, "/.well-known/ai-catalog.json")).headers["etag"] == etag
     assert response.text == catalog.model_dump_json(by_alias=True, exclude_none=True)
 
+    not_modified = await _get(app, "/.well-known/ai-catalog.json", headers={"If-None-Match": etag})
+    assert not_modified.status_code == 304
+    assert not_modified.headers["etag"] == etag
+    assert not_modified.headers["cache-control"] == "public, max-age=3600"
+    assert not_modified.content == b""
+
+    weak_match = await _get(app, "/.well-known/ai-catalog.json", headers={"If-None-Match": f'"not-it", W/{etag}'})
+    assert weak_match.status_code == 304
+    assert weak_match.content == b""
+
+    wildcard = await _get(app, "/.well-known/ai-catalog.json", headers={"If-None-Match": "*"})
+    assert wildcard.status_code == 304
+    assert wildcard.headers["etag"] == etag
+    assert wildcard.content == b""
+
+    non_matching = await _get(app, "/.well-known/ai-catalog.json", headers={"If-None-Match": '"not-it"'})
+    assert non_matching.status_code == 200
+    assert non_matching.headers["etag"] == etag
+    assert non_matching.text == catalog.model_dump_json(by_alias=True, exclude_none=True)
+
 
 async def test_mount_ai_catalog_on_existing_app() -> None:
     app = Starlette()

tests/experimental/server_card/test_server.py

@@ -2,6 +2,8 @@
 
 from __future__ import annotations
 
+import re
+
 import httpx
 import pytest
 from starlette.applications import Starlette
@@ -59,10 +61,16 @@ def test_build_server_card_requires_description() -> None:
         build_server_card(server, name="example/no-desc")
 
 
-async def _get(app: Starlette, path: str) -> httpx.Response:
+async def _get(app: Starlette, path: str, headers: dict[str, str] | None = None) -> httpx.Response:
+    transport = httpx.ASGITransport(app=app)
+    async with httpx.AsyncClient(transport=transport, base_url="https://dice.example.com") as client:
+        return await client.get(path, headers=headers)
+
+
+async def _head(app: Starlette, path: str) -> httpx.Response:
     transport = httpx.ASGITransport(app=app)
     async with httpx.AsyncClient(transport=transport, base_url="https://dice.example.com") as client:
-        return await client.get(path)
+        return await client.head(path)
 
 
 async def test_server_card_route_serves_card_with_discovery_headers() -> None:
@@ -76,9 +84,33 @@ async def test_server_card_route_serves_card_with_discovery_headers() -> None:
     assert response.headers["access-control-allow-methods"] == "GET"
     assert response.headers["access-control-allow-headers"] == "Content-Type"
     assert response.headers["cache-control"] == "public, max-age=3600"
+    etag = response.headers["etag"]
+    assert re.fullmatch(r'"[0-9a-f]{64}"', etag)
+    assert (await _get(app, CARD_PATH)).headers["etag"] == etag
+    assert (await _head(app, CARD_PATH)).headers["etag"] == etag
     assert response.text == card.model_dump_json(by_alias=True, exclude_none=True)
     assert ServerCard.model_validate(response.json()) == card
 
+    not_modified = await _get(app, CARD_PATH, headers={"If-None-Match": etag})
+    assert not_modified.status_code == 304
+    assert not_modified.headers["etag"] == etag
+    assert not_modified.headers["cache-control"] == "public, max-age=3600"
+    assert not_modified.content == b""
+
+    weak_match = await _get(app, CARD_PATH, headers={"If-None-Match": f'"not-it", W/{etag}'})
+    assert weak_match.status_code == 304
+    assert weak_match.content == b""
+
+    wildcard = await _get(app, CARD_PATH, headers={"If-None-Match": "*"})
+    assert wildcard.status_code == 304
+    assert wildcard.headers["etag"] == etag
+    assert wildcard.content == b""
+
+    non_matching = await _get(app, CARD_PATH, headers={"If-None-Match": '"not-it"'})
+    assert non_matching.status_code == 200
+    assert non_matching.headers["etag"] == etag
+    assert non_matching.text == card.model_dump_json(by_alias=True, exclude_none=True)
+
 
 async def test_mount_server_card_on_existing_app_and_client_fetch() -> None:
     card = build_server_card(