Add http_client parameter to token verifiers for connection pooling

🤖 Generated with Claude Code

https://claude.ai/code/session_012QKWmKd21vypDmxWbwuE4e

docs/servers/auth/token-verification.mdx

@@ -321,6 +321,77 @@ print(f"Test token: {test_token}")
 
 This pattern enables comprehensive testing of JWT validation logic without depending on external token issuers. The generated tokens are cryptographically valid and will pass all standard JWT validation checks.
 
+## HTTP Client Customization
+
+<VersionBadge version="2.18.0" />
+
+All token verifiers that make HTTP calls accept an optional `http_client` parameter. This lets you provide your own `httpx.AsyncClient` for connection pooling, custom TLS configuration, or proxy settings.
+
+### Connection Pooling
+
+By default, each token verification call creates a fresh HTTP client. Under high load, this means repeated TCP connections and TLS handshakes. Providing a shared client enables connection pooling across calls:
+
+```python
+import httpx
+from fastmcp import FastMCP
+from fastmcp.server.auth.providers.introspection import IntrospectionTokenVerifier
+
+# Create a shared client with connection pooling
+http_client = httpx.AsyncClient(
+    timeout=10,
+    limits=httpx.Limits(max_connections=20, max_keepalive_connections=10),
+)
+
+verifier = IntrospectionTokenVerifier(
+    introspection_url="https://auth.yourcompany.com/oauth/introspect",
+    client_id="mcp-resource-server",
+    client_secret="your-client-secret",
+    http_client=http_client,
+)
+
+mcp = FastMCP(name="Protected API", auth=verifier)
+```
+
+The same pattern works for `JWTVerifier` when using JWKS endpoints:
+
+```python
+from fastmcp.server.auth.providers.jwt import JWTVerifier
+
+verifier = JWTVerifier(
+    jwks_uri="https://auth.yourcompany.com/.well-known/jwks.json",
+    issuer="https://auth.yourcompany.com",
+    http_client=http_client,
+)
+```
+
+<Note>
+When you provide an `http_client`, you are responsible for its lifecycle. The verifier will not close it. Use the server's `lifespan` to manage client cleanup:
+
+```python
+from contextlib import asynccontextmanager
+from fastmcp import FastMCP
+from fastmcp.server.auth.providers.introspection import IntrospectionTokenVerifier
+
+http_client = httpx.AsyncClient(timeout=10)
+
+verifier = IntrospectionTokenVerifier(
+    introspection_url="https://auth.example.com/introspect",
+    client_id="my-service",
+    client_secret="secret",
+    http_client=http_client,
+)
+
+@asynccontextmanager
+async def lifespan(app):
+    yield
+    await http_client.aclose()
+
+mcp = FastMCP(name="My API", auth=verifier, lifespan=lifespan)
+```
+</Note>
+
+The convenience providers (`GitHubProvider`, `GoogleProvider`, `DiscordProvider`, `WorkOSProvider`, `AzureProvider`) also accept `http_client` and pass it through to their internal token verifier.
+
 ## Production Configuration
 
 For production deployments, load sensitive configuration from environment variables:

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

@@ -10,6 +10,7 @@ import hashlib
 from collections import OrderedDict
 from typing import TYPE_CHECKING, Any, cast
 
+import httpx
 from key_value.aio.protocols import AsyncKeyValue
 
 from fastmcp.server.auth.oauth_proxy import OAuthProxy
@@ -107,6 +108,7 @@ class AzureProvider(OAuthProxy):
         jwt_signing_key: str | bytes | None = None,
         require_authorization_consent: bool = True,
         base_authority: str = "login.microsoftonline.com",
+        http_client: httpx.AsyncClient | None = None,
     ) -> None:
         """Initialize Azure OAuth provider.
 
@@ -151,6 +153,9 @@ class AzureProvider(OAuthProxy):
                 When True, users see a consent screen before being redirected to Azure.
                 When False, authorization proceeds directly without user confirmation.
                 SECURITY WARNING: Only disable for local development or testing environments.
+            http_client: Optional httpx.AsyncClient for connection pooling in JWKS fetches.
+                When provided, the client is reused for JWT key fetches and the caller
+                is responsible for its lifecycle. When None (default), a fresh client is created per fetch.
         """
         # Parse scopes if provided as string
         parsed_required_scopes = parse_scopes(required_scopes)
