feat: add SEP-1046 client_credentials grant with private_key_jwt auth

Implements OAuth client_credentials grant type with JWT client authentication
per SEP-1046 and RFC 7523 Section 2.2.

Changes to RFC7523OAuthClientProvider:
- Add _exchange_token_client_credentials() for client_credentials grant
- Update _perform_authorization() to check for client_credentials first
- Preserve legacy jwt-bearer grant flow for backwards compatibility

Updates to conformance auth client:
- Accept scenario name as first argument
- Add run_client_credentials_client() for auth/client-credentials-jwt scenario
- Use well-known conformance test EC P-256 private key

Author: pcarleton

examples/clients/conformance-auth-client/mcp_conformance_auth_client/__init__.py

@@ -7,7 +7,11 @@
 fetching the authorization URL and extracting the auth code from the redirect.
 
 Usage:
-    python -m mcp_conformance_auth_client <server-url>
+    python -m mcp_conformance_auth_client <scenario> <server-url>
+
+Scenarios:
+    auth/*                    - Authorization code flow scenarios (default behavior)
+    auth/client-credentials-jwt - Client credentials with JWT authentication (SEP-1046)
 """
 
 import asyncio
@@ -19,10 +23,23 @@
 import httpx
 from mcp import ClientSession
 from mcp.client.auth import OAuthClientProvider, TokenStorage
+from mcp.client.auth.extensions.client_credentials import JWTParameters, RFC7523OAuthClientProvider
 from mcp.client.streamable_http import streamablehttp_client
 from mcp.shared.auth import OAuthClientInformationFull, OAuthClientMetadata, OAuthToken
 from pydantic import AnyUrl
 
+# Well-known test private key for conformance testing (EC P-256).
+# This corresponds to the public key in the conformance test framework.
+# The key is intentionally public for conformance testing purposes.
+CONFORMANCE_TEST_PRIVATE_KEY_PEM = """-----BEGIN EC PRIVATE KEY-----
+MHcCAQEEIAIbFaPsD3ShCQpvGzgyEIvLQo8pLOsEMwQWcRq+CwWGoAoGCCqGSM49
+AwEHoUQDQgAEiYsmK6YG68IEZ4k0vVSg1Cy8PRwsmz/XweRkTsYof+yVMH+FYm5P
+0IhAPaKV/4DjucO6UcQ+d/8q7i//mT7WQA==
+-----END EC PRIVATE KEY-----"""
+
+# The client_id expected by the conformance test
+CONFORMANCE_TEST_CLIENT_ID = "conformance-test-client"
+
 # Set up logging to stderr (stdout is for conformance test output)
 logging.basicConfig(
     level=logging.DEBUG,
@@ -111,17 +128,17 @@ async def handle_callback(self) -> tuple[str, str | None]:
         return auth_code, state
 
 
-async def run_client(server_url: str) -> None:
+async def run_authorization_code_client(server_url: str) -> None:
     """
-    Run the conformance test client against the given server URL.
+    Run the conformance test client with authorization code flow.
 
     This function:
-    1. Connects to the MCP server with OAuth authentication
+    1. Connects to the MCP server with OAuth authorization code flow
     2. Initializes the session
     3. Lists available tools
     4. Calls a test tool
     """
-    logger.debug(f"Starting conformance auth client for {server_url}")
+    logger.debug(f"Starting conformance auth client (authorization_code) for {server_url}")
 
     # Create callback handler that will automatically fetch auth codes
     callback_handler = ConformanceOAuthCallbackHandler()
@@ -140,6 +157,50 @@ async def run_client(server_url: str) -> None:
         callback_handler=callback_handler.handle_callback,
     )
 
+    await _run_session(server_url, oauth_auth)
+
+
+async def run_client_credentials_client(server_url: str) -> None:
+    """
+    Run the conformance test client with client credentials flow (SEP-1046).
+
+    This function:
+    1. Connects to the MCP server with OAuth client_credentials grant
+    2. Uses private_key_jwt authentication with the well-known test key
+    3. Initializes the session
+    4. Lists available tools
+    5. Calls a test tool
+    """
+    logger.debug(f"Starting conformance auth client (client_credentials) for {server_url}")
+
+    # Create JWT parameters for private_key_jwt authentication
+    jwt_params = JWTParameters(
+        issuer=CONFORMANCE_TEST_CLIENT_ID,
+        subject=CONFORMANCE_TEST_CLIENT_ID,
+        jwt_signing_algorithm="ES256",
+        jwt_signing_key=CONFORMANCE_TEST_PRIVATE_KEY_PEM,
+    )
+
+    # Create OAuth authentication handler for client_credentials flow
+    # Note: redirect_uris is required by the model but not used in client_credentials flow
+    oauth_auth = RFC7523OAuthClientProvider(
+        server_url=server_url,
+        client_metadata=OAuthClientMetadata(
+            client_name=CONFORMANCE_TEST_CLIENT_ID,
+            redirect_uris=[AnyUrl("http://localhost:0/unused")],  # Required but unused
+            grant_types=["client_credentials"],
+            response_types=[],
+            token_endpoint_auth_method="private_key_jwt",
+        ),
+        storage=InMemoryTokenStorage(),
+        jwt_parameters=jwt_params,
+    )
+
+    await _run_session(server_url, oauth_auth)
+
+
+async def _run_session(server_url: str, oauth_auth: OAuthClientProvider) -> None:
+    """Common session logic for all OAuth flows."""
     # Connect using streamable HTTP transport with OAuth
     async with streamablehttp_client(
         url=server_url,
@@ -168,14 +229,23 @@ async def run_client(server_url: str) -> None:
 
 def main() -> None:
     """Main entry point for the conformance auth client."""
-    if len(sys.argv) != 2:
-        print(f"Usage: {sys.argv[0]} <server-url>", file=sys.stderr)
+    if len(sys.argv) != 3:
+        print(f"Usage: {sys.argv[0]} <scenario> <server-url>", file=sys.stderr)
+        print("", file=sys.stderr)
+        print("Scenarios:", file=sys.stderr)
+        print("  auth/*                      - Authorization code flow (default)", file=sys.stderr)
+        print("  auth/client-credentials-jwt - Client credentials with JWT auth (SEP-1046)", file=sys.stderr)
         sys.exit(1)
 
-    server_url = sys.argv[1]
+    scenario = sys.argv[1]
+    server_url = sys.argv[2]
 
     try:
-        asyncio.run(run_client(server_url))
+        if scenario == "auth/client-credentials-jwt":
+            asyncio.run(run_client_credentials_client(server_url))
+        else:
+            # Default to authorization code flow for all other auth/* scenarios
+            asyncio.run(run_authorization_code_client(server_url))
     except Exception:
         logger.exception("Client failed")
         sys.exit(1)

src/mcp/client/auth/extensions/client_credentials.py

@@ -92,7 +92,12 @@ async def _exchange_token_authorization_code(
 
     async def _perform_authorization(self) -> httpx.Request:  # pragma: no cover
         """Perform the authorization flow."""
-        if "urn:ietf:params:oauth:grant-type:jwt-bearer" in self.context.client_metadata.grant_types:
+        if "client_credentials" in self.context.client_metadata.grant_types:
+            # SEP-1046: client_credentials grant with private_key_jwt authentication
+            token_request = await self._exchange_token_client_credentials()
+            return token_request
+        elif "urn:ietf:params:oauth:grant-type:jwt-bearer" in self.context.client_metadata.grant_types:
+            # RFC 7523 Section 2.1: JWT bearer grant (legacy)
             token_request = await self._exchange_token_jwt_bearer()
             return token_request
         else:
@@ -117,8 +122,43 @@ def _add_client_authentication_jwt(self, *, token_data: dict[str, Any]):  # prag
         # it represents the resource server that will validate the token
         token_data["audience"] = self.context.get_resource_url()
 
+    async def _exchange_token_client_credentials(self) -> httpx.Request:
+        """Build token exchange request for client_credentials grant with private_key_jwt auth.
+
+        This implements SEP-1046: OAuth Client Credentials Extension for MCP.
+        Uses RFC 7523 Section 2.2 for client authentication via JWT assertion.
+        """
+        if not self.context.client_info:
+            raise OAuthFlowError("Missing client info")  # pragma: no cover
+        if not self.jwt_parameters:
+            raise OAuthFlowError("Missing JWT parameters for client_credentials flow")  # pragma: no cover
+        if not self.context.oauth_metadata:
+            raise OAuthTokenError("Missing OAuth metadata")  # pragma: no cover
+
+        token_data: dict[str, Any] = {
+            "grant_type": "client_credentials",
+        }
+
+        # Add JWT client authentication (RFC 7523 Section 2.2)
+        self._add_client_authentication_jwt(token_data=token_data)
+
+        if self.context.should_include_resource_param(self.context.protocol_version):  # pragma: no branch
+            token_data["resource"] = self.context.get_resource_url()
+
+        if self.context.client_metadata.scope:  # pragma: no branch
+            token_data["scope"] = self.context.client_metadata.scope
+
+        token_url = self._get_token_endpoint()
+        return httpx.Request(
+            "POST", token_url, data=token_data, headers={"Content-Type": "application/x-www-form-urlencoded"}
+        )
+
     async def _exchange_token_jwt_bearer(self) -> httpx.Request:
-        """Build token exchange request for JWT bearer grant."""
+        """Build token exchange request for JWT bearer grant (RFC 7523 Section 2.1).
+
+        This is the legacy JWT bearer authorization grant where the JWT itself
+        is the authorization grant, not client authentication.
+        """
         if not self.context.client_info:
             raise OAuthFlowError("Missing client info")  # pragma: no cover
         if not self.jwt_parameters: