Enhance create_cimd_document with confidential client support

- Add jwks_uri and jwks parameters for private_key_jwt clients
- Add client_uri, logo_uri, contacts optional metadata fields
- Add strict URL validation per IETF draft (no fragments, credentials, dot segments)
- Make redirect_uris required (security: no defaults)

Note: private_key_jwt requires SDK server-side support (not yet available)

src/fastmcp/client/auth/__init__.py

@@ -1,4 +1,5 @@
 from .bearer import BearerAuth
+from .cimd import create_cimd_document
 from .oauth import OAuth
 
-__all__ = ["BearerAuth", "OAuth"]
+__all__ = ["BearerAuth", "OAuth", "create_cimd_document"]

src/fastmcp/client/auth/cimd.py

@@ -0,0 +1,153 @@
+"""
+Utility for creating CIMD (Client ID Metadata Documents).
+
+CIMD allows OAuth clients to be identified by a URL pointing to their
+metadata document, eliminating the need for Dynamic Client Registration (DCR).
+
+See:
+- SEP-991: https://github.com/modelcontextprotocol/modelcontextprotocol/issues/991
+- IETF Draft: https://www.ietf.org/archive/id/draft-ietf-oauth-client-id-metadata-document-00.html
+"""
+
+from __future__ import annotations
+
+from typing import Any
+from urllib.parse import urlparse
+
+from pydantic import AnyHttpUrl
+
+__all__ = ["create_cimd_document"]
+
+
+def _validate_cimd_url(url: str) -> None:
+    """Validate URL meets CIMD requirements per IETF draft."""
+    parsed = urlparse(url)
+
+    if parsed.scheme != "https":
+        raise ValueError("CIMD URL must use HTTPS")
+
+    if parsed.path in ("", "/"):
+        raise ValueError("CIMD URL must have a non-root path")
+
+    if parsed.fragment:
+        raise ValueError("CIMD URL must not contain a fragment")
+
+    if parsed.username or parsed.password:
+        raise ValueError("CIMD URL must not contain credentials")
+
+    # Check for dot segments (. or ..) in path
+    path_segments = parsed.path.split("/")
+    if any(seg in (".", "..") for seg in path_segments):
+        raise ValueError("CIMD URL path must not contain dot segments")
+
+
+def create_cimd_document(
+    url: str,
+    *,
+    redirect_uris: list[str],
+    client_name: str = "FastMCP Client",
+    scopes: list[str] | None = None,
+    jwks_uri: str | None = None,
+    jwks: dict[str, Any] | None = None,
+    client_uri: str | None = None,
+    logo_uri: str | None = None,
+    contacts: list[str] | None = None,
+) -> dict[str, Any]:
+    """
+    Create a CIMD (Client ID Metadata Document) for hosting.
+
+    The returned dict should be serialized as JSON and served at the given URL.
+    The URL itself becomes the OAuth client_id.
+
+    For **public clients** (CLI apps, mobile apps), omit jwks_uri/jwks.
+    These clients use PKCE for security but cannot prove client identity.
+
+    For **confidential clients**, provide jwks_uri or jwks containing your
+    public key(s). The client must sign JWTs with the corresponding private
+    key when calling the token endpoint, proving ownership of the client_id.
+
+    Args:
+        url: The HTTPS URL where this document will be hosted.
+             Must use HTTPS, have a non-root path, no fragment, no credentials.
+             This exact URL becomes the client_id.
+        redirect_uris: OAuth callback URIs. These must exactly match where your
+             OAuth client will receive callbacks.
+        client_name: Human-readable name for the client (displayed during consent).
+        scopes: OAuth scopes to request (e.g., ["openid", "profile", "email"]).
+        jwks_uri: URL to JSON Web Key Set for confidential clients.
+             When provided, token_endpoint_auth_method is set to "private_key_jwt".
+        jwks: Inline JSON Web Key Set (alternative to jwks_uri).
+             When provided, token_endpoint_auth_method is set to "private_key_jwt".
+        client_uri: URL to client's homepage (displayed during consent).
+        logo_uri: URL to client's logo image (displayed during consent).
+        contacts: List of contact emails for the client developer.
+
+    Returns:
+        Dict ready to serialize as JSON and host at the URL.
+
+    Raises:
+        ValueError: If URL is invalid or both jwks_uri and jwks are provided.
+
+    Example (public client):
+        >>> doc = create_cimd_document(
+        ...     "https://example.com/.well-known/oauth-client.json",
+        ...     redirect_uris=["http://localhost:8080/callback"],
+        ...     client_name="My CLI App",
+        ... )
+        >>> doc["token_endpoint_auth_method"]
+        'none'
+
+    Example (confidential client):
+        >>> doc = create_cimd_document(
+        ...     "https://example.com/.well-known/oauth-client.json",
+        ...     redirect_uris=["https://example.com/callback"],
+        ...     client_name="My Web App",
+        ...     jwks_uri="https://example.com/.well-known/jwks.json",
+        ... )
+        >>> doc["token_endpoint_auth_method"]
+        'private_key_jwt'
+    """
+    _validate_cimd_url(url)
+
+    if jwks_uri and jwks:
+        raise ValueError("Provide either jwks_uri or jwks, not both")
+
+    # Determine auth method based on whether keys are provided
+    is_confidential = jwks_uri is not None or jwks is not None
+    token_endpoint_auth_method = "private_key_jwt" if is_confidential else "none"
+
+    # Build the document
+    # Using dict directly instead of OAuthClientInformationFull to include jwks fields
+    doc: dict[str, Any] = {
+        "client_id": url,
+        "client_name": client_name,
+        "redirect_uris": redirect_uris,
+        "grant_types": ["authorization_code", "refresh_token"],
+        "response_types": ["code"],
+        "token_endpoint_auth_method": token_endpoint_auth_method,
+    }
+
+    # Add optional fields
+    if scopes:
+        doc["scope"] = " ".join(scopes)
+
+    if jwks_uri:
+        doc["jwks_uri"] = jwks_uri
+
+    if jwks:
+        doc["jwks"] = jwks
+
+    if client_uri:
+        # Validate it's a valid URL
+        AnyHttpUrl(client_uri)
+        doc["client_uri"] = client_uri
+
+    if logo_uri:
+        # Validate it's a valid URL
+        AnyHttpUrl(logo_uri)
+        doc["logo_uri"] = logo_uri
+
+    if contacts:
+        doc["contacts"] = contacts
+
+    return doc

