Client Authentication

private_key_jwt client authentication (RFC 7523) for OAuth 2.0 token endpoints. Clients sign a JWT with their private key and submit it as a client_assertion; JWTClientAuthenticator verifies it against a public key resolved through a pluggable JWKSProvider, with an asymmetric-only algorithm allowlist and jti replay protection.

For a shared replay cache, pass any object satisfying mcp_authflow.client_auth.AsyncRedisClient. This protocol requires only the Redis SET operation used by replay protection; it is intentionally separate from the sorted-set protocol used by the rate limiter.

client_auth

Client authentication primitives for OAuth 2.0 token endpoints.

Currently provides private_key_jwt (RFC 7523) verification. The package exposes the high-level :class:JWTClientAuthenticator, the :class:JWKSProvider integration point, and the algorithm allowlist / blocklist constants for callers that need them.

AsyncRedisClient

Bases: Protocol

Minimal async Redis interface used for JTI replay-cache storage.

JWKSProvider

Bases: Protocol

Resolve a client's JWKS by client_id.

Implementations choose how to look up the key material — static dict, Dynamic Client Registration record, Client ID Metadata Document, etc. Returning None signals "no keys available" and causes authentication to fail with :class:JWTAuthError.

JWTAuthError

Bases: Exception

Raised when private_key_jwt client authentication fails.

JWTClientAuthenticator

JWTClientAuthenticator(
    token_endpoint: str,
    jwks_provider: JWKSProvider,
    redis: AsyncRedisClient | None = None,
)

Verify private_key_jwt client assertions per RFC 7523.

Parameters:

Name	Type	Description	Default
token_endpoint	str	The token endpoint URL, used as the expected aud claim per RFC 7523 section 3.	required
jwks_provider	JWKSProvider	Resolves the client's JWKS by client_id.	required
redis	AsyncRedisClient | None	Optional async Redis client for a persistent / shared JTI replay cache. When None, an in-memory cache with periodic TTL cleanup is used.	None

Source code in mcp_authflow/client_auth/jwt.py

def __init__(
    self,
    token_endpoint: str,
    jwks_provider: JWKSProvider,
    redis: AsyncRedisClient | None = None,
) -> None:
    self.token_endpoint = token_endpoint
    self.jwks_provider = jwks_provider
    self._redis = redis

    self._used_jtis: dict[str, float] = {}
    self._jti_lock = threading.Lock()
    self._last_cleanup = time.time()

authenticate async

authenticate(
    client_id: str,
    client_assertion: str,
    client_assertion_type: str,
) -> bool

Authenticate a client using private_key_jwt.

Returns True on success and raises :class:JWTAuthError otherwise. Rejections are logged at WARNING before the error propagates, so an operator watching this logger sees failed attempts (replay detection, blocked algorithms, ...) and not just successes.

Source code in mcp_authflow/client_auth/jwt.py

async def authenticate(
    self,
    client_id: str,
    client_assertion: str,
    client_assertion_type: str,
) -> bool:
    """Authenticate a client using private_key_jwt.

    Returns ``True`` on success and raises :class:`JWTAuthError` otherwise.
    Rejections are logged at WARNING before the error propagates, so an
    operator watching this logger sees failed attempts (replay detection,
    blocked algorithms, ...) and not just successes.
    """
    try:
        return await self._authenticate(client_id, client_assertion, client_assertion_type)
    except JWTAuthError as e:
        # Only the error message is logged — never the assertion itself.
        logger.warning("private_key_jwt authentication failed for client %s: %s", client_id, e)
        raise