@@ -202,6 +207,7 @@ class AzureProvider(OAuthProxy):
             audience=client_id,
             algorithm="RS256",
             required_scopes=validation_scopes,  # Only validate non-OIDC scopes
+            http_client=http_client,
         )
 
         # Build Azure OAuth endpoints with tenant

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

@@ -21,6 +21,7 @@ Example:
 
 from __future__ import annotations
 
+import contextlib
 import time
 from datetime import datetime
 
@@ -49,20 +50,29 @@ class DiscordTokenVerifier(TokenVerifier):
         *,
         required_scopes: list[str] | None = None,
         timeout_seconds: int = 10,
+        http_client: httpx.AsyncClient | None = None,
     ):
         """Initialize the Discord token verifier.
 
         Args:
             required_scopes: Required OAuth scopes (e.g., ['email'])
             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.timeout_seconds = timeout_seconds
+        self._http_client = http_client
 
     async def verify_token(self, token: str) -> AccessToken | None:
         """Verify Discord OAuth token by calling Discord's tokeninfo API."""
         try:
-            async with httpx.AsyncClient(timeout=self.timeout_seconds) as client:
+            async with (
+                contextlib.nullcontext(self._http_client)
+                if self._http_client is not None
+                else httpx.AsyncClient(timeout=self.timeout_seconds)
+            ) as client:
                 # Use Discord's tokeninfo endpoint to validate the token
                 headers = {
                     "Authorization": f"Bearer {token}",
@@ -183,6 +193,7 @@ class DiscordProvider(OAuthProxy):
         client_storage: AsyncKeyValue | None = None,
         jwt_signing_key: str | bytes | None = None,
         require_authorization_consent: bool = True,
+        http_client: httpx.AsyncClient | None = None,
     ):
         """Initialize Discord OAuth provider.
 
@@ -210,6 +221,9 @@ class DiscordProvider(OAuthProxy):
                 When True, users see a consent screen before being redirected to Discord.
                 When False, authorization proceeds directly without user confirmation.
                 SECURITY WARNING: Only disable for local development or testing environments.
+            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.
         """
         # Parse scopes if provided as string
         required_scopes_final = (
@@ -222,6 +236,7 @@ class DiscordProvider(OAuthProxy):
         token_verifier = DiscordTokenVerifier(
             required_scopes=required_scopes_final,
             timeout_seconds=timeout_seconds,
+            http_client=http_client,
         )
 
         # Initialize OAuth proxy with Discord endpoints

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

@@ -21,6 +21,8 @@ Example:
 
 from __future__ import annotations
 
+import contextlib
+
 import httpx
 from key_value.aio.protocols import AsyncKeyValue
 from pydantic import AnyHttpUrl
@@ -46,20 +48,29 @@ class GitHubTokenVerifier(TokenVerifier):
         *,
         required_scopes: list[str] | None = None,
         timeout_seconds: int = 10,
+        http_client: httpx.AsyncClient | None = None,
     ):
         """Initialize the GitHub token verifier.
 
         Args:
             required_scopes: Required OAuth scopes (e.g., ['user:email'])
             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.timeout_seconds = timeout_seconds
+        self._http_client = http_client
 
     async def verify_token(self, token: str) -> AccessToken | None:
         """Verify GitHub OAuth token by calling GitHub API."""
         try:
-            async with httpx.AsyncClient(timeout=self.timeout_seconds) as client:
+            async with (
+                contextlib.nullcontext(self._http_client)
+                if self._http_client is not None
+                else httpx.AsyncClient(timeout=self.timeout_seconds)
+            ) as client:
                 # Get token info from GitHub API
                 response = await client.get(
                     "https://api.github.com/user",
@@ -181,6 +192,7 @@ class GitHubProvider(OAuthProxy):
         client_storage: AsyncKeyValue | None = None,
         jwt_signing_key: str | bytes | None = None,
         require_authorization_consent: bool = True,
+        http_client: httpx.AsyncClient | None = None,
     ):
         """Initialize GitHub OAuth provider.
 