src/fastmcp/client/auth/oauth.py

@@ -149,6 +149,7 @@ class OAuth(OAuthClientProvider):
         additional_client_metadata: dict[str, Any] | None = None,
         callback_port: int | None = None,
         httpx_client_factory: McpHttpClientFactory | None = None,
+        client_metadata_url: str | None = None,
     ):
         """
         Initialize OAuth client provider for an MCP server.
@@ -161,7 +162,12 @@ class OAuth(OAuthClientProvider):
             token_storage: An AsyncKeyValue-compatible token store, tokens are stored in memory if not provided
             additional_client_metadata: Extra fields for OAuthClientMetadata
             callback_port: Fixed port for OAuth callback (default: random available port)
+            client_metadata_url: URL-based client ID per CIMD (SEP-991). When provided
+                and the server supports CIMD, this URL is used as the client_id instead
+                of performing Dynamic Client Registration. The URL must be HTTPS with
+                a non-root path and point to a JSON document containing client metadata.
         """
+        self.client_metadata_url = client_metadata_url
         parsed_url = urlparse(mcp_url)
         server_base_url = f"{parsed_url.scheme}://{parsed_url.netloc}"
 
@@ -215,6 +221,7 @@ class OAuth(OAuthClientProvider):
             storage=self.token_storage_adapter,
             redirect_handler=self.redirect_handler,
             callback_handler=self.callback_handler,
+            client_metadata_url=client_metadata_url,
         )
 
     async def _initialize(self) -> None:

src/fastmcp/server/auth/_cimd.py

@@ -0,0 +1,269 @@
+"""
+Private CIMD (Client ID Metadata Document) implementation for FastMCP.
+
+This module implements server-side CIMD support (SEP-991) ahead of the MCP SDK.
+When/if the SDK adds official server-side CIMD support, we should migrate to
+using their implementation instead of this private module.
+
+All functions and classes in this module are private (prefixed with _) to make
+it easy to swap out the implementation later without breaking public API.
+
+Reference:
+- SEP-991: https://github.com/modelcontextprotocol/modelcontextprotocol/issues/991
+- IETF draft: https://datatracker.ietf.org/doc/draft-ietf-oauth-client-id-metadata-document/
+"""
+
+from __future__ import annotations
+
+import ipaddress
+import socket
+import time
+from typing import Any
+from urllib.parse import urlparse
+
+import httpx
+from mcp.shared.auth import OAuthClientInformationFull
+
+from fastmcp.utilities.logging import get_logger
+
+logger = get_logger(__name__)
+
+
+def _is_cimd_client_id(client_id: str) -> bool:
+    """Check if client_id is a URL suitable for CIMD lookup.
+
+    Per the spec, CIMD URLs must:
+    - Use HTTPS scheme
+    - Have a non-root path component
+    """
+    if not isinstance(client_id, str):
+        return False
+    if not client_id.startswith("https://"):
+        return False
+    try:
+        parsed = urlparse(client_id)
+        return parsed.scheme == "https" and parsed.path not in ("", "/")
+    except Exception:
+        return False
+
+
+def _is_private_ip(ip_str: str) -> bool:
+    """Check if an IP address is in a private/reserved range.
+
+    This is used for SSRF protection to prevent fetching metadata from
+    internal network addresses.
+    """
+    try:
+        ip = ipaddress.ip_address(ip_str)
+
+        # Check for private, loopback, link-local, etc.
+        if ip.is_private:
+            return True
+        if ip.is_loopback:
+            return True
+        if ip.is_link_local:
+            return True
+        if ip.is_multicast:
+            return True
+        if ip.is_reserved:
+            return True
+
+        # For IPv4, also check for special ranges
+        if isinstance(ip, ipaddress.IPv4Address):
+            # 0.0.0.0/8 - "This" network
+            if ip_str.startswith("0."):
+                return True
+
+        return False
+    except ValueError:
+        # If we can't parse the IP, treat it as suspicious
+        return True
+
+
+def _validate_cimd_url(url: str) -> None:
+    """Validate URL for CIMD fetch, including SSRF protection.
+
+    Raises:
+        ValueError: If URL is invalid or resolves to private IP
+    """
+    parsed = urlparse(url)
+
+    if parsed.scheme != "https":
+        raise ValueError("CIMD URL must use HTTPS")
+
+    if parsed.path in ("", "/"):
+        raise ValueError("CIMD URL must have a non-root path")
+
+    hostname = parsed.hostname
+    if not hostname:
+        raise ValueError("CIMD URL must have a valid hostname")
+
+    # DNS resolution check for SSRF protection
+    try:
+        # Get all IP addresses for the hostname
+        _, _, ipaddrlist = socket.gethostbyname_ex(hostname)
+        for ip in ipaddrlist:
+            if _is_private_ip(ip):
+                raise ValueError(f"CIMD URL resolves to private IP address: {ip}")
+    except socket.gaierror as e:
+        raise ValueError(f"Failed to resolve CIMD URL hostname: {e}") from e
+
+
+async def _fetch_client_metadata(
+    client_id: str,
+    *,
+    timeout: float = 10.0,
+    max_size: int = 1_048_576,  # 1MB
+) -> dict[str, Any]:
+    """Fetch and parse client metadata document from URL.
+
+    Args:
+        client_id: The HTTPS URL to fetch metadata from
+        timeout: Request timeout in seconds
+        max_size: Maximum response size in bytes
+
+    Returns:
+        Parsed JSON metadata as a dictionary
+
+    Raises:
+        ValueError: If URL validation fails or response is too large
+        httpx.HTTPError: If the HTTP request fails
+    """
+    _validate_cimd_url(client_id)
+
+    async with httpx.AsyncClient() as client:
+        response = await client.get(
+            client_id,
+            timeout=timeout,
+            headers={"Accept": "application/json"},
+            follow_redirects=True,
+        )
+        response.raise_for_status()
+
+        # Check content length header before reading body
+        content_length = response.headers.get("content-length")
+        if content_length and int(content_length) > max_size:
+            raise ValueError(
+                f"CIMD response too large: {content_length} bytes (max: {max_size})"
+            )
+
+        # Read response with size check
+        content = response.content
+        if len(content) > max_size:
+            raise ValueError(
+                f"CIMD response too large: {len(content)} bytes (max: {max_size})"
+            )
+
+        return response.json()
+
+
+def _create_client_from_metadata(
+    client_id: str,
+    metadata: dict[str, Any],
+) -> OAuthClientInformationFull:
+    """Convert fetched metadata to OAuthClientInformationFull.
+
+    Args:
+        client_id: The URL used to fetch the metadata (must match client_id in doc)
+        metadata: The parsed metadata document
+
+    Returns:
+        OAuthClientInformationFull instance
+
+    Raises:
+        ValueError: If client_id in metadata doesn't match URL
+    """
+    # Validate that client_id in the document matches the URL
+    doc_client_id = metadata.get("client_id")
+    if doc_client_id != client_id:
+        raise ValueError(
+            f"client_id in metadata ({doc_client_id}) doesn't match URL ({client_id})"
+        )
+
+    # Ensure token_endpoint_auth_method defaults to "none" for CIMD clients
+    # (they don't have client secrets)
+    if "token_endpoint_auth_method" not in metadata:
+        metadata = {**metadata, "token_endpoint_auth_method": "none"}
+
+    return OAuthClientInformationFull.model_validate(metadata)
+
+
+class _CIMDCache:
+    """Simple in-memory cache for CIMD metadata with TTL support.
+
+    Per the spec, servers SHOULD cache metadata respecting HTTP headers,
+    with a maximum of 24 hours.
+    """
+
+    def __init__(
+        self,
+        default_ttl: int = 3600,  # 1 hour default
+        max_ttl: int = 86400,  # 24 hours max per spec
+    ):
+        self._cache: dict[str, tuple[OAuthClientInformationFull, float]] = {}
+        self._default_ttl = default_ttl
+        self._max_ttl = max_ttl
+
+    def get(self, client_id: str) -> OAuthClientInformationFull | None:
+        """Get cached client info if present and not expired."""
+        if client_id not in self._cache:
+            return None
+
+        client_info, expires_at = self._cache[client_id]
+        if time.time() > expires_at:
+            del self._cache[client_id]
+            return None
+
+        return client_info
+
+    def set(
+        self,
+        client_id: str,
+        client_info: OAuthClientInformationFull,
+        ttl: int | None = None,
+    ) -> None:
+        """Cache client info with TTL (capped at max_ttl)."""
+        effective_ttl = min(ttl or self._default_ttl, self._max_ttl)
+        expires_at = time.time() + effective_ttl
+        self._cache[client_id] = (client_info, expires_at)
+
+    def clear(self) -> None:
+        """Clear all cached entries."""
+        self._cache.clear()
+
+
+async def _get_cimd_client(
+    client_id: str,
+    cache: _CIMDCache | None = None,
+) -> OAuthClientInformationFull:
+    """Main entry point - fetch and validate CIMD client.
+
+    Args:
+        client_id: The HTTPS URL to use as client_id
+        cache: Optional cache instance for storing results
+
+    Returns:
+        OAuthClientInformationFull for the client
+
+    Raises:
+        ValueError: If validation fails
+        httpx.HTTPError: If fetch fails
+    """
+    # Check cache first
+    if cache:
+        cached = cache.get(client_id)
+        if cached is not None:
+            logger.debug(f"CIMD cache hit for {client_id}")
+            return cached
+
+    logger.debug(f"Fetching CIMD metadata from {client_id}")
+
+    # Fetch and validate
+    metadata = await _fetch_client_metadata(client_id)
+    client_info = _create_client_from_metadata(client_id, metadata)
+
+    # Cache result
+    if cache:
+        cache.set(client_id, client_info)
+
+    return client_info

