Add client_secret_basic auth support to MCP client

- Implement HTTP Basic auth for OAuth token requests
- Automatically sets selects auth method when OAuthClientProvider is
  configured with OAuthClientMetadata that has
  token_endpoint_auth_method=None.
- Made OAuthClientMetadata.token_endpoint_auth_method optional to
  support the above auto-configuration.
- Removed ` "token_endpoint_auth_method": "client_secret_post"` from the
  simple-auth-client example as is now auto-configured.

Author: jonshea

examples/clients/simple-auth-client/mcp_simple_auth_client/main.py

@@ -177,7 +177,6 @@ async def callback_handler() -> tuple[str, str | None]:
                 "redirect_uris": ["http://localhost:3030/callback"],
                 "grant_types": ["authorization_code", "refresh_token"],
                 "response_types": ["code"],
-                "token_endpoint_auth_method": "client_secret_post",
             }
 
             async def _default_redirect_handler(authorization_url: str) -> None:

src/mcp/client/auth.py

@@ -14,7 +14,7 @@
 from collections.abc import AsyncGenerator, Awaitable, Callable
 from dataclasses import dataclass, field
 from typing import Protocol
-from urllib.parse import urlencode, urljoin, urlparse
+from urllib.parse import quote, urlencode, urljoin, urlparse
 
 import anyio
 import httpx
@@ -175,6 +175,42 @@ def should_include_resource_param(self, protocol_version: str | None = None) ->
         # Version format is YYYY-MM-DD, so string comparison works
         return protocol_version >= "2025-06-18"
 
+    def prepare_token_auth(
+        self, data: dict[str, str], headers: dict[str, str] | None = None
+    ) -> tuple[dict[str, str], dict[str, str]]:
+        """Prepare authentication for token requests.
+
+        Args:
+            data: The form data to send
+            headers: Optional headers dict to update
+
+        Returns:
+            Tuple of (updated_data, updated_headers)
+        """
+        if headers is None:
+            headers = {}
+
+        if not self.client_info:
+            return data, headers
+
+        auth_method = self.client_info.token_endpoint_auth_method
+
+        if auth_method == "client_secret_basic" and self.client_info.client_secret:
+            # URL-encode client ID and secret per RFC 6749 Section 2.3.1
+            encoded_id = quote(self.client_info.client_id, safe="")
+            encoded_secret = quote(self.client_info.client_secret, safe="")
+            credentials = f"{encoded_id}:{encoded_secret}"
+            encoded_credentials = base64.b64encode(credentials.encode()).decode()
+            headers["Authorization"] = f"Basic {encoded_credentials}"
+            # Don't include client_secret in body for basic auth
+            data = {k: v for k, v in data.items() if k != "client_secret"}
+        elif auth_method == "client_secret_post" and self.client_info.client_secret:
+            # Include client_secret in request body
+            data["client_secret"] = self.client_info.client_secret
+        # For auth_method == "none", don't add any client_secret
+
+        return data, headers
+
 
 class OAuthClientProvider(httpx.Auth):
     """
@@ -291,6 +327,27 @@ async def _register_client(self) -> httpx.Request | None:
 
         registration_data = self.context.client_metadata.model_dump(by_alias=True, mode="json", exclude_none=True)
 
+        # If token_endpoint_auth_method is None, auto-select based on server support
+        if self.context.client_metadata.token_endpoint_auth_method is None:
+            preference_order = ["client_secret_basic", "client_secret_post", "none"]
+
+            if self.context.oauth_metadata and self.context.oauth_metadata.token_endpoint_auth_methods_supported:
+                supported = self.context.oauth_metadata.token_endpoint_auth_methods_supported
+                for method in preference_order:
+                    if method in supported:
+                        registration_data["token_endpoint_auth_method"] = method
+                        break
+                else:
+                    # No compatible methods between client and server
+                    raise OAuthRegistrationError(
+                        f"No compatible authentication methods. "
+                        f"Server supports: {supported}, "
+                        f"Client supports: {preference_order}"
+                    )
+            else:
+                # No server metadata available, use our default preference
+                registration_data["token_endpoint_auth_method"] = preference_order[0]
+
         return httpx.Request(
             "POST", registration_url, json=registration_data, headers={"Content-Type": "application/json"}
         )
@@ -378,12 +435,11 @@ async def _exchange_token(self, auth_code: str, code_verifier: str) -> httpx.Req
         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_info.client_secret:
-            token_data["client_secret"] = self.context.client_info.client_secret
+        # Prepare authentication based on preferred method
+        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={"Content-Type": "application/x-www-form-urlencoded"}
-        )
+        return httpx.Request("POST", token_url, data=token_data, headers=headers)
 
     async def _handle_token_response(self, response: httpx.Response) -> None:
         """Handle token exchange response."""
@@ -432,12 +488,11 @@ async def _refresh_token(self) -> httpx.Request:
         if self.context.should_include_resource_param(self.context.protocol_version):
             refresh_data["resource"] = self.context.get_resource_url()  # RFC 8707
 
-        if self.context.client_info.client_secret:
-            refresh_data["client_secret"] = self.context.client_info.client_secret
+        # Prepare authentication based on preferred method
+        headers = {"Content-Type": "application/x-www-form-urlencoded"}
+        refresh_data, headers = self.context.prepare_token_auth(refresh_data, headers)
 
-        return httpx.Request(
-            "POST", token_url, data=refresh_data, headers={"Content-Type": "application/x-www-form-urlencoded"}
-        )
+        return httpx.Request("POST", token_url, data=refresh_data, headers=headers)
 
     async def _handle_refresh_response(self, response: httpx.Response) -> bool:
         """Handle token refresh response. Returns True if successful."""

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

@@ -49,6 +49,11 @@ async def handle(self, request: Request) -> Response:
             )
 
         client_id = str(uuid4())
+
+        # If auth method is None, default to client_secret_post
+        if client_metadata.token_endpoint_auth_method is None:
+            client_metadata.token_endpoint_auth_method = "client_secret_post"
+
         client_secret = None
         if client_metadata.token_endpoint_auth_method != "none":
             # cryptographically secure random 32-byte hex string

src/mcp/shared/auth.py

@@ -42,7 +42,7 @@ class OAuthClientMetadata(BaseModel):
     """
 
     redirect_uris: list[AnyUrl] = Field(..., min_length=1)
