Support client_credentials flow with JWT and Basic auth (#1663)

Co-authored-by: Claude <noreply@anthropic.com>
Co-authored-by: Felix Weinberger <3823880+felixweinberger@users.noreply.github.com>

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

@@ -7,11 +7,27 @@ It automatically handles OAuth flows without user interaction by programmaticall
 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>
+
+Environment Variables:
+    MCP_CONFORMANCE_CONTEXT - JSON object containing test credentials:
+        {
+            "client_id": "...",
+            "client_secret": "...",       # For client_secret_basic flow
+            "private_key_pem": "...",     # For private_key_jwt flow
+            "signing_algorithm": "ES256"  # Optional, defaults to ES256
+        }
+
+Scenarios:
+    auth/*                      - Authorization code flow scenarios (default behavior)
+    auth/client-credentials-jwt   - Client credentials with JWT authentication (SEP-1046)
+    auth/client-credentials-basic - Client credentials with client_secret_basic
 """
 
 import asyncio
+import json
 import logging
+import os
 import sys
 from datetime import timedelta
 from urllib.parse import ParseResult, parse_qs, urlparse
@@ -19,10 +35,30 @@ from urllib.parse import ParseResult, parse_qs, urlparse
 import httpx
 from mcp import ClientSession
 from mcp.client.auth import OAuthClientProvider, TokenStorage
+from mcp.client.auth.extensions.client_credentials import (
+    ClientCredentialsOAuthProvider,
+    PrivateKeyJWTOAuthProvider,
+    SignedJWTParameters,
+)
 from mcp.client.streamable_http import streamablehttp_client
 from mcp.shared.auth import OAuthClientInformationFull, OAuthClientMetadata, OAuthToken
 from pydantic import AnyUrl
 
+
+def get_conformance_context() -> dict:
+    """Load conformance test context from MCP_CONFORMANCE_CONTEXT environment variable."""
+    context_json = os.environ.get("MCP_CONFORMANCE_CONTEXT")
+    if not context_json:
+        raise RuntimeError(
+            "MCP_CONFORMANCE_CONTEXT environment variable not set. "
+            "Expected JSON with client_id, client_secret, and/or private_key_pem."
+        )
+    try:
+        return json.loads(context_json)
+    except json.JSONDecodeError as e:
+        raise RuntimeError(f"Failed to parse MCP_CONFORMANCE_CONTEXT as JSON: {e}") from e
+
+
 # Set up logging to stderr (stdout is for conformance test output)
 logging.basicConfig(
     level=logging.DEBUG,
@@ -111,17 +147,17 @@ class ConformanceOAuthCallbackHandler:
         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 +176,89 @@ 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_jwt_client(server_url: str) -> None:
+    """
+    Run the conformance test client with client credentials flow using private_key_jwt (SEP-1046).
+
+    This function:
+    1. Connects to the MCP server with OAuth client_credentials grant
+    2. Uses private_key_jwt authentication with credentials from MCP_CONFORMANCE_CONTEXT
+    3. Initializes the session
+    4. Lists available tools
+    5. Calls a test tool
+    """
+    logger.debug(f"Starting conformance auth client (client_credentials_jwt) for {server_url}")
+
+    # Load credentials from environment
+    context = get_conformance_context()
+    client_id = context.get("client_id")
+    private_key_pem = context.get("private_key_pem")
+    signing_algorithm = context.get("signing_algorithm", "ES256")
+
+    if not client_id:
+        raise RuntimeError("MCP_CONFORMANCE_CONTEXT missing 'client_id'")
+    if not private_key_pem:
+        raise RuntimeError("MCP_CONFORMANCE_CONTEXT missing 'private_key_pem'")
+
+    # Create JWT parameters for SDK-signed assertions
+    jwt_params = SignedJWTParameters(
+        issuer=client_id,
+        subject=client_id,
+        signing_algorithm=signing_algorithm,
+        signing_key=private_key_pem,
+    )
+
+    # Create OAuth provider for client_credentials with private_key_jwt
+    oauth_auth = PrivateKeyJWTOAuthProvider(
+        server_url=server_url,
+        storage=InMemoryTokenStorage(),
+        client_id=client_id,
+        assertion_provider=jwt_params.create_assertion_provider(),
+    )
+
+    await _run_session(server_url, oauth_auth)
+
+
+async def run_client_credentials_basic_client(server_url: str) -> None:
+    """
+    Run the conformance test client with client credentials flow using client_secret_basic.
+
+    This function:
+    1. Connects to the MCP server with OAuth client_credentials grant
+    2. Uses client_secret_basic authentication with credentials from MCP_CONFORMANCE_CONTEXT
+    3. Initializes the session
+    4. Lists available tools
+    5. Calls a test tool
+    """
+    logger.debug(f"Starting conformance auth client (client_credentials_basic) for {server_url}")
+
+    # Load credentials from environment
+    context = get_conformance_context()
+    client_id = context.get("client_id")
+    client_secret = context.get("client_secret")
+
+    if not client_id:
+        raise RuntimeError("MCP_CONFORMANCE_CONTEXT missing 'client_id'")
+    if not client_secret:
+        raise RuntimeError("MCP_CONFORMANCE_CONTEXT missing 'client_secret'")
+
+    # Create OAuth provider for client_credentials with client_secret_basic
+    oauth_auth = ClientCredentialsOAuthProvider(
+        server_url=server_url,
+        storage=InMemoryTokenStorage(),
+        client_id=client_id,
+        client_secret=client_secret,
+        token_endpoint_auth_method="client_secret_basic",
+    )
+
+    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 +287,26 @@ async def run_client(server_url: str) -> None:
 
 def main() -> None:
     """Main entry point for the conformance auth client."""
-    if len(sys.argv) != 2:
+    if len(sys.argv) != 3:
-        print(f"Usage: {sys.argv[0]} <server-url>", file=sys.stderr)
+        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)
+        print("  auth/client-credentials-basic - Client credentials with client_secret_basic", 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_jwt_client(server_url))
+        elif scenario == "auth/client-credentials-basic":
+            asyncio.run(run_client_credentials_basic_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

@@ -1,6 +1,16 @@
+"""
+OAuth client credential extensions for MCP.
+
+Provides OAuth providers for machine-to-machine authentication flows:
+- ClientCredentialsOAuthProvider: For client_credentials with client_id + client_secret
+- PrivateKeyJWTOAuthProvider: For client_credentials with private_key_jwt authentication
+  (typically using a pre-built JWT from workload identity federation)
+- RFC7523OAuthClientProvider: For jwt-bearer grant (RFC 7523 Section 2.1)
+"""
+
 import time
 from collections.abc import Awaitable, Callable
-from typing import Any
+from typing import Any, Literal
 from uuid import uuid4
 
 import httpx
@@ -8,7 +18,321 @@ import jwt
 from pydantic import BaseModel, Field
 
 from mcp.client.auth import OAuthClientProvider, OAuthFlowError, OAuthTokenError, TokenStorage
-from mcp.shared.auth import OAuthClientMetadata
+from mcp.shared.auth import OAuthClientInformationFull, OAuthClientMetadata
+
+
+class ClientCredentialsOAuthProvider(OAuthClientProvider):
+    """OAuth provider for client_credentials grant with client_id + client_secret.
+
+    This provider sets client_info directly, bypassing dynamic client registration.
+    Use this when you already have client credentials (client_id and client_secret).
+
+    Example:
+        ```python
+        provider = ClientCredentialsOAuthProvider(
+            server_url="https://api.example.com",
+            storage=my_token_storage,
+            client_id="my-client-id",
+            client_secret="my-client-secret",
+        )
+        ```
+    """
+
+    def __init__(
+        self,
+        server_url: str,
+        storage: TokenStorage,
+        client_id: str,
+        client_secret: str,
+        token_endpoint_auth_method: Literal["client_secret_basic", "client_secret_post"] = "client_secret_basic",
+        scopes: str | None = None,
+    ) -> None:
+        """Initialize client_credentials OAuth provider.
+
+        Args:
+            server_url: The MCP server URL.
+            storage: Token storage implementation.
+            client_id: The OAuth client ID.
+            client_secret: The OAuth client secret.
+            token_endpoint_auth_method: Authentication method for token endpoint.
+                Either "client_secret_basic" (default) or "client_secret_post".
+            scopes: Optional space-separated list of scopes to request.
+        """
+        # Build minimal client_metadata for the base class
+        client_metadata = OAuthClientMetadata(
+            redirect_uris=None,
+            grant_types=["client_credentials"],
+            token_endpoint_auth_method=token_endpoint_auth_method,
+            scope=scopes,
+        )
+        super().__init__(server_url, client_metadata, storage, None, None, 300.0)
+        # Store client_info to be set during _initialize - no dynamic registration needed
+        self._fixed_client_info = OAuthClientInformationFull(
+            redirect_uris=None,
+            client_id=client_id,
+            client_secret=client_secret,
+            grant_types=["client_credentials"],
+            token_endpoint_auth_method=token_endpoint_auth_method,
+            scope=scopes,
+        )
+
+    async def _initialize(self) -> None:
+        """Load stored tokens and set pre-configured client_info."""
+        self.context.current_tokens = await self.context.storage.get_tokens()
+        self.context.client_info = self._fixed_client_info
+        self._initialized = True
+
+    async def _perform_authorization(self) -> httpx.Request:
+        """Perform client_credentials authorization."""
+        return await self._exchange_token_client_credentials()
+
+    async def _exchange_token_client_credentials(self) -> httpx.Request:
+        """Build token exchange request for client_credentials grant."""
+        token_data: dict[str, Any] = {
+            "grant_type": "client_credentials",
+        }
+
+        headers: dict[str, str] = {"Content-Type": "application/x-www-form-urlencoded"}
+
+        # Use standard auth methods (client_secret_basic, client_secret_post, none)
+        token_data, headers = self.context.prepare_token_auth(token_data, headers)
+
+        if self.context.should_include_resource_param(self.context.protocol_version):
+            token_data["resource"] = self.context.get_resource_url()
+
+        if self.context.client_metadata.scope:
+            token_data["scope"] = self.context.client_metadata.scope
+
+        token_url = self._get_token_endpoint()
+        return httpx.Request("POST", token_url, data=token_data, headers=headers)
+
+
+def static_assertion_provider(token: str) -> Callable[[str], Awaitable[str]]:
+    """Create an assertion provider that returns a static JWT token.
+
+    Use this when you have a pre-built JWT (e.g., from workload identity federation)
+    that doesn't need the audience parameter.
+
+    Example:
+        ```python
+        provider = PrivateKeyJWTOAuthProvider(
+            server_url="https://api.example.com",
+            storage=my_token_storage,
+            client_id="my-client-id",
+            assertion_provider=static_assertion_provider(my_prebuilt_jwt),
+        )
+        ```
+
+    Args:
+        token: The pre-built JWT assertion string.
+
+    Returns:
+        An async callback suitable for use as an assertion_provider.
+    """
+
+    async def provider(audience: str) -> str:
+        return token
+
+    return provider
+
+
+class SignedJWTParameters(BaseModel):
+    """Parameters for creating SDK-signed JWT assertions.
+
+    Use `create_assertion_provider()` to create an assertion provider callback
+    for use with `PrivateKeyJWTOAuthProvider`.
+
+    Example:
+        ```python
+        jwt_params = SignedJWTParameters(
+            issuer="my-client-id",
+            subject="my-client-id",
+            signing_key=private_key_pem,
+        )
+        provider = PrivateKeyJWTOAuthProvider(
+            server_url="https://api.example.com",
+            storage=my_token_storage,
+            client_id="my-client-id",
+            assertion_provider=jwt_params.create_assertion_provider(),
+        )
+        ```
+    """
+
+    issuer: str = Field(description="Issuer for JWT assertions (typically client_id).")
+    subject: str = Field(description="Subject identifier for JWT assertions (typically client_id).")
+    signing_key: str = Field(description="Private key for JWT signing (PEM format).")
+    signing_algorithm: str = Field(default="RS256", description="Algorithm for signing JWT assertions.")
+    lifetime_seconds: int = Field(default=300, description="Lifetime of generated JWT in seconds.")
+    additional_claims: dict[str, Any] | None = Field(default=None, description="Additional claims.")
+
+    def create_assertion_provider(self) -> Callable[[str], Awaitable[str]]:
+        """Create an assertion provider callback for use with PrivateKeyJWTOAuthProvider.
+
+        Returns:
+            An async callback that takes the audience (authorization server issuer URL)
+            and returns a signed JWT assertion.
+        """
+
+        async def provider(audience: str) -> str:
+            now = int(time.time())
+            claims: dict[str, Any] = {
+                "iss": self.issuer,
+                "sub": self.subject,
+                "aud": audience,
+                "exp": now + self.lifetime_seconds,
+                "iat": now,
+                "jti": str(uuid4()),
+            }
+            if self.additional_claims:
+                claims.update(self.additional_claims)
+
+            return jwt.encode(claims, self.signing_key, algorithm=self.signing_algorithm)
+
+        return provider
+
+
+class PrivateKeyJWTOAuthProvider(OAuthClientProvider):
+    """OAuth provider for client_credentials grant with private_key_jwt authentication.
+
+    Uses RFC 7523 Section 2.2 for client authentication via JWT assertion.
+
+    The JWT assertion's audience MUST be the authorization server's issuer identifier
+    (per RFC 7523bis security updates). The `assertion_provider` callback receives
+    this audience value and must return a JWT with that audience.
+
+    **Option 1: Pre-built JWT via Workload Identity Federation**
+
+    In production scenarios, the JWT assertion is typically obtained from a workload
+    identity provider (e.g., GCP, AWS IAM, Azure AD):
+
+        ```python
+        async def get_workload_identity_token(audience: str) -> str:
+            # Fetch JWT from your identity provider
+            # The JWT's audience must match the provided audience parameter
+            return await fetch_token_from_identity_provider(audience=audience)
+
+        provider = PrivateKeyJWTOAuthProvider(
+            server_url="https://api.example.com",
+            storage=my_token_storage,
+            client_id="my-client-id",
+            assertion_provider=get_workload_identity_token,
+        )
+        ```
+
+    **Option 2: Static pre-built JWT**
+
+    If you have a static JWT that doesn't need the audience parameter:
+
+        ```python
+        provider = PrivateKeyJWTOAuthProvider(
+            server_url="https://api.example.com",
+            storage=my_token_storage,
+            client_id="my-client-id",
+            assertion_provider=static_assertion_provider(my_prebuilt_jwt),
+        )
+        ```
+
+    **Option 3: SDK-signed JWT (for testing/simple setups)**
+
+    For testing or simple deployments, use `SignedJWTParameters.create_assertion_provider()`:
+
+        ```python
+        jwt_params = SignedJWTParameters(
+            issuer="my-client-id",
+            subject="my-client-id",
+            signing_key=private_key_pem,
+        )
+        provider = PrivateKeyJWTOAuthProvider(
+            server_url="https://api.example.com",
+            storage=my_token_storage,
+            client_id="my-client-id",
+            assertion_provider=jwt_params.create_assertion_provider(),
+        )
+        ```
+    """
+
+    def __init__(
+        self,
+        server_url: str,
+        storage: TokenStorage,
+        client_id: str,
+        assertion_provider: Callable[[str], Awaitable[str]],
+        scopes: str | None = None,
+    ) -> None:
+        """Initialize private_key_jwt OAuth provider.
+
+        Args:
+            server_url: The MCP server URL.
+            storage: Token storage implementation.
+            client_id: The OAuth client ID.
+            assertion_provider: Async callback that takes the audience (authorization
+                server's issuer identifier) and returns a JWT assertion. Use
+                `SignedJWTParameters.create_assertion_provider()` for SDK-signed JWTs,
+                `static_assertion_provider()` for pre-built JWTs, or provide your own
+                callback for workload identity federation.
+            scopes: Optional space-separated list of scopes to request.
+        """
+        # Build minimal client_metadata for the base class
+        client_metadata = OAuthClientMetadata(
+            redirect_uris=None,
+            grant_types=["client_credentials"],
+            token_endpoint_auth_method="private_key_jwt",
+            scope=scopes,
+        )
+        super().__init__(server_url, client_metadata, storage, None, None, 300.0)
+        self._assertion_provider = assertion_provider
+        # Store client_info to be set during _initialize - no dynamic registration needed
+        self._fixed_client_info = OAuthClientInformationFull(
+            redirect_uris=None,
+            client_id=client_id,
+            grant_types=["client_credentials"],
+            token_endpoint_auth_method="private_key_jwt",
+            scope=scopes,
+        )
+
+    async def _initialize(self) -> None:
+        """Load stored tokens and set pre-configured client_info."""
+        self.context.current_tokens = await self.context.storage.get_tokens()
+        self.context.client_info = self._fixed_client_info
+        self._initialized = True
+
+    async def _perform_authorization(self) -> httpx.Request:
+        """Perform client_credentials authorization with private_key_jwt."""
+        return await self._exchange_token_client_credentials()
+
+    async def _add_client_authentication_jwt(self, *, token_data: dict[str, Any]) -> None:
+        """Add JWT assertion for client authentication to token endpoint parameters."""
+        if not self.context.oauth_metadata:
+            raise OAuthFlowError("Missing OAuth metadata for private_key_jwt flow")  # pragma: no cover
+
+        # Audience MUST be the issuer identifier of the authorization server
+        # https://datatracker.ietf.org/doc/html/draft-ietf-oauth-rfc7523bis-01
+        audience = str(self.context.oauth_metadata.issuer)
+        assertion = await self._assertion_provider(audience)
+
+        # RFC 7523 Section 2.2: client authentication via JWT
+        token_data["client_assertion"] = assertion
+        token_data["client_assertion_type"] = "urn:ietf:params:oauth:client-assertion-type:jwt-bearer"
+
+    async def _exchange_token_client_credentials(self) -> httpx.Request:
+        """Build token exchange request for client_credentials grant with private_key_jwt."""
+        token_data: dict[str, Any] = {
+            "grant_type": "client_credentials",
+        }
+
+        headers: dict[str, str] = {"Content-Type": "application/x-www-form-urlencoded"}
+
+        # Add JWT client authentication (RFC 7523 Section 2.2)
+        await self._add_client_authentication_jwt(token_data=token_data)
+
+        if self.context.should_include_resource_param(self.context.protocol_version):
+            token_data["resource"] = self.context.get_resource_url()
+
+        if self.context.client_metadata.scope:
+            token_data["scope"] = self.context.client_metadata.scope
+
+        token_url = self._get_token_endpoint()
+        return httpx.Request("POST", token_url, data=token_data, headers=headers)
 
 
 class JWTParameters(BaseModel):
@@ -64,9 +388,16 @@ class JWTParameters(BaseModel):
 
 
 class RFC7523OAuthClientProvider(OAuthClientProvider):
-    """OAuth client provider for RFC7532 clients."""
+    """OAuth client provider for RFC 7523 jwt-bearer grant.
 
-    jwt_parameters: JWTParameters | None = None
+    .. deprecated::
+        Use :class:`ClientCredentialsOAuthProvider` for client_credentials with
+        client_id + client_secret, or :class:`PrivateKeyJWTOAuthProvider` for
+        client_credentials with private_key_jwt authentication instead.
+
+    This provider supports the jwt-bearer authorization grant (RFC 7523 Section 2.1)
+    where the JWT itself is the authorization grant.
+    """
 
     def __init__(
         self,
@@ -78,6 +409,14 @@ class RFC7523OAuthClientProvider(OAuthClientProvider):
         timeout: float = 300.0,
         jwt_parameters: JWTParameters | None = None,
     ) -> None:
+        import warnings
+
+        warnings.warn(
+            "RFC7523OAuthClientProvider is deprecated. Use ClientCredentialsOAuthProvider "
+            "or PrivateKeyJWTOAuthProvider instead.",
+            DeprecationWarning,
+            stacklevel=2,
+        )
         super().__init__(server_url, client_metadata, storage, redirect_handler, callback_handler, timeout)
         self.jwt_parameters = jwt_parameters
 

tests/client/auth/extensions/test_client_credentials.py

@@ -4,7 +4,14 @@ import jwt
 import pytest
 from pydantic import AnyHttpUrl, AnyUrl
 
-from mcp.client.auth.extensions.client_credentials import JWTParameters, RFC7523OAuthClientProvider
+from mcp.client.auth.extensions.client_credentials import (
+    ClientCredentialsOAuthProvider,
+    JWTParameters,
+    PrivateKeyJWTOAuthProvider,
+    RFC7523OAuthClientProvider,
+    SignedJWTParameters,
+    static_assertion_provider,
+)
 from mcp.shared.auth import OAuthClientInformationFull, OAuthClientMetadata, OAuthMetadata, OAuthToken
 
 
@@ -53,13 +60,17 @@ def rfc7523_oauth_provider(client_metadata: OAuthClientMetadata, mock_storage: M
         """Mock callback handler."""
         return "test_auth_code", "test_state"
 
-    return RFC7523OAuthClientProvider(
+    import warnings
-        server_url="https://api.example.com/v1/mcp",
+
-        client_metadata=client_metadata,
+    with warnings.catch_warnings():
-        storage=mock_storage,
+        warnings.simplefilter("ignore", DeprecationWarning)
-        redirect_handler=redirect_handler,
+        return RFC7523OAuthClientProvider(
-        callback_handler=callback_handler,
+            server_url="https://api.example.com/v1/mcp",
-    )
+            client_metadata=client_metadata,
+            storage=mock_storage,
+            redirect_handler=redirect_handler,
+            callback_handler=callback_handler,
+        )
 
 
 class TestOAuthFlowClientCredentials:
@@ -161,3 +172,260 @@ class TestOAuthFlowClientCredentials:
         assert claims["name"] == "John Doe"
         assert claims["admin"]
         assert claims["iat"] == 1516239022
+
+
+class TestClientCredentialsOAuthProvider:
+    """Test ClientCredentialsOAuthProvider."""
+
+    @pytest.mark.anyio
+    async def test_init_sets_client_info(self, mock_storage: MockTokenStorage):
+        """Test that _initialize sets client_info."""
+        provider = ClientCredentialsOAuthProvider(
+            server_url="https://api.example.com",
+            storage=mock_storage,
+            client_id="test-client-id",
+            client_secret="test-client-secret",
+        )
+
+        # client_info is set during _initialize
+        await provider._initialize()
+
+        assert provider.context.client_info is not None
+        assert provider.context.client_info.client_id == "test-client-id"
+        assert provider.context.client_info.client_secret == "test-client-secret"
+        assert provider.context.client_info.grant_types == ["client_credentials"]
+        assert provider.context.client_info.token_endpoint_auth_method == "client_secret_basic"
+
+    @pytest.mark.anyio
+    async def test_init_with_scopes(self, mock_storage: MockTokenStorage):
+        """Test that constructor accepts scopes."""
+        provider = ClientCredentialsOAuthProvider(
+            server_url="https://api.example.com",
+            storage=mock_storage,
+            client_id="test-client-id",
+            client_secret="test-client-secret",
+            scopes="read write",
+        )
+
+        await provider._initialize()
+        assert provider.context.client_info is not None
+        assert provider.context.client_info.scope == "read write"
+
+    @pytest.mark.anyio
+    async def test_init_with_client_secret_post(self, mock_storage: MockTokenStorage):
+        """Test that constructor accepts client_secret_post auth method."""
+        provider = ClientCredentialsOAuthProvider(
+            server_url="https://api.example.com",
+            storage=mock_storage,
+            client_id="test-client-id",
+            client_secret="test-client-secret",
+            token_endpoint_auth_method="client_secret_post",
+        )
+
+        await provider._initialize()
+        assert provider.context.client_info is not None
+        assert provider.context.client_info.token_endpoint_auth_method == "client_secret_post"
+
+    @pytest.mark.anyio
+    async def test_exchange_token_client_credentials(self, mock_storage: MockTokenStorage):
+        """Test token exchange request building."""
+        provider = ClientCredentialsOAuthProvider(
+            server_url="https://api.example.com/v1/mcp",
+            storage=mock_storage,
+            client_id="test-client-id",
+            client_secret="test-client-secret",
+            scopes="read write",
+        )
+        provider.context.oauth_metadata = OAuthMetadata(
+            issuer=AnyHttpUrl("https://api.example.com"),
+            authorization_endpoint=AnyHttpUrl("https://api.example.com/authorize"),
+            token_endpoint=AnyHttpUrl("https://api.example.com/token"),
+        )
+        provider.context.protocol_version = "2025-06-18"
+
+        request = await provider._perform_authorization()
+
+        assert request.method == "POST"
+        assert str(request.url) == "https://api.example.com/token"
+
+        content = urllib.parse.unquote_plus(request.content.decode())
+        assert "grant_type=client_credentials" in content
+        assert "scope=read write" in content
+        assert "resource=https://api.example.com/v1/mcp" in content
+
+    @pytest.mark.anyio
+    async def test_exchange_token_without_scopes(self, mock_storage: MockTokenStorage):
+        """Test token exchange without scopes."""
+        provider = ClientCredentialsOAuthProvider(
+            server_url="https://api.example.com/v1/mcp",
+            storage=mock_storage,
+            client_id="test-client-id",
+            client_secret="test-client-secret",
+        )
+        provider.context.oauth_metadata = OAuthMetadata(
+            issuer=AnyHttpUrl("https://api.example.com"),
+            authorization_endpoint=AnyHttpUrl("https://api.example.com/authorize"),
+            token_endpoint=AnyHttpUrl("https://api.example.com/token"),
+        )
+        provider.context.protocol_version = "2024-11-05"  # Old version - no resource param
+
+        request = await provider._perform_authorization()
+
+        content = urllib.parse.unquote_plus(request.content.decode())
+        assert "grant_type=client_credentials" in content
+        assert "scope=" not in content
+        assert "resource=" not in content
+
+
+class TestPrivateKeyJWTOAuthProvider:
+    """Test PrivateKeyJWTOAuthProvider."""
+
+    @pytest.mark.anyio
+    async def test_init_sets_client_info(self, mock_storage: MockTokenStorage):
+        """Test that _initialize sets client_info."""
+
+        async def mock_assertion_provider(audience: str) -> str:  # pragma: no cover
+            return "mock-jwt"
+
+        provider = PrivateKeyJWTOAuthProvider(
+            server_url="https://api.example.com",
+            storage=mock_storage,
+            client_id="test-client-id",
+            assertion_provider=mock_assertion_provider,
+        )
+
+        # client_info is set during _initialize
+        await provider._initialize()
+
+        assert provider.context.client_info is not None
+        assert provider.context.client_info.client_id == "test-client-id"
+        assert provider.context.client_info.grant_types == ["client_credentials"]
+        assert provider.context.client_info.token_endpoint_auth_method == "private_key_jwt"
+
+    @pytest.mark.anyio
+    async def test_exchange_token_client_credentials(self, mock_storage: MockTokenStorage):
+        """Test token exchange request building with assertion provider."""
+
+        async def mock_assertion_provider(audience: str) -> str:
+            return f"jwt-for-{audience}"
+
+        provider = PrivateKeyJWTOAuthProvider(
+            server_url="https://api.example.com/v1/mcp",
+            storage=mock_storage,
+            client_id="test-client-id",
+            assertion_provider=mock_assertion_provider,
+            scopes="read write",
+        )
+        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"),
+        )
+        provider.context.protocol_version = "2025-06-18"
+
+        request = await provider._perform_authorization()
+
+        assert request.method == "POST"
+        assert str(request.url) == "https://auth.example.com/token"
+
+        content = urllib.parse.unquote_plus(request.content.decode())
+        assert "grant_type=client_credentials" in content
+        assert "client_assertion=jwt-for-https://auth.example.com/" in content
+        assert "client_assertion_type=urn:ietf:params:oauth:client-assertion-type:jwt-bearer" in content
+        assert "scope=read write" in content
+
+    @pytest.mark.anyio
+    async def test_exchange_token_without_scopes(self, mock_storage: MockTokenStorage):
+        """Test token exchange without scopes."""
+
+        async def mock_assertion_provider(audience: str) -> str:
+            return f"jwt-for-{audience}"
+
+        provider = PrivateKeyJWTOAuthProvider(
+            server_url="https://api.example.com/v1/mcp",
+            storage=mock_storage,
+            client_id="test-client-id",
+            assertion_provider=mock_assertion_provider,
+        )
+        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"),
+        )
+        provider.context.protocol_version = "2024-11-05"  # Old version - no resource param
+
+        request = await provider._perform_authorization()
+
+        content = urllib.parse.unquote_plus(request.content.decode())
+        assert "grant_type=client_credentials" in content
+        assert "scope=" not in content
+        assert "resource=" not in content
+
+
+class TestSignedJWTParameters:
+    """Test SignedJWTParameters."""
+
+    @pytest.mark.anyio
+    async def test_create_assertion_provider(self):
+        """Test that create_assertion_provider creates valid JWTs."""
+        params = SignedJWTParameters(
+            issuer="test-issuer",
+            subject="test-subject",
+            signing_key="a-string-secret-at-least-256-bits-long",
+            signing_algorithm="HS256",
+            lifetime_seconds=300,
+        )
+
+        provider = params.create_assertion_provider()
+        assertion = await provider("https://auth.example.com")
+
+        claims = jwt.decode(
+            assertion,
+            key="a-string-secret-at-least-256-bits-long",
+            algorithms=["HS256"],
+            audience="https://auth.example.com",
+        )
+        assert claims["iss"] == "test-issuer"
+        assert claims["sub"] == "test-subject"
+        assert claims["aud"] == "https://auth.example.com"
+        assert "exp" in claims
+        assert "iat" in claims
+        assert "jti" in claims
+
+    @pytest.mark.anyio
+    async def test_create_assertion_provider_with_additional_claims(self):
+        """Test that additional_claims are included in the JWT."""
+        params = SignedJWTParameters(
+            issuer="test-issuer",
+            subject="test-subject",
+            signing_key="a-string-secret-at-least-256-bits-long",
+            signing_algorithm="HS256",
+            additional_claims={"custom": "value"},
+        )
+
+        provider = params.create_assertion_provider()
+        assertion = await provider("https://auth.example.com")
+
+        claims = jwt.decode(
+            assertion,
+            key="a-string-secret-at-least-256-bits-long",
+            algorithms=["HS256"],
+            audience="https://auth.example.com",
+        )
+        assert claims["custom"] == "value"
+
+
+class TestStaticAssertionProvider:
+    """Test static_assertion_provider helper."""
+
+    @pytest.mark.anyio
+    async def test_returns_static_token(self):
+        """Test that static_assertion_provider returns the same token regardless of audience."""
+        token = "my-static-jwt-token"
+        provider = static_assertion_provider(token)
+
+        result1 = await provider("https://auth1.example.com")
+        result2 = await provider("https://auth2.example.com")
+
+        assert result1 == token
+        assert result2 == token