src/fastmcp/server/auth/auth.py

@@ -4,6 +4,7 @@ import json
 from typing import Any, cast
 from urllib.parse import urlparse
 
+from mcp.server.auth.handlers.metadata import MetadataHandler
 from mcp.server.auth.handlers.token import TokenErrorResponse
 from mcp.server.auth.handlers.token import TokenHandler as _SDKTokenHandler
 from mcp.server.auth.json_response import PydanticJSONResponse
@@ -30,11 +31,13 @@ from mcp.server.auth.settings import (
     ClientRegistrationOptions,
     RevocationOptions,
 )
+from mcp.shared.auth import OAuthClientInformationFull, OAuthMetadata
 from pydantic import AnyHttpUrl, Field
 from starlette.middleware import Middleware
 from starlette.middleware.authentication import AuthenticationMiddleware
 from starlette.routing import Route
 
+from fastmcp.server.auth._cimd import _CIMDCache, _get_cimd_client, _is_cimd_client_id
 from fastmcp.utilities.logging import get_logger
 
 logger = get_logger(__name__)
@@ -335,6 +338,7 @@ class OAuthProvider(
         client_registration_options: ClientRegistrationOptions | None = None,
         revocation_options: RevocationOptions | None = None,
         required_scopes: list[str] | None = None,
+        cimd_enabled: bool = True,
     ):
         """
         Initialize the OAuth provider.
@@ -346,6 +350,9 @@ class OAuthProvider(
             client_registration_options: The client registration options.
             revocation_options: The revocation options.
             required_scopes: Scopes that are required for all requests.
+            cimd_enabled: Whether to enable Client ID Metadata Document (CIMD)
+                support per SEP-991. When enabled, URL-based client_ids will be
+                fetched and validated automatically. Defaults to True.
         """
 
         super().__init__(base_url=base_url, required_scopes=required_scopes)
