Add Clerk OAuth provider (#3677)

examples/auth/clerk_oauth/README.md

@@ -0,0 +1,36 @@
+# Clerk OAuth Example
+
+Demonstrates FastMCP server protection with Clerk OAuth.
+
+## Setup
+
+1. Create a Clerk OAuth Application:
+   - Go to [Clerk Dashboard](https://dashboard.clerk.com/)
+   - Create or select an application
+   - Go to Developers > OAuth Applications
+   - Create an OAuth application
+   - Add Authorized redirect URI: `http://localhost:8000/auth/callback`
+   - Copy the Client ID and Client Secret
+   - Note your instance domain (e.g., `saving-primate-16.clerk.accounts.dev`)
+
+2. Set environment variables:
+
+   ```bash
+   export FASTMCP_SERVER_AUTH_CLERK_DOMAIN="your-instance.clerk.accounts.dev"
+   export FASTMCP_SERVER_AUTH_CLERK_CLIENT_ID="your-clerk-client-id"
+   export FASTMCP_SERVER_AUTH_CLERK_CLIENT_SECRET="your-clerk-client-secret"
+   ```
+
+3. Run the server:
+
+   ```bash
+   python server.py
+   ```
+
+4. In another terminal, run the client:
+
+   ```bash
+   python client.py
+   ```
+
+The client will open your browser for Clerk authentication.

examples/auth/clerk_oauth/client.py

@@ -0,0 +1,33 @@
+"""OAuth client example for connecting to a Clerk-protected FastMCP server.
+
+This example demonstrates how to connect to an OAuth-protected FastMCP server
+using Clerk as the identity provider.
+
+To run:
+    python client.py
+"""
+
+import asyncio
+
+from fastmcp.client import Client
+
+SERVER_URL = "http://127.0.0.1:8000/mcp"
+
+
+async def main():
+    try:
+        async with Client(SERVER_URL, auth="oauth") as client:
+            assert await client.ping()
+            print("✅ Successfully authenticated!")
+
+            tools = await client.list_tools()
+            print(f"🔧 Available tools ({len(tools)}):")
+            for tool in tools:
+                print(f"   - {tool.name}: {tool.description}")
+    except Exception as e:
+        print(f"❌ Authentication failed: {e}")
+        raise
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

examples/auth/clerk_oauth/server.py

@@ -0,0 +1,40 @@
+"""Clerk OAuth server example for FastMCP.
+
+This example demonstrates how to protect a FastMCP server with Clerk OAuth.
+
+Required environment variables:
+- FASTMCP_SERVER_AUTH_CLERK_DOMAIN: Your Clerk instance domain
+    (e.g., "saving-primate-16.clerk.accounts.dev")
+- FASTMCP_SERVER_AUTH_CLERK_CLIENT_ID: Your Clerk OAuth client ID
+- FASTMCP_SERVER_AUTH_CLERK_CLIENT_SECRET: Your Clerk OAuth client secret
+
+To run:
+    python server.py
+"""
+
+import os
+
+from fastmcp import FastMCP
+from fastmcp.server.auth.providers.clerk import ClerkProvider
+
+auth = ClerkProvider(
+    domain=os.getenv("FASTMCP_SERVER_AUTH_CLERK_DOMAIN") or "",
+    client_id=os.getenv("FASTMCP_SERVER_AUTH_CLERK_CLIENT_ID") or "",
+    client_secret=os.getenv("FASTMCP_SERVER_AUTH_CLERK_CLIENT_SECRET") or "",
+    base_url="http://localhost:8000",
+    # redirect_path="/auth/callback",  # Default path - change if using a different callback URL
+    # Optional: specify required scopes (defaults to ["openid", "email", "profile"])
+    # required_scopes=["openid", "email", "profile", "public_metadata"],
+)
+
+mcp = FastMCP("Clerk OAuth Example Server", auth=auth)
+
+
+@mcp.tool
+def echo(message: str) -> str:
+    """Echo the provided message."""
+    return message
+
+
+if __name__ == "__main__":
+    mcp.run(transport="http", port=8000)

src/fastmcp/server/auth/providers/clerk.py

@@ -0,0 +1,382 @@
+"""Clerk OAuth provider for FastMCP.
+
+This module provides a complete Clerk OAuth integration that's ready to use
+with a Clerk domain, client ID, and client secret. It handles all the complexity
+of Clerk's OAuth/OIDC flow, token validation, and user management.
+
+Clerk uses standard OIDC endpoints derived from the instance domain
+(e.g., ``https://<instance>.clerk.accounts.dev``). Token verification is
+performed via the introspection endpoint (RFC 7662) for security-critical
+checks (active status, audience, scopes), followed by the userinfo endpoint
+for profile enrichment. Userinfo failure is non-fatal.
+
+Example:
+    ```python
+    from fastmcp import FastMCP
+    from fastmcp.server.auth.providers.clerk import ClerkProvider
+
+    auth = ClerkProvider(
+        domain="saving-primate-16.clerk.accounts.dev",
+        client_id="your-clerk-client-id",
+        client_secret="your-clerk-client-secret",
+        base_url="https://my-server.com",
+    )
+
+    mcp = FastMCP("My Protected Server", auth=auth)
+    ```
+"""
+
+from __future__ import annotations
+
+import contextlib
+from typing import Literal
+
+import httpx
+from key_value.aio.protocols import AsyncKeyValue
+from pydantic import AnyHttpUrl
+
+from fastmcp.server.auth import TokenVerifier
+from fastmcp.server.auth.auth import AccessToken
+from fastmcp.server.auth.oauth_proxy import OAuthProxy
+from fastmcp.utilities.auth import parse_scopes
+from fastmcp.utilities.logging import get_logger
+
+logger = get_logger(__name__)
+
+
+class ClerkTokenVerifier(TokenVerifier):
+    """Token verifier for Clerk OAuth tokens.
+
+    Clerk issues standard OIDC tokens. Verification uses the introspection
+    endpoint (RFC 7662) as the primary security gate — it confirms the token
+    is active and provides metadata (scopes, expiry, audience). The userinfo
+    endpoint is called second for profile enrichment (name, email, picture)
+    and its failure is non-fatal.
+
+    When a ``client_id`` is configured, the audience from introspection is
+    validated against it. When ``required_scopes`` are configured,
+    introspection must return the token's scopes — the verifier will not
+    assume scopes when introspection is unavailable.
+    """
+
+    def __init__(
+        self,
+        *,
+        domain: str,
+        client_id: str | None = None,
+        client_secret: str | None = None,
+        required_scopes: list[str] | None = None,
+        timeout_seconds: int = 10,
+        http_client: httpx.AsyncClient | None = None,
+    ):
+        """Initialize the Clerk token verifier.
+
+        Args:
+            domain: Clerk instance domain (e.g., "saving-primate-16.clerk.accounts.dev")
+            client_id: Clerk OAuth client ID, used for introspection endpoint authentication
+            client_secret: Clerk OAuth client secret, used for introspection endpoint authentication
+            required_scopes: Required OAuth scopes (e.g., ["openid", "email", "profile"])
+            timeout_seconds: HTTP request timeout
+            http_client: Optional httpx.AsyncClient for connection pooling. When provided,
+                the client is reused across calls and the caller is responsible for its
+                lifecycle. When None (default), a fresh client is created per call.
+        """
+        super().__init__(required_scopes=required_scopes)
+        self.domain = domain.rstrip("/")
+        self._client_id = client_id
+        self._client_secret = client_secret
+        self.timeout_seconds = timeout_seconds
+        self._http_client = http_client
+
+        self._userinfo_url = f"https://{self.domain}/oauth/userinfo"
+        self._introspection_url = f"https://{self.domain}/oauth/token_info"
+
+    async def verify_token(self, token: str) -> AccessToken | None:
+        """Verify a Clerk OAuth token via introspection and userinfo.
+
+        Calls the introspection endpoint first to validate the token and
+        retrieve auth metadata (active status, scopes, expiry, audience).
+        If the token passes security checks, the userinfo endpoint is called
+        for profile enrichment. Userinfo failure is non-fatal.
+
+        When a ``client_id`` is configured, the token's audience must match it.
+        When ``required_scopes`` are configured, introspection must confirm
+        them; tokens are rejected if scope information is unavailable.
+        """
+        try:
+            async with (
+                contextlib.nullcontext(self._http_client)
+                if self._http_client is not None
+                else httpx.AsyncClient(timeout=self.timeout_seconds)
+            ) as client:
+                # Step 1: Validate token via introspection (RFC 7662).
+                # Security-critical checks (active, audience, scopes) come first.
+                introspect_data_payload: dict = {"token": token}
+                introspect_kwargs: dict = {
+                    "data": introspect_data_payload,
+                    "headers": {"User-Agent": "FastMCP-Clerk-OAuth"},
+                }
+
+                if self._client_id and self._client_secret:
+                    introspect_kwargs["auth"] = (
+                        self._client_id,
+                        self._client_secret,
+                    )
+                elif self._client_id:
+                    introspect_data_payload["client_id"] = self._client_id
+
+                introspect_response = await client.post(
+                    self._introspection_url,
+                    **introspect_kwargs,
+                )
+
+                if introspect_response.status_code != 200:
+                    logger.debug(
+                        "Clerk introspection failed: %d",
+                        introspect_response.status_code,
+                    )
+                    return None
+
+                introspect_data = introspect_response.json()
+
+                # RFC 7662 requires the 'active' field in the response.
+                # A missing field indicates a malformed response — reject.
+                if "active" not in introspect_data or not introspect_data["active"]:
+                    logger.debug(
+                        "Clerk introspection: token inactive or missing 'active' field"
+                    )
+                    return None
+
+                scope_str = introspect_data.get("scope", "")
+                token_scopes = scope_str.split() if scope_str else []
+
+                aud = introspect_data.get("aud") or introspect_data.get("client_id")
+
+                expires_at: int | None = None
+                exp = introspect_data.get("exp")
+                if exp is not None:
+                    with contextlib.suppress(ValueError, TypeError):
+                        expires_at = int(exp)
+
+                if self._client_id and aud != self._client_id:
+                    logger.debug(
+                        "Clerk token audience mismatch: got %s, expected %s",
+                        aud,
+                        self._client_id,
+                    )
+                    return None
+
+                if self.required_scopes:
+                    if not token_scopes:
+                        logger.debug(
+                            "Clerk token missing scope information; "
+                            "cannot verify required scopes %s",
+                            self.required_scopes,
+                        )
+                        return None
+                    token_scopes_set = set(token_scopes)
+                    required_scopes_set = set(self.required_scopes)
+                    if not required_scopes_set.issubset(token_scopes_set):
+                        logger.debug(
+                            "Clerk token missing required scopes. Has %s, needs %s",
+                            token_scopes_set,
+                            required_scopes_set,
+                        )
+                        return None
+
+                # Step 2: Fetch user profile via userinfo.
+                # Enriches the token with profile data (name, email, picture).
+                sub = introspect_data.get("sub")
+                user_data: dict = {}
+                try:
+                    userinfo_response = await client.get(
+                        self._userinfo_url,
+                        headers={
+                            "Authorization": f"Bearer {token}",
+                            "User-Agent": "FastMCP-Clerk-OAuth",
+                        },
+                    )
+                    if userinfo_response.status_code == 200:
+                        user_data = userinfo_response.json()
+                        if not sub:
+                            sub = user_data.get("sub")
+                except Exception as e:
+                    logger.debug("Clerk userinfo call failed: %s", e)
+
+                if not sub:
+                    logger.debug("Clerk token missing 'sub' claim")
+                    return None
+
+                access_token = AccessToken(
+                    token=token,
+                    client_id=aud or sub,
+                    scopes=token_scopes,
+                    expires_at=expires_at,
+                    claims={
+                        "sub": sub,
+                        "aud": aud,
+                        "email": user_data.get("email"),
+                        "email_verified": user_data.get("email_verified"),
+                        "name": user_data.get("name"),
+                        "picture": user_data.get("picture"),
+                        "given_name": user_data.get("given_name"),
+                        "family_name": user_data.get("family_name"),
+                        "preferred_username": user_data.get("preferred_username"),
+                        "iss": user_data.get("iss"),
+                        "clerk_user_data": user_data or None,
+                    },
+                )
+                logger.debug("Clerk token verified successfully for sub=%s", sub)
+                return access_token
+
+        except httpx.RequestError as e:
+            logger.debug("Failed to verify Clerk token: %s", e)
+            return None
+        except Exception as e:
+            logger.debug("Clerk token verification error: %s", e)
+            return None
+
+
+class ClerkProvider(OAuthProxy):
+    """Complete Clerk OAuth provider for FastMCP.
+
+    This provider makes it trivial to add Clerk OAuth protection to any
+    FastMCP server. Provide your Clerk instance domain, OAuth app credentials,
+    and a base URL, and you're ready to go.
+
+    Clerk uses standard OIDC endpoints derived from the instance domain.
+    All endpoint URLs are constructed automatically from the domain parameter.
+
+    Features:
+    - Transparent OAuth proxy to Clerk
+    - Automatic token validation via Clerk's userinfo & introspection APIs
+    - User information extraction from Clerk's OIDC claims
+    - PKCE support (S256)
+    - Minimal configuration required
+
+    Example:
+        ```python
+        from fastmcp import FastMCP
+        from fastmcp.server.auth.providers.clerk import ClerkProvider
+
+        auth = ClerkProvider(
+            domain="saving-primate-16.clerk.accounts.dev",
+            client_id="your-clerk-client-id",
+            client_secret="your-clerk-client-secret",
+            base_url="https://my-server.com",
+        )
+
+        mcp = FastMCP("My App", auth=auth)
+        ```
+    """
+
+    def __init__(
+        self,
+        *,
+        domain: str,
+        client_id: str,
+        client_secret: str | None = None,
+        base_url: AnyHttpUrl | str,
+        issuer_url: AnyHttpUrl | str | None = None,
+        redirect_path: str | None = None,
+        required_scopes: list[str] | None = None,
+        valid_scopes: list[str] | None = None,
+        timeout_seconds: int = 10,
+        allowed_client_redirect_uris: list[str] | None = None,
+        client_storage: AsyncKeyValue | None = None,
+        jwt_signing_key: str | bytes | None = None,
+        require_authorization_consent: bool | Literal["external"] = True,
+        consent_csp_policy: str | None = None,
+        extra_authorize_params: dict[str, str] | None = None,
+        http_client: httpx.AsyncClient | None = None,
+        enable_cimd: bool = True,
+    ):
+        """Initialize Clerk OAuth provider.
+
+        Args:
+            domain: Clerk instance domain (e.g., "saving-primate-16.clerk.accounts.dev").
+                This is used to derive all OAuth/OIDC endpoint URLs.
+            client_id: Clerk OAuth application client ID
+            client_secret: Clerk OAuth application client secret.
+                Optional for PKCE public clients. When omitted, jwt_signing_key must be provided.
+            base_url: Public URL where OAuth endpoints will be accessible (includes any mount path)
+            issuer_url: Issuer URL for OAuth metadata (defaults to base_url). Use root-level URL
+                to avoid 404s during discovery when mounting under a path.
+            redirect_path: Redirect path configured in Clerk OAuth app (defaults to "/auth/callback")
+            required_scopes: Required Clerk scopes (defaults to ["openid", "email", "profile"]).
+                Clerk supports: "openid", "email", "profile", "public_metadata",
+                "private_metadata", "offline_access".
+            valid_scopes: All scopes that clients are allowed to request, advertised through
+                well-known endpoints. Defaults to required_scopes if not provided.
+            timeout_seconds: HTTP request timeout for Clerk API calls (defaults to 10)
+            allowed_client_redirect_uris: List of allowed redirect URI patterns for MCP clients.
+                If None (default), all URIs are allowed. If empty list, no URIs are allowed.
+            client_storage: Storage backend for OAuth state (client registrations, encrypted tokens).
+                If None, an encrypted file store will be created in the data directory
+                (derived from ``platformdirs``).
+            jwt_signing_key: Secret for signing FastMCP JWT tokens (any string or bytes). If bytes
+                are provided, they will be used as is. If a string is provided, it will be derived
+                into a 32-byte key. If not provided, the upstream client secret will be used to
+                derive a 32-byte key using PBKDF2.
+            require_authorization_consent: Whether to require user consent before authorizing
+                clients (default True). When "external", the built-in consent screen is skipped
+                but no warning is logged, indicating that consent is handled externally by Clerk.
+            consent_csp_policy: Custom CSP policy for the consent page.
+            extra_authorize_params: Additional parameters to forward to Clerk's authorization
+                endpoint. Example: {"prompt": "login"} to force re-authentication.
+            http_client: Optional httpx.AsyncClient for connection pooling in token verification.
+                When provided, the client is reused across verify_token calls and the caller
+                is responsible for its lifecycle. When None (default), a fresh client is created
+                per call.
+            enable_cimd: Enable CIMD (Client ID Metadata Document) support for URL-based
+                client IDs (default True). Set to False to disable.
+        """
+        domain = domain.rstrip("/")
+
+        required_scopes_final = (
+            parse_scopes(required_scopes)
+            if required_scopes is not None
+            else ["openid", "email", "profile"]
+        )
+
+        parsed_valid_scopes = (
+            parse_scopes(valid_scopes) if valid_scopes is not None else None
+        )
+
+        token_verifier = ClerkTokenVerifier(
+            domain=domain,
+            client_id=client_id,
+            client_secret=client_secret,
+            required_scopes=required_scopes_final,
+            timeout_seconds=timeout_seconds,
+            http_client=http_client,
+        )
+
+        extra_authorize_params_final = (
+            dict(extra_authorize_params) if extra_authorize_params else {}
+        )
+
+        super().__init__(
+            upstream_authorization_endpoint=f"https://{domain}/oauth/authorize",
+            upstream_token_endpoint=f"https://{domain}/oauth/token",
+            upstream_client_id=client_id,
+            upstream_client_secret=client_secret,
+            token_verifier=token_verifier,
+            base_url=base_url,
+            redirect_path=redirect_path,
+            issuer_url=issuer_url or base_url,
+            allowed_client_redirect_uris=allowed_client_redirect_uris,
+            client_storage=client_storage,
+            jwt_signing_key=jwt_signing_key,
+            require_authorization_consent=require_authorization_consent,
+            consent_csp_policy=consent_csp_policy,
+            extra_authorize_params=extra_authorize_params_final or None,
+            valid_scopes=parsed_valid_scopes,
+            enable_cimd=enable_cimd,
+        )
+
+        logger.debug(
+            "Initialized Clerk OAuth provider for domain %s with scopes: %s",
+            domain,
+            required_scopes_final,
+        )

tests/server/auth/providers/test_clerk.py

@@ -0,0 +1,572 @@
+"""Tests for Clerk OAuth provider."""
+
+import re
+
+import httpx
+import pytest
+from key_value.aio.stores.memory import MemoryStore
+from pytest_httpx import HTTPXMock
+
+from fastmcp.server.auth.providers.clerk import ClerkProvider, ClerkTokenVerifier
+
+CLERK_DOMAIN = "test-instance.clerk.accounts.dev"
+
+_USERINFO_RE = re.compile(rf"https://{re.escape(CLERK_DOMAIN)}/oauth/userinfo")
+_INTROSPECTION_RE = re.compile(rf"https://{re.escape(CLERK_DOMAIN)}/oauth/token_info")
+
+
+@pytest.fixture
+def memory_storage() -> MemoryStore:
+    """Provide a MemoryStore for tests to avoid SQLite initialization on Windows."""
+    return MemoryStore()
+
+
+class TestClerkProvider:
+    """Test Clerk OAuth provider functionality."""
+
+    def test_init_with_explicit_params(self, memory_storage: MemoryStore):
+        """Test ClerkProvider initialization with explicit parameters."""
+        provider = ClerkProvider(
+            domain=CLERK_DOMAIN,
+            client_id="clerk-client-id",
+            client_secret="clerk-client-secret",
+            base_url="https://myserver.com",
+            required_scopes=["openid", "email", "profile"],
+            jwt_signing_key="test-secret",
+            client_storage=memory_storage,
+        )
+
+        assert provider._upstream_client_id == "clerk-client-id"
+        assert provider._upstream_client_secret is not None
+        assert (
+            provider._upstream_client_secret.get_secret_value() == "clerk-client-secret"
+        )
+        assert str(provider.base_url) == "https://myserver.com/"
+
+    def test_init_defaults(self, memory_storage: MemoryStore):
+        """Test that default values are applied correctly."""
+        provider = ClerkProvider(
+            domain=CLERK_DOMAIN,
+            client_id="clerk-client-id",
+            client_secret="clerk-client-secret",
+            base_url="https://myserver.com",
+            jwt_signing_key="test-secret",
+            client_storage=memory_storage,
+        )
+
+        assert provider._redirect_path == "/auth/callback"
+
+    def test_oauth_endpoints_configured_correctly(self, memory_storage: MemoryStore):
+        """Test that OAuth endpoints are derived from the domain."""
+        provider = ClerkProvider(
+            domain=CLERK_DOMAIN,
+            client_id="clerk-client-id",
+            client_secret="clerk-client-secret",
+            base_url="https://myserver.com",
+            jwt_signing_key="test-secret",
+            client_storage=memory_storage,
+        )
+
+        assert (
+            provider._upstream_authorization_endpoint
+            == f"https://{CLERK_DOMAIN}/oauth/authorize"
+        )
+        assert (
+            provider._upstream_token_endpoint == f"https://{CLERK_DOMAIN}/oauth/token"
+        )
+        assert provider._upstream_revocation_endpoint is None
+
+    def test_domain_trailing_slash_stripped(self, memory_storage: MemoryStore):
+        """Test that trailing slashes are stripped from the domain."""
+        provider = ClerkProvider(
+            domain=f"{CLERK_DOMAIN}/",
+            client_id="clerk-client-id",
+            client_secret="clerk-client-secret",
+            base_url="https://myserver.com",
+            jwt_signing_key="test-secret",
+            client_storage=memory_storage,
+        )
+
+        assert (
+            provider._upstream_authorization_endpoint
+            == f"https://{CLERK_DOMAIN}/oauth/authorize"
+        )
+
+    def test_default_scopes(self, memory_storage: MemoryStore):
+        """Test that default required scopes are openid, email, profile."""
+        provider = ClerkProvider(
+            domain=CLERK_DOMAIN,
+            client_id="clerk-client-id",
+            client_secret="clerk-client-secret",
+            base_url="https://myserver.com",
+            jwt_signing_key="test-secret",
+            client_storage=memory_storage,
+        )
+
+        assert provider is not None
+
+    def test_custom_scopes(self, memory_storage: MemoryStore):
+        """Test that custom scopes are accepted."""
+        provider = ClerkProvider(
+            domain=CLERK_DOMAIN,
+            client_id="clerk-client-id",
+            client_secret="clerk-client-secret",
+            base_url="https://myserver.com",
+            required_scopes=["openid", "email", "profile", "public_metadata"],
+            jwt_signing_key="test-secret",
+            client_storage=memory_storage,
+        )
+
+        assert provider is not None
+
+    def test_no_extra_authorize_params_by_default(self, memory_storage: MemoryStore):
+        """Test that no extra authorize params are set by default."""
+        provider = ClerkProvider(
+            domain=CLERK_DOMAIN,
+            client_id="clerk-client-id",
+            client_secret="clerk-client-secret",
+            base_url="https://myserver.com",
+            jwt_signing_key="test-secret",
+            client_storage=memory_storage,
+        )
+
+        assert provider._extra_authorize_params in (None, {})
+
+    def test_extra_authorize_params_passed_through(self, memory_storage: MemoryStore):
+        """Test that extra authorize params are forwarded."""
+        provider = ClerkProvider(
+            domain=CLERK_DOMAIN,
+            client_id="clerk-client-id",
+            client_secret="clerk-client-secret",
+            base_url="https://myserver.com",
+            jwt_signing_key="test-secret",
+            extra_authorize_params={"prompt": "login"},
+            client_storage=memory_storage,
+        )
+
+        assert provider._extra_authorize_params == {"prompt": "login"}
+
+    def test_valid_scopes_passed_through(self, memory_storage: MemoryStore):
+        """Test that valid_scopes is passed to OAuthProxy."""
+        provider = ClerkProvider(
+            domain=CLERK_DOMAIN,
+            client_id="clerk-client-id",
+            client_secret="clerk-client-secret",
+            base_url="https://myserver.com",
+            required_scopes=["openid"],
+            valid_scopes=["openid", "email", "profile", "public_metadata"],
+            jwt_signing_key="test-secret",
+            client_storage=memory_storage,
+        )
+
+        reg_options = provider.client_registration_options
+        assert reg_options is not None
+        assert reg_options.valid_scopes is not None
+        assert set(reg_options.valid_scopes) == {
+            "openid",
+            "email",
+            "profile",
+            "public_metadata",
+        }
+
+    def test_issuer_url_defaults_to_base_url(self, memory_storage: MemoryStore):
+        """Test that issuer_url defaults to base_url when not provided."""
+        provider = ClerkProvider(
+            domain=CLERK_DOMAIN,
+            client_id="clerk-client-id",
+            client_secret="clerk-client-secret",
+            base_url="https://myserver.com",
+            jwt_signing_key="test-secret",
+            client_storage=memory_storage,
+        )
+
+        assert str(provider.issuer_url) == "https://myserver.com/"
+
+    def test_custom_issuer_url(self, memory_storage: MemoryStore):
+        """Test that a custom issuer_url is used when provided."""
+        provider = ClerkProvider(
+            domain=CLERK_DOMAIN,
+            client_id="clerk-client-id",
+            client_secret="clerk-client-secret",
+            base_url="https://myserver.com/mcp",
+            issuer_url="https://myserver.com",
+            jwt_signing_key="test-secret",
+            client_storage=memory_storage,
+        )
+
+        assert str(provider.issuer_url) == "https://myserver.com/"
+
+
+class TestClerkTokenVerifier:
+    """Test ClerkTokenVerifier.verify_token() using introspection + userinfo."""
+
+    async def test_valid_token_basic(self, httpx_mock: HTTPXMock):
+        """A valid token returns an AccessToken with user claims from userinfo."""
+        httpx_mock.add_response(
+            url=_USERINFO_RE,
+            json={
+                "sub": "user_abc123",
+                "email": "user@example.com",
+                "email_verified": True,
+                "name": "Test User",
+                "picture": "https://img.clerk.com/photo.jpg",
+                "given_name": "Test",
+                "family_name": "User",
+                "preferred_username": "testuser",
+                "iss": f"https://{CLERK_DOMAIN}",
+            },
+        )
+        httpx_mock.add_response(
+            url=_INTROSPECTION_RE,
+            json={
+                "active": True,
+                "scope": "openid email profile",
+                "aud": "clerk-client-id",
+                "exp": 9999999999,
+            },
+        )
+
+        verifier = ClerkTokenVerifier(
+            domain=CLERK_DOMAIN,
+            client_id="clerk-client-id",
+            client_secret="clerk-client-secret",
+        )
+        result = await verifier.verify_token("valid-token")
+
+        assert result is not None
+        assert result.client_id == "clerk-client-id"
+        assert result.scopes == ["openid", "email", "profile"]
+        assert result.expires_at == 9999999999
+        assert result.claims["sub"] == "user_abc123"
+        assert result.claims["email"] == "user@example.com"
+        assert result.claims["name"] == "Test User"
+        assert result.claims["picture"] == "https://img.clerk.com/photo.jpg"
+        assert result.claims["given_name"] == "Test"
+        assert result.claims["family_name"] == "User"
+        assert result.claims["preferred_username"] == "testuser"
+        assert result.claims["aud"] == "clerk-client-id"
+
+    async def test_invalid_token_returns_none(self, httpx_mock: HTTPXMock):
+        """Token marked inactive by introspection is rejected."""
+        httpx_mock.add_response(
+            url=_INTROSPECTION_RE,
+            json={"active": False},
+        )
+
+        verifier = ClerkTokenVerifier(domain=CLERK_DOMAIN)
+        result = await verifier.verify_token("expired-token")
+
+        assert result is None
+
+    async def test_missing_sub_returns_none(self, httpx_mock: HTTPXMock):
+        """Token with no 'sub' in introspection or userinfo is rejected."""
+        httpx_mock.add_response(
+            url=_INTROSPECTION_RE,
+            json={"active": True},
+        )
+        httpx_mock.add_response(
+            url=_USERINFO_RE,
+            json={"email": "user@example.com"},
+        )
+
+        verifier = ClerkTokenVerifier(domain=CLERK_DOMAIN)
+        result = await verifier.verify_token("token-without-sub")
+
+        assert result is None
+
+    async def test_introspection_inactive_token_returns_none(
+        self, httpx_mock: HTTPXMock
+    ):
+        """Token marked inactive by introspection is rejected before userinfo."""
+        httpx_mock.add_response(
+            url=_INTROSPECTION_RE,
+            json={"active": False},
+        )
+
+        verifier = ClerkTokenVerifier(
+            domain=CLERK_DOMAIN,
+            client_id="clerk-client-id",
+            client_secret="clerk-client-secret",
+        )
+        result = await verifier.verify_token("inactive-token")
+
+        assert result is None
+
+    async def test_introspection_missing_active_field_returns_none(
+        self, httpx_mock: HTTPXMock
+    ):
+        """RFC 7662 requires the 'active' field; a missing field is malformed and rejected."""
+        httpx_mock.add_response(
+            url=_INTROSPECTION_RE,
+            json={"scope": "openid email profile", "aud": "clerk-client-id"},
+        )
+
+        verifier = ClerkTokenVerifier(
+            domain=CLERK_DOMAIN,
+            client_id="clerk-client-id",
+            client_secret="clerk-client-secret",
+        )
+        result = await verifier.verify_token("token-malformed-response")
+
+        assert result is None
+
+    async def test_introspection_failure_rejects_when_scopes_required(
+        self, httpx_mock: HTTPXMock
+    ):
+        """When introspection fails (non-200), token is rejected regardless of scopes."""
+        httpx_mock.add_response(
+            url=_INTROSPECTION_RE,
+            status_code=500,
+            json={"error": "internal_server_error"},
+        )
+
+        verifier = ClerkTokenVerifier(
+            domain=CLERK_DOMAIN,
+            required_scopes=["openid", "email"],
+        )
+        result = await verifier.verify_token("valid-token")
+
+        assert result is None
+
+    async def test_empty_scopes_rejects_when_required(self, httpx_mock: HTTPXMock):
+        """When introspection returns no scopes and required_scopes are set, token is rejected."""
+        httpx_mock.add_response(
+            url=_INTROSPECTION_RE,
+            json={"active": True, "scope": ""},
+        )
+
+        verifier = ClerkTokenVerifier(
+            domain=CLERK_DOMAIN,
+            required_scopes=["openid", "email", "profile"],
+        )
+        result = await verifier.verify_token("valid-token")
+
+        assert result is None
+
+    async def test_required_scopes_not_satisfied_returns_none(
+        self, httpx_mock: HTTPXMock
+    ):
+        """Token without required scopes is rejected before userinfo."""
+        httpx_mock.add_response(
+            url=_INTROSPECTION_RE,
+            json={"active": True, "scope": "openid"},
+        )
+
+        verifier = ClerkTokenVerifier(
+            domain=CLERK_DOMAIN,
+            required_scopes=["openid", "email", "profile"],
+        )
+        result = await verifier.verify_token("token-missing-scopes")
+
+        assert result is None
+
+    async def test_uses_bearer_header_for_userinfo(self, httpx_mock: HTTPXMock):
+        """verify_token sends the token as a Bearer header to userinfo."""
+        httpx_mock.add_response(
+            url=_INTROSPECTION_RE,
+            json={"active": True, "scope": "openid", "sub": "user_abc123"},
+        )
+        httpx_mock.add_response(
+            url=_USERINFO_RE,
+            json={"sub": "user_abc123"},
+        )
+
+        verifier = ClerkTokenVerifier(domain=CLERK_DOMAIN)
+        await verifier.verify_token("my-access-token")
+
+        requests = httpx_mock.get_requests()
+        userinfo_req = requests[1]
+        assert userinfo_req.headers["Authorization"] == "Bearer my-access-token"
+
+    async def test_introspection_sends_client_credentials(self, httpx_mock: HTTPXMock):
+        """Introspection request sends credentials via HTTP Basic Auth when both are set."""
+        httpx_mock.add_response(
+            url=_INTROSPECTION_RE,
+            json={"active": True, "scope": "openid", "aud": "clerk-client-id"},
+        )
+        httpx_mock.add_response(
+            url=_USERINFO_RE,
+            json={"sub": "user_abc123"},
+        )
+
+        verifier = ClerkTokenVerifier(
+            domain=CLERK_DOMAIN,
+            client_id="clerk-client-id",
+            client_secret="clerk-client-secret",
+        )
+        await verifier.verify_token("my-access-token")
+
+        requests = httpx_mock.get_requests()
+        introspect_req = requests[0]
+        body = introspect_req.content.decode()
+        assert "token=my-access-token" in body
+        assert introspect_req.headers.get("Authorization", "").startswith("Basic ")
+
+    async def test_expires_at_from_introspection(self, httpx_mock: HTTPXMock):
+        """expires_at is set from the 'exp' claim in the introspection response."""
+        httpx_mock.add_response(
+            url=_USERINFO_RE,
+            json={"sub": "user_abc123"},
+        )
+        httpx_mock.add_response(
+            url=_INTROSPECTION_RE,
+            json={"active": True, "scope": "openid", "exp": 1700000000},
+        )
+
+        verifier = ClerkTokenVerifier(domain=CLERK_DOMAIN)
+        result = await verifier.verify_token("valid-token")
+
+        assert result is not None
+        assert result.expires_at == 1700000000
+
+    async def test_client_id_falls_back_to_sub(self, httpx_mock: HTTPXMock):
+        """When introspection has no aud/client_id, client_id falls back to sub."""
+        httpx_mock.add_response(
+            url=_USERINFO_RE,
+            json={"sub": "user_abc123"},
+        )
+        httpx_mock.add_response(
+            url=_INTROSPECTION_RE,
+            json={"active": True, "scope": "openid"},
+        )
+
+        verifier = ClerkTokenVerifier(domain=CLERK_DOMAIN)
+        result = await verifier.verify_token("valid-token")
+
+        assert result is not None
+        assert result.client_id == "user_abc123"
+
+    async def test_aud_from_introspection_client_id_field(self, httpx_mock: HTTPXMock):
+        """When introspection returns client_id but not aud, client_id is used."""
+        httpx_mock.add_response(
+            url=_USERINFO_RE,
+            json={"sub": "user_abc123"},
+        )
+        httpx_mock.add_response(
+            url=_INTROSPECTION_RE,
+            json={"active": True, "scope": "openid", "client_id": "my-app-id"},
+        )
+
+        verifier = ClerkTokenVerifier(domain=CLERK_DOMAIN)
+        result = await verifier.verify_token("valid-token")
+
+        assert result is not None
+        assert result.client_id == "my-app-id"
+        assert result.claims["aud"] == "my-app-id"
+
+    async def test_no_required_scopes_accepts_any(self, httpx_mock: HTTPXMock):
+        """When no required_scopes are set, any valid token is accepted."""
+        httpx_mock.add_response(
+            url=_USERINFO_RE,
+            json={"sub": "user_abc123"},
+        )
+        httpx_mock.add_response(
+            url=_INTROSPECTION_RE,
+            json={"active": True, "scope": "openid custom_scope"},
+        )
+
+        verifier = ClerkTokenVerifier(domain=CLERK_DOMAIN)
+        result = await verifier.verify_token("valid-token")
+
+        assert result is not None
+        assert result.scopes == ["openid", "custom_scope"]
+
+    async def test_clerk_user_data_in_claims(self, httpx_mock: HTTPXMock):
+        """The full userinfo response is stored in clerk_user_data claim."""
+        user_data = {
+            "sub": "user_abc123",
+            "email": "user@example.com",
+            "name": "Test User",
+        }
+        httpx_mock.add_response(
+            url=_USERINFO_RE,
+            json=user_data,
+        )
+        httpx_mock.add_response(
+            url=_INTROSPECTION_RE,
+            json={"active": True},
+        )
+
+        verifier = ClerkTokenVerifier(domain=CLERK_DOMAIN)
+        result = await verifier.verify_token("valid-token")
+
+        assert result is not None
+        assert result.claims["clerk_user_data"] == user_data
+
+    async def test_network_error_returns_none(self, httpx_mock: HTTPXMock):
+        """Network errors during introspection return None instead of raising."""
+        httpx_mock.add_exception(
+            httpx.ConnectError("Connection refused"),
+            url=_INTROSPECTION_RE,
+        )
+
+        verifier = ClerkTokenVerifier(domain=CLERK_DOMAIN)
+        result = await verifier.verify_token("valid-token")
+
+        assert result is None
+
+    async def test_introspection_failure_rejects_without_required_scopes(
+        self, httpx_mock: HTTPXMock
+    ):
+        """Introspection failure (non-200) rejects the token even without required_scopes."""
+        httpx_mock.add_response(
+            url=_INTROSPECTION_RE,
+            status_code=500,
+            json={"error": "internal_server_error"},
+        )
+
+        verifier = ClerkTokenVerifier(domain=CLERK_DOMAIN)
+        result = await verifier.verify_token("valid-token")
+
+        assert result is None
+
+    async def test_audience_mismatch_returns_none(self, httpx_mock: HTTPXMock):
+        """Token with wrong audience is rejected before userinfo is called."""
+        httpx_mock.add_response(
+            url=_INTROSPECTION_RE,
+            json={"active": True, "scope": "openid", "aud": "wrong-client-id"},
+        )
+
+        verifier = ClerkTokenVerifier(
+            domain=CLERK_DOMAIN,
+            client_id="my-client-id",
+            client_secret="my-client-secret",
+        )
+        result = await verifier.verify_token("valid-token")
+
+        assert result is None
+
+    async def test_audience_missing_returns_none_when_client_id_set(
+        self, httpx_mock: HTTPXMock
+    ):
+        """Token without audience is rejected before userinfo is called."""
+        httpx_mock.add_response(
+            url=_INTROSPECTION_RE,
+            json={"active": True, "scope": "openid"},
+        )
+
+        verifier = ClerkTokenVerifier(
+            domain=CLERK_DOMAIN,
+            client_id="my-client-id",
+            client_secret="my-client-secret",
+        )
+        result = await verifier.verify_token("valid-token")
+
+        assert result is None
+
+    async def test_audience_not_checked_without_client_id(self, httpx_mock: HTTPXMock):
+        """Without client_id configured, any audience is accepted."""
+        httpx_mock.add_response(
+            url=_USERINFO_RE,
+            json={"sub": "user_abc123"},
+        )
+        httpx_mock.add_response(
+            url=_INTROSPECTION_RE,
+            json={"active": True, "scope": "openid", "aud": "some-other-id"},
+        )
+
+        verifier = ClerkTokenVerifier(domain=CLERK_DOMAIN)
+        result = await verifier.verify_token("valid-token")
+
+        assert result is not None
+        assert result.claims["aud"] == "some-other-id"