@@ -205,6 +217,9 @@ class GitHubProvider(OAuthProxy):
                 When True, users see a consent screen before being redirected to GitHub.
                 When False, authorization proceeds directly without user confirmation.
                 SECURITY WARNING: Only disable for local development or testing environments.
+            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.
         """
         # Parse scopes if provided as string
         required_scopes_final = (
@@ -215,6 +230,7 @@ class GitHubProvider(OAuthProxy):
         token_verifier = GitHubTokenVerifier(
             required_scopes=required_scopes_final,
             timeout_seconds=timeout_seconds,
+            http_client=http_client,
         )
 
         # Initialize OAuth proxy with GitHub endpoints

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

@@ -21,6 +21,7 @@ Example:
 
 from __future__ import annotations
 
+import contextlib
 import time
 
 import httpx
@@ -48,20 +49,29 @@ class GoogleTokenVerifier(TokenVerifier):
         *,
         required_scopes: list[str] | None = None,
         timeout_seconds: int = 10,
+        http_client: httpx.AsyncClient | None = None,
     ):
         """Initialize the Google token verifier.
 
         Args:
             required_scopes: Required OAuth scopes (e.g., ['openid', 'https://www.googleapis.com/auth/userinfo.email'])
             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.timeout_seconds = timeout_seconds
+        self._http_client = http_client
 
     async def verify_token(self, token: str) -> AccessToken | None:
         """Verify Google OAuth token by calling Google's tokeninfo API."""
         try:
-            async with httpx.AsyncClient(timeout=self.timeout_seconds) as client:
+            async with (
+                contextlib.nullcontext(self._http_client)
+                if self._http_client is not None
+                else httpx.AsyncClient(timeout=self.timeout_seconds)
+            ) as client:
                 # Use Google's tokeninfo endpoint to validate the token
                 response = await client.get(
                     "https://www.googleapis.com/oauth2/v1/tokeninfo",
@@ -198,6 +208,7 @@ class GoogleProvider(OAuthProxy):
         jwt_signing_key: str | bytes | None = None,
         require_authorization_consent: bool = True,
         extra_authorize_params: dict[str, str] | None = None,
+        http_client: httpx.AsyncClient | None = None,
     ):
         """Initialize Google OAuth provider.
 
@@ -229,6 +240,9 @@ class GoogleProvider(OAuthProxy):
                 By default, GoogleProvider sets {"access_type": "offline", "prompt": "consent"} to ensure
                 refresh tokens are returned. You can override these defaults or add additional parameters.
                 Example: {"prompt": "select_account"} to let users choose their Google account.
+            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.
         """
         # Parse scopes if provided as string
         # Google requires at least one scope - openid is the minimal OIDC scope
@@ -240,6 +254,7 @@ class GoogleProvider(OAuthProxy):
         token_verifier = GoogleTokenVerifier(
             required_scopes=required_scopes_final,
             timeout_seconds=timeout_seconds,
+            http_client=http_client,
         )
 
         # Set Google-specific defaults for extra authorize params

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

@@ -24,6 +24,7 @@ Example:
 from __future__ import annotations
 
 import base64
+import contextlib
 import time
 from typing import Any, Literal, get_args
 
@@ -80,6 +81,7 @@ class IntrospectionTokenVerifier(TokenVerifier):
         timeout_seconds: int = 10,
         required_scopes: list[str] | None = None,
         base_url: AnyHttpUrl | str | None = None,