@@ -379,6 +386,69 @@ class OAuthProvider(
         self.client_registration_options = client_registration_options
         self.revocation_options = revocation_options
 
+        # CIMD (Client ID Metadata Document) support - SEP-991
+        self._cimd_enabled = cimd_enabled
+        self._cimd_cache = _CIMDCache() if cimd_enabled else None
+
+    async def _lookup_cimd_client(
+        self, client_id: str
+    ) -> OAuthClientInformationFull | None:
+        """Try to look up client via CIMD if it's a URL-based client_id.
+
+        Subclasses should call this in their get_client() implementation
+        before falling back to their normal client lookup.
+
+        Args:
+            client_id: The client ID to look up
+
+        Returns:
+            OAuthClientInformationFull if found via CIMD, None otherwise
+        """
+        if not self._cimd_enabled:
+            return None
+        if not _is_cimd_client_id(client_id):
+            return None
+        try:
+            return await _get_cimd_client(client_id, cache=self._cimd_cache)
+        except Exception as e:
+            logger.debug(f"CIMD lookup failed for {client_id}: {e}")
+            return None
+
+    def _build_cimd_metadata(self) -> OAuthMetadata:
+        """Build OAuth metadata with CIMD support flag enabled.
+
+        This creates metadata identical to what the SDK builds, but with
+        client_id_metadata_document_supported set to True.
+        """
+        from mcp.server.auth.routes import build_metadata
+
+        assert self.base_url is not None
+
+        # Build base metadata using SDK function
+        metadata = build_metadata(
+            issuer_url=self.base_url,
+            service_documentation_url=self.service_documentation_url,
+            client_registration_options=self.client_registration_options
+            or ClientRegistrationOptions(),
+            revocation_options=self.revocation_options or RevocationOptions(),
+        )
+
+        # Add CIMD support flag
+        metadata.client_id_metadata_document_supported = True
+
+        return metadata
+
+    def _create_cimd_metadata_route(self) -> Route:
+        """Create a metadata route that advertises CIMD support."""
+        metadata = self._build_cimd_metadata()
+        metadata_handler = MetadataHandler(metadata=metadata)
+
+        return Route(
+            path="/.well-known/oauth-authorization-server",
+            endpoint=cors_middleware(metadata_handler.handle, ["GET", "OPTIONS"]),
+            methods=["GET", "OPTIONS"],
+        )
+
     async def verify_token(self, token: str) -> AccessToken | None:
         """
         Verify a bearer token and return access info if valid.
@@ -425,8 +495,9 @@ class OAuthProvider(
             revocation_options=self.revocation_options,
         )
 
-        # Replace the token endpoint with our custom handler that returns
+        # Replace certain endpoints with our custom handlers:
-        # proper OAuth 2.1 error codes (invalid_client instead of unauthorized_client)
+        # - Token endpoint: OAuth 2.1 compliant error codes
+        # - Metadata endpoint: Add CIMD support flag when enabled
         oauth_routes: list[Route] = []
         for route in sdk_routes:
             if (
@@ -448,6 +519,13 @@ class OAuthProvider(
                         methods=["POST", "OPTIONS"],
                     )
                 )
+            elif (
+                isinstance(route, Route)
+                and route.path == "/.well-known/oauth-authorization-server"
+                and self._cimd_enabled
+            ):
+                # Replace metadata endpoint to advertise CIMD support
+                oauth_routes.append(self._create_cimd_metadata_route())
             else:
                 oauth_routes.append(route)
 

src/fastmcp/server/auth/oauth_proxy.py

@@ -911,8 +911,13 @@ class OAuthProxy(OAuthProvider):
         provided to the DCR client during registration, not the upstream client ID.
 
         For unregistered clients, returns None (which will raise an error in the SDK).
+        CIMD (URL-based) client_ids are also supported if cimd_enabled is True.
         """
-        # Load from storage
+        # Try CIMD lookup first for URL-based client_ids
+        if cimd_client := await self._lookup_cimd_client(client_id):
+            return cimd_client
+
+        # Load from storage (DCR-registered clients)
         if not (client := await self._client_store.get(key=client_id)):
             return None
 

src/fastmcp/server/auth/providers/in_memory.py

@@ -63,6 +63,10 @@ class InMemoryOAuthProvider(OAuthProvider):
         ] = {}  # refresh_token_str -> access_token_str
 
     async def get_client(self, client_id: str) -> OAuthClientInformationFull | None:
+        # Try CIMD lookup first for URL-based client_ids
+        if cimd_client := await self._lookup_cimd_client(client_id):
+            return cimd_client
+        # Fall back to registered clients
         return self.clients.get(client_id)
 
     async def register_client(self, client_info: OAuthClientInformationFull) -> None:

tests/client/auth/test_cimd.py

