diff --git a/docs/servers/auth/token-verification.mdx b/docs/servers/auth/token-verification.mdx
index 890214c6d..5925a2d43 100644
--- a/docs/servers/auth/token-verification.mdx
+++ b/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
+
+
+
+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,
+)
+```
+
+
+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)
+```
+
+
+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:
diff --git a/src/fastmcp/server/auth/providers/azure.py b/src/fastmcp/server/auth/providers/azure.py
index 868631d13..244ab91c0 100644
--- a/src/fastmcp/server/auth/providers/azure.py
+++ b/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
diff --git a/src/fastmcp/server/auth/providers/discord.py b/src/fastmcp/server/auth/providers/discord.py
index 4fb5ebb53..7af28e026 100644
--- a/src/fastmcp/server/auth/providers/discord.py
+++ b/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
diff --git a/src/fastmcp/server/auth/providers/github.py b/src/fastmcp/server/auth/providers/github.py
index abaaa439a..01331ba25 100644
--- a/src/fastmcp/server/auth/providers/github.py
+++ b/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
diff --git a/src/fastmcp/server/auth/providers/google.py b/src/fastmcp/server/auth/providers/google.py
index 80deac6e3..0dd509e33 100644
--- a/src/fastmcp/server/auth/providers/google.py
+++ b/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
diff --git a/src/fastmcp/server/auth/providers/introspection.py b/src/fastmcp/server/auth/providers/introspection.py
index 707e25471..b6f03a105 100644
--- a/src/fastmcp/server/auth/providers/introspection.py
+++ b/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 = {
diff --git a/src/fastmcp/server/auth/providers/jwt.py b/src/fastmcp/server/auth/providers/jwt.py
index 828b9238f..90e1d608f 100644
--- a/src/fastmcp/server/auth/providers/jwt.py
+++ b/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()
diff --git a/src/fastmcp/server/auth/providers/workos.py b/src/fastmcp/server/auth/providers/workos.py
index 4354d405b..48ed825e1 100644
--- a/src/fastmcp/server/auth/providers/workos.py
+++ b/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
diff --git a/tests/server/auth/providers/test_http_client.py b/tests/server/auth/providers/test_http_client.py
new file mode 100644
index 000000000..39c8d51b0
--- /dev/null
+++ b/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