-    token_endpoint_auth_method: Literal["none", "client_secret_post", "client_secret_basic"] = "client_secret_post"
+    token_endpoint_auth_method: Literal["none", "client_secret_post", "client_secret_basic"] | None = None
     # grant_types: this implementation only supports authorization_code & refresh_token
     grant_types: list[Literal["authorization_code", "refresh_token"] | str] = [
         "authorization_code",

tests/client/test_auth.py

@@ -2,16 +2,25 @@
 Tests for refactored OAuth client authentication implementation.
 """
 
+import base64
+import json
 import time
 from unittest import mock
+from urllib.parse import unquote
 
 import httpx
 import pytest
 from inline_snapshot import Is, snapshot
 from pydantic import AnyHttpUrl, AnyUrl
 
-from mcp.client.auth import OAuthClientProvider, PKCEParameters
-from mcp.shared.auth import OAuthClientInformationFull, OAuthClientMetadata, OAuthToken, ProtectedResourceMetadata
+from mcp.client.auth import OAuthClientProvider, OAuthRegistrationError, PKCEParameters
+from mcp.shared.auth import (
+    OAuthClientInformationFull,
+    OAuthClientMetadata,
+    OAuthMetadata,
+    OAuthToken,
+    ProtectedResourceMetadata,
+)
 
 
 class MockTokenStorage:
@@ -415,6 +424,43 @@ async def test_register_client_skip_if_registered(self, oauth_provider: OAuthCli
         request = await oauth_provider._register_client()
         assert request is None
 
+    @pytest.mark.anyio
+    async def test_register_client_none_auth_method_with_server_metadata(self, oauth_provider: OAuthClientProvider):
+        """Test that token_endpoint_auth_method=None selects from server's supported methods."""
+        # Set server metadata with specific supported methods
+        oauth_provider.context.oauth_metadata = OAuthMetadata(
+            issuer=AnyHttpUrl("https://auth.example.com"),
+            authorization_endpoint=AnyHttpUrl("https://auth.example.com/authorize"),
+            token_endpoint=AnyHttpUrl("https://auth.example.com/token"),
+            token_endpoint_auth_methods_supported=["client_secret_post"],
+        )
+        # Ensure client_metadata has None for token_endpoint_auth_method
+        assert oauth_provider.context.client_metadata.token_endpoint_auth_method is None
+
+        request = await oauth_provider._register_client()
+        assert request is not None
+
+        body = json.loads(request.content)
+        assert body["token_endpoint_auth_method"] == "client_secret_post"
+
+    @pytest.mark.anyio
+    async def test_register_client_none_auth_method_no_compatible(self, oauth_provider: OAuthClientProvider):
+        """Test that registration raises error when no compatible auth methods."""
+        # Set server metadata with unsupported methods only
+        oauth_provider.context.oauth_metadata = OAuthMetadata(
+            issuer=AnyHttpUrl("https://auth.example.com"),
+            authorization_endpoint=AnyHttpUrl("https://auth.example.com/authorize"),
+            token_endpoint=AnyHttpUrl("https://auth.example.com/token"),
+            token_endpoint_auth_methods_supported=["private_key_jwt", "client_secret_jwt"],
+        )
+        assert oauth_provider.context.client_metadata.token_endpoint_auth_method is None
+
+        with pytest.raises(OAuthRegistrationError) as exc_info:
+            await oauth_provider._register_client()
+
+        assert "No compatible authentication methods" in str(exc_info.value)
+        assert "private_key_jwt" in str(exc_info.value)
+
     @pytest.mark.anyio
     async def test_token_exchange_request(self, oauth_provider: OAuthClientProvider):
         """Test token exchange request building."""
@@ -423,6 +469,7 @@ async def test_token_exchange_request(self, oauth_provider: OAuthClientProvider)
             client_id="test_client",
             client_secret="test_secret",
             redirect_uris=[AnyUrl("http://localhost:3030/callback")],
+            token_endpoint_auth_method="client_secret_post",
         )
 
         request = await oauth_provider._exchange_token("test_auth_code", "test_verifier")
