feat(auth): OAuth2 client_credentials grant and fixed_client_info

- OAuthClientProvider: add fixed_client_info, _exchange_token_client_credentials
- OAuth2Protocol: accept fixed_client_info for M2M flows
- MultiProtocolAuthProvider: pass fixed_client_info into OAuthClientProvider
- _oauth_401_flow: require client_info for client_credentials, skip dynamic registration
- Server token handler: ClientCredentialsRequest, exchange_client_credentials
- Server routes: advertise client_credentials in grant_types_supported

Author: nypdmax

src/mcp/client/auth/_oauth_401_flow.py

@@ -11,6 +11,7 @@
 
 import httpx
 
+from mcp.client.auth.exceptions import OAuthFlowError
 from mcp.client.auth.utils import (
     build_oauth_authorization_server_metadata_discovery_urls,
     build_protected_resource_metadata_discovery_urls,
@@ -117,6 +118,10 @@ async def oauth_401_flow_generator(
     )
 
     # Step 4: Register client or use URL-based client ID (CIMD)
+    # For client_credentials, a fixed client_id/client_secret must be provided; do not attempt DCR/CIMD.
+    if "client_credentials" in (ctx.client_metadata.grant_types or []) and not ctx.client_info:
+        raise OAuthFlowError("Missing client_info for client_credentials flow")
+
     if not ctx.client_info:
         if should_use_client_metadata_url(ctx.oauth_metadata, ctx.client_metadata_url):
             logger.debug("Using URL-based client ID (CIMD): %s", ctx.client_metadata_url)

src/mcp/client/auth/multi_protocol.py

@@ -393,22 +393,13 @@ def _push(pid: str | None) -> None:
                                         oauth_protocol, "_client_metadata"
                                     ),
                                     storage=cast(OAuth2TokenStorage, self.storage),
-                                    redirect_handler=getattr(
-                                        oauth_protocol, "_redirect_handler", None
-                                    ),
-                                    callback_handler=getattr(
-                                        oauth_protocol, "_callback_handler", None
-                                    ),
-                                    timeout=getattr(
-                                        oauth_protocol, "_timeout", self.timeout
-                                    ),
-                                    client_metadata_url=getattr(
-                                        oauth_protocol, "_client_metadata_url", None
-                                    ),
-                                )
-                                provider.context.protocol_version = request.headers.get(
-                                    MCP_PROTOCOL_VERSION
+                                    redirect_handler=getattr(oauth_protocol, "_redirect_handler", None),
+                                    callback_handler=getattr(oauth_protocol, "_callback_handler", None),
+                                    timeout=getattr(oauth_protocol, "_timeout", self.timeout),
+                                    client_metadata_url=getattr(oauth_protocol, "_client_metadata_url", None),
+                                    fixed_client_info=getattr(oauth_protocol, "_fixed_client_info", None),
                                 )
+                                provider.context.protocol_version = request.headers.get(MCP_PROTOCOL_VERSION)
                                 gen = oauth_401_flow_generator(
                                     provider, original_request, original_401_response, initial_prm=prm
                                 )

src/mcp/client/auth/oauth2.py