+        http_client: httpx.AsyncClient | None = None,
     ):
         """
         Initialize the introspection token verifier.
@@ -93,6 +95,9 @@ class IntrospectionTokenVerifier(TokenVerifier):
             timeout_seconds: HTTP request timeout in seconds (default: 10)
             required_scopes: Required scopes for all tokens (optional)
             base_url: Base URL for TokenVerifier protocol
+            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.
         """
         # Parse scopes if provided as string
         parsed_required_scopes = (
@@ -120,6 +125,7 @@ class IntrospectionTokenVerifier(TokenVerifier):
         self.client_auth_method: ClientAuthMethod = client_auth_method
 
         self.timeout_seconds = timeout_seconds
+        self._http_client = http_client
         self.logger = get_logger(__name__)
 
     def _create_basic_auth_header(self) -> str:
@@ -166,7 +172,11 @@ class IntrospectionTokenVerifier(TokenVerifier):
             AccessToken object if valid and active, None if invalid, inactive, or expired
         """
         try:
-            async with httpx.AsyncClient(timeout=self.timeout_seconds) as client:
+            async with (
+                contextlib.nullcontext(self._http_client)
+                if self._http_client is not None
+                else httpx.AsyncClient(timeout=self.timeout_seconds)
+            ) as client:
                 # Prepare introspection request per RFC 7662
                 # Build request data with token and token_type_hint
                 data = {

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

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import contextlib
 import json
 import time
 from dataclasses import dataclass
@@ -168,6 +169,7 @@ class JWTVerifier(TokenVerifier):
         required_scopes: list[str] | None = None,
         base_url: AnyHttpUrl | str | None = None,
         ssrf_safe: bool = False,
+        http_client: httpx.AsyncClient | None = None,
     ):
         """
         Initialize a JWTVerifier configured to validate JWTs using either a static key or a JWKS endpoint.
@@ -184,6 +186,10 @@ class JWTVerifier(TokenVerifier):
                 public IPs, DNS pinning). Enable when the JWKS URI comes from
                 untrusted input (e.g. CIMD documents). Defaults to False so
                 operator-configured JWKS URIs (including localhost) work normally.
+            http_client: Optional httpx.AsyncClient for connection pooling. When provided,
+                the client is reused for JWKS fetches and the caller is responsible for
+                its lifecycle. When None (default), a fresh client is created per fetch.
+                Only used when ssrf_safe is False; SSRF-safe fetches use their own transport.
 
         Raises:
             ValueError: If neither or both of `public_key` and `jwks_uri` are provided, or if `algorithm` is unsupported.
@@ -228,6 +234,7 @@ class JWTVerifier(TokenVerifier):
         self.public_key = public_key
         self.jwks_uri = jwks_uri
         self.ssrf_safe = ssrf_safe
+        self._http_client = http_client
         self.jwt = JsonWebToken([self.algorithm])
         self.logger = get_logger(__name__)
 
@@ -328,7 +335,11 @@ class JWTVerifier(TokenVerifier):
             )
             return json.loads(content)
         else:
-            async with httpx.AsyncClient(timeout=httpx.Timeout(10.0)) as client:
+            async with (
+                contextlib.nullcontext(self._http_client)
+                if self._http_client is not None
+                else httpx.AsyncClient(timeout=httpx.Timeout(10.0))
+            ) as client:
                 response = await client.get(self.jwks_uri)
                 response.raise_for_status()
                 return response.json()

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

@@ -10,6 +10,8 @@ Choose based on your WorkOS setup and authentication requirements.
 
 from __future__ import annotations
 
+import contextlib
+
 import httpx
 from key_value.aio.protocols import AsyncKeyValue
 from pydantic import AnyHttpUrl
@@ -38,6 +40,7 @@ class WorkOSTokenVerifier(TokenVerifier):
         authkit_domain: str,
         required_scopes: list[str] | None = None,
         timeout_seconds: int = 10,
+        http_client: httpx.AsyncClient | None = None,
     ):
         """Initialize the WorkOS token verifier.
 
@@ -45,15 +48,23 @@ class WorkOSTokenVerifier(TokenVerifier):
             authkit_domain: WorkOS AuthKit domain (e.g., "https://your-app.authkit.app")
             required_scopes: Required OAuth scopes
             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.authkit_domain = authkit_domain.rstrip("/")
         self.timeout_seconds = timeout_seconds
+        self._http_client = http_client
 
     async def verify_token(self, token: str) -> AccessToken | None:
         """Verify WorkOS OAuth token by calling userinfo endpoint."""
         try:
-            async with httpx.AsyncClient(timeout=self.timeout_seconds) as client:
+            async with (
+                contextlib.nullcontext(self._http_client)
+                if self._http_client is not None
+                else httpx.AsyncClient(timeout=self.timeout_seconds)
+            ) as client:
                 # Use WorkOS AuthKit userinfo endpoint to validate token
                 response = await client.get(
                     f"{self.authkit_domain}/oauth2/userinfo",
@@ -146,6 +157,7 @@ class WorkOSProvider(OAuthProxy):
         client_storage: AsyncKeyValue | None = None,
         jwt_signing_key: str | bytes | None = None,
         require_authorization_consent: bool = True,
+        http_client: httpx.AsyncClient | None = None,
     ):
         """Initialize WorkOS OAuth provider.
 