@@ -0,0 +1,237 @@
+"""Tests for CIMD document creation utility."""
+
+import pytest
+
+from fastmcp.client.auth import create_cimd_document
+
+
+class TestCreateCimdDocument:
+    """Tests for create_cimd_document utility."""
+
+    def test_basic_public_client(self):
+        """Public client with minimal required fields."""
+        doc = create_cimd_document(
+            "https://example.com/oauth/client.json",
+            redirect_uris=["http://localhost:8080/callback"],
+        )
+
+        assert doc["client_id"] == "https://example.com/oauth/client.json"
+        assert doc["client_name"] == "FastMCP Client"
+        assert doc["redirect_uris"] == ["http://localhost:8080/callback"]
+        assert doc["grant_types"] == ["authorization_code", "refresh_token"]
+        assert doc["response_types"] == ["code"]
+        assert doc["token_endpoint_auth_method"] == "none"
+
+    def test_custom_client_name(self):
+        doc = create_cimd_document(
+            "https://example.com/oauth/client.json",
+            redirect_uris=["http://localhost:8080/callback"],
+            client_name="My Custom Client",
+        )
+
+        assert doc["client_name"] == "My Custom Client"
+
+    def test_multiple_redirect_uris(self):
+        doc = create_cimd_document(
+            "https://example.com/oauth/client.json",
+            redirect_uris=[
+                "http://localhost:3000/callback",
+                "http://localhost:8080/callback",
+            ],
+        )
+
+        assert doc["redirect_uris"] == [
+            "http://localhost:3000/callback",
+            "http://localhost:8080/callback",
+        ]
+
+    def test_scopes(self):
+        doc = create_cimd_document(
+            "https://example.com/oauth/client.json",
+            redirect_uris=["http://localhost:8080/callback"],
+            scopes=["openid", "profile", "email"],
+        )
+
+        assert doc["scope"] == "openid profile email"
+
+    def test_no_scopes_excludes_field(self):
+        doc = create_cimd_document(
+            "https://example.com/oauth/client.json",
+            redirect_uris=["http://localhost:8080/callback"],
+        )
+
+        assert "scope" not in doc
+
+
+class TestConfidentialClients:
+    """Tests for confidential client support with private_key_jwt."""
+
+    def test_jwks_uri_sets_private_key_jwt(self):
+        """Providing jwks_uri makes it a confidential client."""
+        doc = create_cimd_document(
+            "https://example.com/oauth/client.json",
+            redirect_uris=["https://example.com/callback"],
+            jwks_uri="https://example.com/.well-known/jwks.json",
+        )
+
+        assert doc["token_endpoint_auth_method"] == "private_key_jwt"
+        assert doc["jwks_uri"] == "https://example.com/.well-known/jwks.json"
+        assert "jwks" not in doc
+
+    def test_inline_jwks_sets_private_key_jwt(self):
+        """Providing inline jwks makes it a confidential client."""
+        jwks = {
+            "keys": [
+                {
+                    "kty": "RSA",
+                    "use": "sig",
+                    "kid": "test-key-1",
+                    "n": "0vx7agoebGcQSuuPiLJXZptN9...",
+                    "e": "AQAB",
+                }
+            ]
+        }
+        doc = create_cimd_document(
+            "https://example.com/oauth/client.json",
+            redirect_uris=["https://example.com/callback"],
+            jwks=jwks,
+        )
+
+        assert doc["token_endpoint_auth_method"] == "private_key_jwt"
+        assert doc["jwks"] == jwks
+        assert "jwks_uri" not in doc
+
+    def test_cannot_provide_both_jwks_uri_and_jwks(self):
+        """Cannot provide both jwks_uri and inline jwks."""
+        with pytest.raises(ValueError, match="either jwks_uri or jwks"):
+            create_cimd_document(
+                "https://example.com/oauth/client.json",
+                redirect_uris=["https://example.com/callback"],
+                jwks_uri="https://example.com/.well-known/jwks.json",
+                jwks={"keys": []},
+            )
+
+
+class TestOptionalMetadataFields:
+    """Tests for optional metadata fields."""
+
+    def test_client_uri(self):
+        doc = create_cimd_document(
+            "https://example.com/oauth/client.json",
+            redirect_uris=["http://localhost:8080/callback"],
+            client_uri="https://example.com",
+        )
+
+        assert doc["client_uri"] == "https://example.com"
+
+    def test_logo_uri(self):
+        doc = create_cimd_document(
+            "https://example.com/oauth/client.json",
+            redirect_uris=["http://localhost:8080/callback"],
+            logo_uri="https://example.com/logo.png",
+        )
+
+        assert doc["logo_uri"] == "https://example.com/logo.png"
+
+    def test_contacts(self):
+        doc = create_cimd_document(
+            "https://example.com/oauth/client.json",
+            redirect_uris=["http://localhost:8080/callback"],
+            contacts=["admin@example.com", "security@example.com"],
+        )
+
+        assert doc["contacts"] == ["admin@example.com", "security@example.com"]
+
+    def test_invalid_client_uri_rejected(self):
+        with pytest.raises(Exception):  # Pydantic validation error
+            create_cimd_document(
+                "https://example.com/oauth/client.json",
+                redirect_uris=["http://localhost:8080/callback"],
+                client_uri="not-a-valid-url",
+            )
+
+    def test_invalid_logo_uri_rejected(self):
+        with pytest.raises(Exception):  # Pydantic validation error
+            create_cimd_document(
+                "https://example.com/oauth/client.json",
+                redirect_uris=["http://localhost:8080/callback"],
+                logo_uri="not-a-valid-url",
+            )
+
+
+class TestUrlValidation:
+    """Tests for CIMD URL validation per IETF draft."""
+
+    def test_rejects_http_url(self):
+        with pytest.raises(ValueError, match="must use HTTPS"):
+            create_cimd_document(
+                "http://example.com/oauth/client.json",
+                redirect_uris=["http://localhost:8080/callback"],
+            )
+
+    def test_rejects_root_path(self):
+        with pytest.raises(ValueError, match="non-root path"):
+            create_cimd_document(
+                "https://example.com/",
+                redirect_uris=["http://localhost:8080/callback"],
+            )
+
+    def test_rejects_no_path(self):
+        with pytest.raises(ValueError, match="non-root path"):
+            create_cimd_document(
+                "https://example.com",
+                redirect_uris=["http://localhost:8080/callback"],
+            )
+
+    def test_rejects_fragment(self):
+        with pytest.raises(ValueError, match="fragment"):
+            create_cimd_document(
+                "https://example.com/client.json#section",
+                redirect_uris=["http://localhost:8080/callback"],
+            )
+
+    def test_rejects_credentials(self):
+        with pytest.raises(ValueError, match="credentials"):
+            create_cimd_document(
+                "https://user:pass@example.com/client.json",
+                redirect_uris=["http://localhost:8080/callback"],
+            )
+
+    def test_rejects_dot_segments(self):
+        with pytest.raises(ValueError, match="dot segments"):
+            create_cimd_document(
+                "https://example.com/../client.json",
+                redirect_uris=["http://localhost:8080/callback"],
+            )
+
+    def test_rejects_single_dot_segment(self):
+        with pytest.raises(ValueError, match="dot segments"):
+            create_cimd_document(
+                "https://example.com/./client.json",
+                redirect_uris=["http://localhost:8080/callback"],
+            )
+
+    def test_well_known_path_allowed(self):
+        doc = create_cimd_document(
+            "https://example.com/.well-known/oauth-client.json",
+            redirect_uris=["http://localhost:8080/callback"],
+        )
+
+        assert doc["client_id"] == "https://example.com/.well-known/oauth-client.json"
+
+    def test_query_string_allowed(self):
+        """Query strings are discouraged but permitted."""
+        doc = create_cimd_document(
+            "https://example.com/client.json?version=1",
+            redirect_uris=["http://localhost:8080/callback"],
+        )
+
+        assert doc["client_id"] == "https://example.com/client.json?version=1"
+
+    def test_port_allowed(self):
+        doc = create_cimd_document(
+            "https://example.com:8443/client.json",
+            redirect_uris=["http://localhost:8080/callback"],
+        )
+
+        assert doc["client_id"] == "https://example.com:8443/client.json"