@@ -448,6 +495,7 @@ async def test_refresh_token_request(self, oauth_provider: OAuthClientProvider,
             client_id="test_client",
             client_secret="test_secret",
             redirect_uris=[AnyUrl("http://localhost:3030/callback")],
+            token_endpoint_auth_method="client_secret_post",
         )
 
         request = await oauth_provider._refresh_token()
@@ -463,6 +511,114 @@ async def test_refresh_token_request(self, oauth_provider: OAuthClientProvider,
         assert "client_id=test_client" in content
         assert "client_secret=test_secret" in content
 
+    @pytest.mark.anyio
+    async def test_basic_auth_token_exchange(self, oauth_provider: OAuthClientProvider):
+        """Test token exchange with client_secret_basic authentication."""
+        # Set up OAuth metadata to support basic auth
+        oauth_provider.context.oauth_metadata = OAuthMetadata(
+            issuer=AnyHttpUrl("https://auth.example.com"),
+            authorization_endpoint=AnyHttpUrl("https://auth.example.com/authorize"),
+            token_endpoint=AnyHttpUrl("https://auth.example.com/token"),
+            token_endpoint_auth_methods_supported=["client_secret_basic", "client_secret_post"],
+        )
+
+        client_id_raw = "test@client"  # Include special character to test URL encoding
+        client_secret_raw = "test:secret"  # Include colon to test URL encoding
+
+        oauth_provider.context.client_info = OAuthClientInformationFull(
+            client_id=client_id_raw,
+            client_secret=client_secret_raw,
+            redirect_uris=[AnyUrl("http://localhost:3030/callback")],
+            token_endpoint_auth_method="client_secret_basic",
+        )
+
+        request = await oauth_provider._exchange_token("test_auth_code", "test_verifier")
+
+        # Should use basic auth (registered method)
+        assert "Authorization" in request.headers
+        assert request.headers["Authorization"].startswith("Basic ")
+
+        # Decode and verify credentials are properly URL-encoded
+        encoded_creds = request.headers["Authorization"][6:]  # Remove "Basic " prefix
+        decoded = base64.b64decode(encoded_creds).decode()
+        client_id, client_secret = decoded.split(":", 1)
+
+        # Check URL encoding was applied
+        assert client_id == "test%40client"  # @ should be encoded as %40
+        assert client_secret == "test%3Asecret"  # : should be encoded as %3A
+
+        # Verify decoded values match original
+        assert unquote(client_id) == client_id_raw
+        assert unquote(client_secret) == client_secret_raw
+
+        # client_secret should NOT be in body for basic auth
+        content = request.content.decode()
+        assert "client_secret=" not in content
+        assert "client_id=test%40client" in content  # client_id still in body
+
+    @pytest.mark.anyio
+    async def test_basic_auth_refresh_token(self, oauth_provider: OAuthClientProvider, valid_tokens: OAuthToken):
+        """Test token refresh with client_secret_basic authentication."""
+        oauth_provider.context.current_tokens = valid_tokens
+
+        # Set up OAuth metadata to only support basic auth
+        oauth_provider.context.oauth_metadata = OAuthMetadata(
+            issuer=AnyHttpUrl("https://auth.example.com"),
+            authorization_endpoint=AnyHttpUrl("https://auth.example.com/authorize"),
+            token_endpoint=AnyHttpUrl("https://auth.example.com/token"),
+            token_endpoint_auth_methods_supported=["client_secret_basic"],
+        )
+
+        client_id = "test_client"
+        client_secret = "test_secret"
+        oauth_provider.context.client_info = OAuthClientInformationFull(
+            client_id=client_id,
+            client_secret=client_secret,
+            redirect_uris=[AnyUrl("http://localhost:3030/callback")],
+            token_endpoint_auth_method="client_secret_basic",
+        )
+
+        request = await oauth_provider._refresh_token()
+
+        assert "Authorization" in request.headers
+        assert request.headers["Authorization"].startswith("Basic ")
+
+        encoded_creds = request.headers["Authorization"][6:]
+        decoded = base64.b64decode(encoded_creds).decode()
+        assert decoded == f"{client_id}:{client_secret}"
+
+        # client_secret should NOT be in body
+        content = request.content.decode()
+        assert "client_secret=" not in content
+
+    @pytest.mark.anyio
+    async def test_none_auth_method(self, oauth_provider: OAuthClientProvider):
+        """Test 'none' authentication method (public client)."""
+        oauth_provider.context.oauth_metadata = OAuthMetadata(
+            issuer=AnyHttpUrl("https://auth.example.com"),
+            authorization_endpoint=AnyHttpUrl("https://auth.example.com/authorize"),
+            token_endpoint=AnyHttpUrl("https://auth.example.com/token"),
+            token_endpoint_auth_methods_supported=["none"],
+        )
+
+        client_id = "public_client"
+        oauth_provider.context.client_info = OAuthClientInformationFull(
+            client_id=client_id,
+            client_secret=None,  # No secret for public client
+            redirect_uris=[AnyUrl("http://localhost:3030/callback")],
+            token_endpoint_auth_method="none",
+        )
+
+        request = await oauth_provider._exchange_token("test_auth_code", "test_verifier")
+
+        # Should NOT have Authorization header
+        assert "Authorization" not in request.headers
+
+        # Should NOT have client_secret in body
+        content = request.content.decode()
+        assert "client_secret=" not in content
+        assert "client_id=public_client" in content
+
 
 class TestProtectedResourceMetadata:
     """Test protected resource handling."""