@@ -171,6 +183,9 @@ class WorkOSProvider(OAuthProxy):
                 When True, users see a consent screen before being redirected to WorkOS.
                 When False, authorization proceeds directly without user confirmation.
                 SECURITY WARNING: Only disable for local development or testing environments.
+            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.
         """
         # Apply defaults and ensure authkit_domain is a full URL
         authkit_domain_str = authkit_domain
@@ -186,6 +201,7 @@ class WorkOSProvider(OAuthProxy):
             authkit_domain=authkit_domain_final,
             required_scopes=scopes_final,
             timeout_seconds=timeout_seconds,
+            http_client=http_client,
         )
 
         # Initialize OAuth proxy with WorkOS AuthKit endpoints

tests/server/auth/providers/test_http_client.py

@@ -0,0 +1,354 @@
+"""Tests for http_client parameter on token verifiers.
+
+Verifies that all token verifiers accept an optional httpx.AsyncClient for
+connection pooling (issues #3287 and #3293).
+"""
+
+import time
+
+import httpx
+import pytest
+from pytest_httpx import HTTPXMock
+
+from fastmcp.server.auth.providers.introspection import IntrospectionTokenVerifier
+from fastmcp.server.auth.providers.jwt import JWTVerifier, RSAKeyPair
+
+
+class TestIntrospectionHttpClient:
+    """Test http_client parameter on IntrospectionTokenVerifier."""
+
+    @pytest.fixture
+    def shared_client(self) -> httpx.AsyncClient:
+        return httpx.AsyncClient(timeout=30)
+
+    def test_stores_http_client(self, shared_client: httpx.AsyncClient):
+        verifier = IntrospectionTokenVerifier(
+            introspection_url="https://auth.example.com/introspect",
+            client_id="test",
+            client_secret="secret",
+            http_client=shared_client,
+        )
+        assert verifier._http_client is shared_client
+
+    def test_default_http_client_is_none(self):
+        verifier = IntrospectionTokenVerifier(
+            introspection_url="https://auth.example.com/introspect",
+            client_id="test",
+            client_secret="secret",
+        )
+        assert verifier._http_client is None
+
+    async def test_uses_provided_client(
+        self, shared_client: httpx.AsyncClient, httpx_mock: HTTPXMock
+    ):
+        """When http_client is provided, it should be used for requests."""
+        httpx_mock.add_response(
+            url="https://auth.example.com/introspect",
+            method="POST",
+            json={
+                "active": True,
+                "client_id": "user-1",
+                "scope": "read",
+                "exp": int(time.time()) + 3600,
+            },
+        )
+
+        verifier = IntrospectionTokenVerifier(
+            introspection_url="https://auth.example.com/introspect",
+            client_id="test",
+            client_secret="secret",
+            http_client=shared_client,
+        )
+
+        result = await verifier.verify_token("tok")
+        assert result is not None
+        assert result.client_id == "user-1"
+
+    async def test_client_not_closed_after_call(
+        self, shared_client: httpx.AsyncClient, httpx_mock: HTTPXMock
+    ):
+        """User-provided client must not be closed by the verifier."""
+        httpx_mock.add_response(
+            url="https://auth.example.com/introspect",
+            method="POST",
+            json={
+                "active": True,
+                "client_id": "user-1",
+                "scope": "read",
+                "exp": int(time.time()) + 3600,
+            },
+        )
+
+        verifier = IntrospectionTokenVerifier(
+            introspection_url="https://auth.example.com/introspect",
+            client_id="test",
+            client_secret="secret",
+            http_client=shared_client,
+        )
+
+        await verifier.verify_token("tok")
+        # Client should still be open — not closed by the verifier
+        assert not shared_client.is_closed
+
+    async def test_reuses_client_across_calls(
+        self, shared_client: httpx.AsyncClient, httpx_mock: HTTPXMock
+    ):
+        """Same client instance should be reused across multiple verify_token calls."""
+        for _ in range(3):
+            httpx_mock.add_response(
+                url="https://auth.example.com/introspect",
+                method="POST",
+                json={
+                    "active": True,
+                    "client_id": "user-1",
+                    "scope": "read",
+                    "exp": int(time.time()) + 3600,
+                },
+            )
+
+        verifier = IntrospectionTokenVerifier(
+            introspection_url="https://auth.example.com/introspect",
+            client_id="test",
+            client_secret="secret",
+            http_client=shared_client,
+        )
+
+        for _ in range(3):
+            result = await verifier.verify_token("tok")
+            assert result is not None
+
+        assert not shared_client.is_closed
+
+
+class TestJWTVerifierHttpClient:
+    """Test http_client parameter on JWTVerifier."""
+
+    @pytest.fixture(scope="class")
+    def rsa_key_pair(self) -> RSAKeyPair:
+        return RSAKeyPair.generate()
+
+    @pytest.fixture
+    def shared_client(self) -> httpx.AsyncClient:
+        return httpx.AsyncClient(timeout=30)
+
+    def test_stores_http_client(self, shared_client: httpx.AsyncClient):
+        verifier = JWTVerifier(
+            jwks_uri="https://auth.example.com/.well-known/jwks.json",
+            http_client=shared_client,
+        )
+        assert verifier._http_client is shared_client
+
+    def test_default_http_client_is_none(self):
+        verifier = JWTVerifier(
+            jwks_uri="https://auth.example.com/.well-known/jwks.json",
+        )
+        assert verifier._http_client is None
+
+    async def test_jwks_fetch_uses_provided_client(
+        self,
+        rsa_key_pair: RSAKeyPair,
+        shared_client: httpx.AsyncClient,
+        httpx_mock: HTTPXMock,
+    ):
+        """When http_client is provided, JWKS fetches should use it."""
+        from authlib.jose import JsonWebKey
+
+        # Build a JWKS response from the RSA key pair
+        public_key_obj = JsonWebKey.import_key(rsa_key_pair.public_key)
+        jwk_dict = dict(public_key_obj.as_dict())
+        jwk_dict["kid"] = "test-key-1"
+        jwk_dict["use"] = "sig"
+        jwk_dict["alg"] = "RS256"
+
+        httpx_mock.add_response(
+            url="https://auth.example.com/.well-known/jwks.json",
+            json={"keys": [jwk_dict]},
+        )
+
+        verifier = JWTVerifier(
+            jwks_uri="https://auth.example.com/.well-known/jwks.json",
+            issuer="https://auth.example.com",
+            http_client=shared_client,
+        )
+
+        token = rsa_key_pair.create_token(
+            issuer="https://auth.example.com",
+            kid="test-key-1",
+        )
+
+        result = await verifier.verify_token(token)
+        assert result is not None
+        assert not shared_client.is_closed
+
+    async def test_ssrf_safe_ignores_http_client(
+        self,
+        shared_client: httpx.AsyncClient,
+    ):
+        """When ssrf_safe=True, the custom http_client should NOT be used."""
+        verifier = JWTVerifier(
+            jwks_uri="https://auth.example.com/.well-known/jwks.json",
+            ssrf_safe=True,
+            http_client=shared_client,
+        )
+
+        # ssrf_safe uses ssrf_safe_fetch instead of httpx.AsyncClient
+        # The http_client is stored but not used in this code path
+        assert verifier._http_client is shared_client
+        assert verifier.ssrf_safe is True
+
+
+class TestGitHubHttpClient:
+    """Test http_client parameter on GitHubTokenVerifier."""
+
+    def test_stores_http_client(self):
+        from fastmcp.server.auth.providers.github import GitHubTokenVerifier
+
+        client = httpx.AsyncClient()
+        verifier = GitHubTokenVerifier(http_client=client)
+        assert verifier._http_client is client
+
+    async def test_uses_provided_client(self, httpx_mock: HTTPXMock):
+        from fastmcp.server.auth.providers.github import GitHubTokenVerifier
+
+        client = httpx.AsyncClient()
+        httpx_mock.add_response(
+            url="https://api.github.com/user",
+            json={"id": 123, "login": "testuser"},
+        )
+        httpx_mock.add_response(
+            url="https://api.github.com/user/repos",
+            headers={"x-oauth-scopes": "user,repo"},
+            json=[],
+        )
+
+        verifier = GitHubTokenVerifier(http_client=client)
+        result = await verifier.verify_token("ghp_test")
+        assert result is not None
+        assert not client.is_closed
+
+
+class TestDiscordHttpClient:
+    """Test http_client parameter on DiscordTokenVerifier."""
+
+    def test_stores_http_client(self):
+        from fastmcp.server.auth.providers.discord import DiscordTokenVerifier
+
+        client = httpx.AsyncClient()
+        verifier = DiscordTokenVerifier(http_client=client)
+        assert verifier._http_client is client
+
+
+class TestGoogleHttpClient:
+    """Test http_client parameter on GoogleTokenVerifier."""
+
+    def test_stores_http_client(self):
+        from fastmcp.server.auth.providers.google import GoogleTokenVerifier
+
+        client = httpx.AsyncClient()
+        verifier = GoogleTokenVerifier(http_client=client)
+        assert verifier._http_client is client
+
+
+class TestWorkOSHttpClient:
+    """Test http_client parameter on WorkOSTokenVerifier."""
+
+    def test_stores_http_client(self):
+        from fastmcp.server.auth.providers.workos import WorkOSTokenVerifier
+
+        client = httpx.AsyncClient()
+        verifier = WorkOSTokenVerifier(
+            authkit_domain="https://test.authkit.app",
+            http_client=client,
+        )
+        assert verifier._http_client is client
+
+
+class TestProviderHttpClientPassthrough:
+    """Test that convenience providers pass http_client to their verifiers."""
+
+    def test_github_provider_threads_http_client(self):
+        from fastmcp.server.auth.providers.github import (
+            GitHubProvider,
+            GitHubTokenVerifier,
+        )
+
+        client = httpx.AsyncClient()
+        provider = GitHubProvider(
+            client_id="test",
+            client_secret="secret",
+            base_url="https://example.com",
+            http_client=client,
+        )
+        # OAuthProxy stores token verifier as _token_validator
+        verifier = provider._token_validator
+        assert isinstance(verifier, GitHubTokenVerifier)
+        assert verifier._http_client is client
+
+    def test_discord_provider_threads_http_client(self):
+        from fastmcp.server.auth.providers.discord import (
+            DiscordProvider,
+            DiscordTokenVerifier,
+        )
+
+        client = httpx.AsyncClient()
+        provider = DiscordProvider(
+            client_id="test",
+            client_secret="secret",
+            base_url="https://example.com",
+            http_client=client,
+        )
+        verifier = provider._token_validator
+        assert isinstance(verifier, DiscordTokenVerifier)
+        assert verifier._http_client is client
+
+    def test_google_provider_threads_http_client(self):
+        from fastmcp.server.auth.providers.google import (
+            GoogleProvider,
+            GoogleTokenVerifier,
+        )
+
+        client = httpx.AsyncClient()
+        provider = GoogleProvider(
+            client_id="test",
+            client_secret="secret",
+            base_url="https://example.com",
+            http_client=client,
+        )
+        verifier = provider._token_validator
+        assert isinstance(verifier, GoogleTokenVerifier)
+        assert verifier._http_client is client
+
+    def test_workos_provider_threads_http_client(self):
+        from fastmcp.server.auth.providers.workos import (
+            WorkOSProvider,
+            WorkOSTokenVerifier,
+        )
+
+        client = httpx.AsyncClient()
+        provider = WorkOSProvider(
+            client_id="test",
+            client_secret="secret",
+            authkit_domain="https://test.authkit.app",
+            base_url="https://example.com",
+            http_client=client,
+        )
+        verifier = provider._token_validator
+        assert isinstance(verifier, WorkOSTokenVerifier)
+        assert verifier._http_client is client
+
+    def test_azure_provider_threads_http_client(self):
+        from fastmcp.server.auth.providers.azure import AzureProvider
+        from fastmcp.server.auth.providers.jwt import JWTVerifier
+
+        client = httpx.AsyncClient()
+        provider = AzureProvider(
+            client_id="test-client-id",
+            client_secret="secret",
+            tenant_id="test-tenant-id",
+            required_scopes=["read"],
+            base_url="https://example.com",
+            http_client=client,
+        )
+        verifier = provider._token_validator
+        assert isinstance(verifier, JWTVerifier)
+        assert verifier._http_client is client