tests/server/auth/test_cimd.py

@@ -0,0 +1,461 @@
+"""Tests for CIMD (Client ID Metadata Document) support per SEP-991."""
+
+from __future__ import annotations
+
+import time
+from unittest.mock import patch
+
+import pytest
+from mcp.shared.auth import OAuthClientInformationFull
+from pydantic import AnyUrl
+from pytest_httpx import HTTPXMock
+
+from fastmcp.server.auth._cimd import (
+    _CIMDCache,
+    _create_client_from_metadata,
+    _fetch_client_metadata,
+    _get_cimd_client,
+    _is_cimd_client_id,
+    _is_private_ip,
+    _validate_cimd_url,
+)
+from fastmcp.server.auth.providers.in_memory import InMemoryOAuthProvider
+
+
+class TestIsCimdClientId:
+    """Test _is_cimd_client_id function."""
+
+    def test_valid_https_with_path(self):
+        assert _is_cimd_client_id("https://example.com/oauth/client.json") is True
+        assert _is_cimd_client_id("https://app.example.com/metadata") is True
+        assert _is_cimd_client_id("https://example.com/a") is True
+
+    def test_rejects_http(self):
+        assert _is_cimd_client_id("http://example.com/oauth/client.json") is False
+
+    def test_rejects_root_path(self):
+        assert _is_cimd_client_id("https://example.com") is False
+        assert _is_cimd_client_id("https://example.com/") is False
+
+    def test_rejects_non_url(self):
+        assert _is_cimd_client_id("not-a-url") is False
+        assert _is_cimd_client_id("client-id-123") is False
+        assert _is_cimd_client_id("") is False
+
+    def test_rejects_non_string(self):
+        assert _is_cimd_client_id(123) is False  # type: ignore[arg-type]
+        assert _is_cimd_client_id(None) is False  # type: ignore[arg-type]
+
+
+class TestIsPrivateIp:
+    """Test _is_private_ip function."""
+
+    def test_private_ipv4_10_x(self):
+        assert _is_private_ip("10.0.0.1") is True
+        assert _is_private_ip("10.255.255.255") is True
+
+    def test_private_ipv4_172_16_x(self):
+        assert _is_private_ip("172.16.0.1") is True
+        assert _is_private_ip("172.31.255.255") is True
+        # 172.15.x and 172.32.x are NOT private
+        assert _is_private_ip("172.15.0.1") is False
+        assert _is_private_ip("172.32.0.1") is False
+
+    def test_private_ipv4_192_168_x(self):
+        assert _is_private_ip("192.168.0.1") is True
+        assert _is_private_ip("192.168.255.255") is True
+
+    def test_localhost(self):
+        assert _is_private_ip("127.0.0.1") is True
+        assert _is_private_ip("127.0.0.2") is True
+
+    def test_link_local(self):
+        assert _is_private_ip("169.254.0.1") is True
+        assert _is_private_ip("169.254.255.255") is True
+
+    def test_public_ipv4(self):
+        assert _is_private_ip("8.8.8.8") is False
+        assert _is_private_ip("1.1.1.1") is False
+        assert _is_private_ip("93.184.216.34") is False  # example.com
+
+    def test_ipv6_loopback(self):
+        assert _is_private_ip("::1") is True
+
+    def test_invalid_ip(self):
+        # Invalid IPs are treated as suspicious (returns True)
+        assert _is_private_ip("not-an-ip") is True
+
+
+class TestValidateCimdUrl:
+    """Test _validate_cimd_url function."""
+
+    def test_requires_https(self):
+        with pytest.raises(ValueError, match="must use HTTPS"):
+            _validate_cimd_url("http://example.com/metadata")
+
+    def test_requires_non_root_path(self):
+        with pytest.raises(ValueError, match="non-root path"):
+            _validate_cimd_url("https://example.com")
+        with pytest.raises(ValueError, match="non-root path"):
+            _validate_cimd_url("https://example.com/")
+
+    def test_rejects_private_ip(self):
+        # Mock DNS resolution to return private IP
+        with patch("socket.gethostbyname_ex") as mock_dns:
+            mock_dns.return_value = ("example.com", [], ["10.0.0.1"])
+            with pytest.raises(ValueError, match="private IP"):
+                _validate_cimd_url("https://internal.example.com/metadata")
+
+    def test_accepts_public_ip(self):
+        # Mock DNS resolution to return public IP
+        with patch("socket.gethostbyname_ex") as mock_dns:
+            mock_dns.return_value = ("example.com", [], ["93.184.216.34"])
+            # Should not raise
+            _validate_cimd_url("https://example.com/metadata")
+
+
+class TestFetchClientMetadata:
+    """Test _fetch_client_metadata function."""
+
+    async def test_fetch_success(self, httpx_mock: HTTPXMock):
+        client_id = "https://client.example.com/metadata.json"
+        metadata = {
+            "client_id": client_id,
+            "client_name": "Test Client",
+            "redirect_uris": ["http://localhost:3000/callback"],
+        }
+
+        # Mock DNS to return public IP
+        with patch("socket.gethostbyname_ex") as mock_dns:
+            mock_dns.return_value = ("client.example.com", [], ["93.184.216.34"])
+            httpx_mock.add_response(url=client_id, json=metadata)
+
+            result = await _fetch_client_metadata(client_id)
+
+        assert result == metadata
+
+    async def test_fetch_timeout(self, httpx_mock: HTTPXMock):
+        import httpx
+
+        client_id = "https://slow.example.com/metadata.json"
+
+        with patch("socket.gethostbyname_ex") as mock_dns:
+            mock_dns.return_value = ("slow.example.com", [], ["93.184.216.34"])
+            httpx_mock.add_exception(httpx.TimeoutException("Connection timed out"))
+
+            with pytest.raises(httpx.TimeoutException):
+                await _fetch_client_metadata(client_id, timeout=0.1)
+
+    async def test_fetch_size_limit_via_header(self, httpx_mock: HTTPXMock):
+        client_id = "https://large.example.com/metadata.json"
+
+        with patch("socket.gethostbyname_ex") as mock_dns:
+            mock_dns.return_value = ("large.example.com", [], ["93.184.216.34"])
+            # Return a response with content-length exceeding limit
+            httpx_mock.add_response(
+                url=client_id,
+                json={"client_id": client_id},
+                headers={"content-length": "2000000"},  # 2MB
+            )
+
+            with pytest.raises(ValueError, match="too large"):
+                await _fetch_client_metadata(client_id, max_size=1_048_576)
+
+    async def test_fetch_invalid_json(self, httpx_mock: HTTPXMock):
+        client_id = "https://bad.example.com/metadata.json"
+
+        with patch("socket.gethostbyname_ex") as mock_dns:
+            mock_dns.return_value = ("bad.example.com", [], ["93.184.216.34"])
+            httpx_mock.add_response(url=client_id, text="not json {{{")
+
+            with pytest.raises(Exception):  # JSONDecodeError
+                await _fetch_client_metadata(client_id)
+
+
+class TestCreateClientFromMetadata:
+    """Test _create_client_from_metadata function."""
+
+    def test_creates_client_from_valid_metadata(self):
+        client_id = "https://client.example.com/metadata.json"
+        metadata = {
+            "client_id": client_id,
+            "client_name": "Test Client",
+            "redirect_uris": ["http://localhost:3000/callback"],
+            "token_endpoint_auth_method": "none",
+        }
+
+        result = _create_client_from_metadata(client_id, metadata)
+
+        assert isinstance(result, OAuthClientInformationFull)
+        assert result.client_id == client_id
+        assert result.client_name == "Test Client"
+        assert result.token_endpoint_auth_method == "none"
+
+    def test_defaults_token_endpoint_auth_method_to_none(self):
+        client_id = "https://client.example.com/metadata.json"
+        metadata = {
+            "client_id": client_id,
+            "client_name": "Test Client",
+            "redirect_uris": ["http://localhost:3000/callback"],
+            # No token_endpoint_auth_method specified
+        }
+
+        result = _create_client_from_metadata(client_id, metadata)
+        assert result.token_endpoint_auth_method == "none"
+
+    def test_rejects_mismatched_client_id(self):
+        client_id = "https://client.example.com/metadata.json"
+        metadata = {
+            "client_id": "https://other.example.com/different.json",
+            "client_name": "Test Client",
+            "redirect_uris": ["http://localhost:3000/callback"],
+        }
+
+        with pytest.raises(ValueError, match="doesn't match"):
+            _create_client_from_metadata(client_id, metadata)
+
+
+class TestCIMDCache:
+    """Test _CIMDCache class."""
+
+    def test_cache_hit(self):
+        cache = _CIMDCache()
+        client_id = "https://client.example.com/metadata.json"
+        client_info = OAuthClientInformationFull(
+            client_id=client_id,
+            redirect_uris=[AnyUrl("http://localhost:3000/callback")],
+        )
+
+        cache.set(client_id, client_info)
+        result = cache.get(client_id)
+
+        assert result is not None
+        assert result.client_id == client_id
+
+    def test_cache_miss(self):
+        cache = _CIMDCache()
+        result = cache.get("https://unknown.example.com/metadata.json")
+        assert result is None
+
+    def test_cache_expiry(self):
+        cache = _CIMDCache(default_ttl=1)  # 1 second TTL
+        client_id = "https://client.example.com/metadata.json"
+        client_info = OAuthClientInformationFull(
+            client_id=client_id,
+            redirect_uris=[AnyUrl("http://localhost:3000/callback")],
+        )
+
+        cache.set(client_id, client_info, ttl=1)
+
+        # Should be in cache
+        assert cache.get(client_id) is not None
+
+        # Wait for expiry
+        time.sleep(1.1)
+
+        # Should be expired
+        assert cache.get(client_id) is None
+
+    def test_cache_respects_max_ttl(self):
+        cache = _CIMDCache(default_ttl=3600, max_ttl=2)  # 2 second max
+        client_id = "https://client.example.com/metadata.json"
+        client_info = OAuthClientInformationFull(
+            client_id=client_id,
+            redirect_uris=[AnyUrl("http://localhost:3000/callback")],
+        )
+
+        # Try to set with 1 hour TTL, should be capped to 2 seconds
+        cache.set(client_id, client_info, ttl=3600)
+
+        # Should be in cache
+        assert cache.get(client_id) is not None
+
+        # Wait longer than max_ttl
+        time.sleep(2.1)
+
+        # Should be expired due to max_ttl cap
+        assert cache.get(client_id) is None
+
+    def test_cache_clear(self):
+        cache = _CIMDCache()
+        client_id = "https://client.example.com/metadata.json"
+        client_info = OAuthClientInformationFull(
+            client_id=client_id,
+            redirect_uris=[AnyUrl("http://localhost:3000/callback")],
+        )
+
+        cache.set(client_id, client_info)
+        assert cache.get(client_id) is not None
+
+        cache.clear()
+        assert cache.get(client_id) is None
+
+
+class TestGetCimdClient:
+    """Test _get_cimd_client function."""
+
+    async def test_fetches_and_validates(self, httpx_mock: HTTPXMock):
+        client_id = "https://client.example.com/metadata.json"
+        metadata = {
+            "client_id": client_id,
+            "client_name": "Test Client",
+            "redirect_uris": ["http://localhost:3000/callback"],
+        }
+
+        with patch("socket.gethostbyname_ex") as mock_dns:
+            mock_dns.return_value = ("client.example.com", [], ["93.184.216.34"])
+            httpx_mock.add_response(url=client_id, json=metadata)
+
+            result = await _get_cimd_client(client_id)
+
+        assert isinstance(result, OAuthClientInformationFull)
+        assert result.client_id == client_id
+
+    async def test_uses_cache(self, httpx_mock: HTTPXMock):
+        client_id = "https://client.example.com/metadata.json"
+        metadata = {
+            "client_id": client_id,
+            "client_name": "Test Client",
+            "redirect_uris": ["http://localhost:3000/callback"],
+        }
+
+        cache = _CIMDCache()
+
+        with patch("socket.gethostbyname_ex") as mock_dns:
+            mock_dns.return_value = ("client.example.com", [], ["93.184.216.34"])
+            httpx_mock.add_response(url=client_id, json=metadata)
+
+            # First call - should fetch
+            result1 = await _get_cimd_client(client_id, cache=cache)
+
+            # Second call - should use cache (no HTTP request)
+            result2 = await _get_cimd_client(client_id, cache=cache)
+
+        assert result1.client_id == client_id
+        assert result2.client_id == client_id
+
+        # Verify only one HTTP request was made
+        assert len(httpx_mock.get_requests()) == 1
+
+
+class TestOAuthProviderCIMD:
+    """Test CIMD integration in OAuthProvider base class."""
+
+    async def test_lookup_cimd_client_returns_none_for_non_url(self):
+        """CIMD lookup should return None for non-URL client_ids."""
+        provider = InMemoryOAuthProvider(base_url="http://localhost:8000")
+
+        result = await provider._lookup_cimd_client("regular-client-id")
+        assert result is None
+
+    async def test_lookup_cimd_client_returns_none_when_disabled(self):
+        """CIMD lookup should return None when cimd_enabled=False."""
+        provider = InMemoryOAuthProvider(
+            base_url="http://localhost:8000",
+        )
+        provider._cimd_enabled = False
+
+        result = await provider._lookup_cimd_client(
+            "https://client.example.com/metadata.json"
+        )
+        assert result is None
+
+    async def test_lookup_cimd_client_success(self, httpx_mock: HTTPXMock):
+        """CIMD lookup should successfully fetch and return client info."""
+        provider = InMemoryOAuthProvider(base_url="http://localhost:8000")
+
+        client_id = "https://client.example.com/metadata.json"
+        metadata = {
+            "client_id": client_id,
+            "client_name": "Test Client",
+            "redirect_uris": ["http://localhost:3000/callback"],
+        }
+
+        with patch("socket.gethostbyname_ex") as mock_dns:
+            mock_dns.return_value = ("client.example.com", [], ["93.184.216.34"])
+            httpx_mock.add_response(url=client_id, json=metadata)
+
+            result = await provider._lookup_cimd_client(client_id)
+
+        assert result is not None
+        assert result.client_id == client_id
+
+    async def test_lookup_cimd_client_returns_none_on_error(
+        self, httpx_mock: HTTPXMock
+    ):
+        """CIMD lookup should return None and log on error, not raise."""
+        provider = InMemoryOAuthProvider(base_url="http://localhost:8000")
+
+        client_id = "https://failing.example.com/metadata.json"
+
+        with patch("socket.gethostbyname_ex") as mock_dns:
+            mock_dns.return_value = ("failing.example.com", [], ["93.184.216.34"])
+            httpx_mock.add_response(url=client_id, status_code=404)
+
+            result = await provider._lookup_cimd_client(client_id)
+
+        assert result is None
+
+
+class TestInMemoryProviderCIMD:
+    """Test CIMD integration in InMemoryOAuthProvider."""
+
+    async def test_get_client_uses_cimd_for_url(self, httpx_mock: HTTPXMock):
+        """get_client should use CIMD for URL-based client_ids."""
+        provider = InMemoryOAuthProvider(base_url="http://localhost:8000")
+
+        client_id = "https://client.example.com/metadata.json"
+        metadata = {
+            "client_id": client_id,
+            "client_name": "Test Client",
+            "redirect_uris": ["http://localhost:3000/callback"],
+        }
+
+        with patch("socket.gethostbyname_ex") as mock_dns:
+            mock_dns.return_value = ("client.example.com", [], ["93.184.216.34"])
+            httpx_mock.add_response(url=client_id, json=metadata)
+
+            result = await provider.get_client(client_id)
+
+        assert result is not None
+        assert result.client_id == client_id
+
+    async def test_get_client_uses_registered_for_non_url(self):
+        """get_client should use registered clients for non-URL client_ids."""
+        provider = InMemoryOAuthProvider(base_url="http://localhost:8000")
+
+        # Register a client
+        client_info = OAuthClientInformationFull(
+            client_id="regular-client-id",
+            redirect_uris=[AnyUrl("http://localhost:3000/callback")],
+        )
+        await provider.register_client(client_info)
+
+        result = await provider.get_client("regular-client-id")
+
+        assert result is not None
+        assert result.client_id == "regular-client-id"
+
+    async def test_get_client_returns_none_for_unknown(self):
+        """get_client should return None for unknown non-URL client_ids."""
+        provider = InMemoryOAuthProvider(base_url="http://localhost:8000")
+
+        result = await provider.get_client("unknown-client-id")
+        assert result is None
+
+
+class TestOAuthMetadataAdvertisesCIMD:
+    """Test that OAuth metadata advertises CIMD support."""
+
+    def test_cimd_enabled_by_default(self):
+        """CIMD should be enabled by default."""
+        provider = InMemoryOAuthProvider(base_url="http://localhost:8000")
+        assert provider._cimd_enabled is True
+        assert provider._cimd_cache is not None
+
+    def test_cimd_can_be_disabled(self):
+        """CIMD can be disabled via constructor parameter."""
+        provider = InMemoryOAuthProvider(base_url="http://localhost:8000")
+        provider._cimd_enabled = False
+        # Note: We can't easily test the full disable path without
+        # exposing cimd_enabled in InMemoryOAuthProvider's __init__