@@ -228,6 +228,7 @@ def __init__(
         callback_handler: Callable[[], Awaitable[tuple[str, str | None]]] | None = None,
         timeout: float = 300.0,
         client_metadata_url: str | None = None,
+        fixed_client_info: OAuthClientInformationFull | None = None,
     ):
         """Initialize OAuth2 authentication.
 
@@ -262,6 +263,11 @@ def __init__(
             timeout=timeout,
             client_metadata_url=client_metadata_url,
         )
+        self._fixed_client_info = fixed_client_info
+        if fixed_client_info is not None:
+            # In multi-protocol OAuth flow, we may drive oauth_401_flow_generator directly
+            # without calling _initialize(); ensure client_info is available upfront.
+            self.context.client_info = fixed_client_info
         self._initialized = False
 
     async def _handle_protected_resource_response(self, response: httpx.Response) -> bool:
@@ -297,6 +303,10 @@ async def _handle_protected_resource_response(self, response: httpx.Response) ->
 
     async def _perform_authorization(self) -> httpx.Request:
         """Perform the authorization flow."""
+        grant_types = set(self.context.client_metadata.grant_types or [])
+        if "client_credentials" in grant_types:
+            token_request = await self._exchange_token_client_credentials()
+            return token_request
         auth_code, code_verifier = await self._perform_authorization_code_grant()
         token_request = await self._exchange_token_authorization_code(auth_code, code_verifier)
         return token_request
@@ -362,6 +372,31 @@ def _get_token_endpoint(self) -> str:
             token_url = urljoin(auth_base_url, "/token")
         return token_url
 
+    async def _exchange_token_client_credentials(self) -> httpx.Request:
+        """Build token exchange request for client_credentials flow."""
+        if not self.context.client_info:
+            raise OAuthFlowError("Missing client info for client_credentials flow")
+
+        token_url = self._get_token_endpoint()
+        token_data: dict[str, str] = {
+            "grant_type": "client_credentials",
+        }
+
+        # Some servers require explicit client_id in the form body (especially for client_secret_post).
+        if self.context.client_info.client_id:
+            token_data["client_id"] = self.context.client_info.client_id
+
+        # Only include resource param if conditions are met
+        if self.context.should_include_resource_param(self.context.protocol_version):
+            token_data["resource"] = self.context.get_resource_url()  # RFC 8707
+
+        if self.context.client_metadata.scope:
+            token_data["scope"] = self.context.client_metadata.scope
+
+        headers = {"Content-Type": "application/x-www-form-urlencoded"}
+        token_data, headers = self.context.prepare_token_auth(token_data, headers)
+        return httpx.Request("POST", token_url, data=token_data, headers=headers)
+
     async def _exchange_token_authorization_code(
         self, auth_code: str, code_verifier: str, *, token_data: dict[str, Any] | None = {}
     ) -> httpx.Request:
@@ -543,15 +578,14 @@ async def run_authentication(
                 self.context.client_info = client_information
                 await self.context.storage.set_client_info(client_information)
 
-        auth_code, code_verifier = await self._perform_authorization_code_grant()
-        token_request = await self._exchange_token_authorization_code(auth_code, code_verifier)
+        token_request = await self._perform_authorization()
         token_response = await http_client.send(token_request)
         await self._handle_token_response(token_response)
 
     async def _initialize(self) -> None:  # pragma: no cover
         """Load stored tokens and client info."""
         self.context.current_tokens = await self.context.storage.get_tokens()
-        self.context.client_info = await self.context.storage.get_client_info()
+        self.context.client_info = self._fixed_client_info or await self.context.storage.get_client_info()
         self._initialized = True
 
     def _add_auth_header(self, request: httpx.Request) -> None:

src/mcp/client/auth/protocols/oauth2.py

@@ -31,6 +31,7 @@
 from mcp.shared.auth import (
     AuthCredentials,
     AuthProtocolMetadata,
+    OAuthClientInformationFull,
     OAuthClientMetadata,
     OAuthCredentials,
     OAuthMetadata,
@@ -104,6 +105,7 @@ def __init__(
         callback_handler: Callable[[], Awaitable[tuple[str, str | None]]] | None = None,
         timeout: float = 300.0,
         client_metadata_url: str | None = None,
+        fixed_client_info: OAuthClientInformationFull | None = None,
         dpop_enabled: bool = False,
         dpop_algorithm: DPoPAlgorithm = "ES256",
         dpop_rsa_key_size: int = RSA_KEY_SIZE_DEFAULT,
@@ -113,6 +115,7 @@ def __init__(
         self._callback_handler = callback_handler
         self._timeout = timeout
         self._client_metadata_url = client_metadata_url
+        self._fixed_client_info = fixed_client_info
         self._dpop_enabled = dpop_enabled
         self._dpop_algorithm: DPoPAlgorithm = dpop_algorithm
         self._dpop_rsa_key_size = dpop_rsa_key_size
@@ -133,6 +136,7 @@ async def authenticate(self, context: AuthContext) -> AuthCredentials:
             callback_handler=self._callback_handler,
             timeout=self._timeout,
             client_metadata_url=self._client_metadata_url,
+            fixed_client_info=self._fixed_client_info,
         )
         protocol_version: str | None = None
         if context.protocol_metadata is not None:

src/mcp/server/auth/handlers/token.py

@@ -40,7 +40,20 @@ class RefreshTokenRequest(BaseModel):
     resource: str | None = Field(None, description="Resource indicator for the token")
 
 
-TokenRequest = Annotated[AuthorizationCodeRequest | RefreshTokenRequest, Field(discriminator="grant_type")]
+class ClientCredentialsRequest(BaseModel):
+    # See https://datatracker.ietf.org/doc/html/rfc6749#section-4.4.2
+    grant_type: Literal["client_credentials"]
+    scope: str | None = Field(None, description="Optional scope parameter")
+    client_id: str
+    # we use the client_secret param, per https://datatracker.ietf.org/doc/html/rfc6749#section-2.3.1
+    client_secret: str | None = None
+    # RFC 8707 resource indicator
+    resource: str | None = Field(None, description="Resource indicator for the token")
+
+TokenRequest = Annotated[
+    AuthorizationCodeRequest | RefreshTokenRequest | ClientCredentialsRequest,
+    Field(discriminator="grant_type"),
+]
 token_request_adapter = TypeAdapter[TokenRequest](TokenRequest)
 
 
@@ -216,4 +229,26 @@ async def handle(self, request: Request):
                 except TokenError as e:
                     return self.response(TokenErrorResponse(error=e.error, error_description=e.error_description))
 
+            case ClientCredentialsRequest():
+                # Exchange client credentials for access token
+                scope_str = token_request.scope or getattr(client_info, "scope", None) or ""
+                scopes = scope_str.split(" ") if scope_str else []
+                exchange = getattr(self.provider, "exchange_client_credentials", None)
+                if exchange is None:
+                    return self.response(
+                        TokenErrorResponse(
+                            error="unsupported_grant_type",
+                            error_description="client_credentials is not supported by this authorization server",
+                        )
+                    )
+                try:
+                    tokens = await exchange(client_info, scopes=scopes, resource=token_request.resource)
+                except TokenError as e:
+                    return self.response(
+                        TokenErrorResponse(
+                            error=e.error,
+                            error_description=e.error_description,
+                        )
+                    )
+
         return self.response(tokens)