Merge pull request #652 from jlowin/bearer-env

src/fastmcp/server/auth/auth.py

@@ -5,7 +5,6 @@ from mcp.server.auth.provider import (
     RefreshToken,
 )
 from mcp.server.auth.settings import (
-    AuthSettings,
     ClientRegistrationOptions,
     RevocationOptions,
 )
@@ -39,10 +38,8 @@ class OAuthProvider(
         if isinstance(service_documentation_url, str):
             service_documentation_url = AnyHttpUrl(service_documentation_url)
 
-        self.settings = AuthSettings(
-            issuer_url=issuer_url,
-            service_documentation_url=service_documentation_url,
-            client_registration_options=client_registration_options,
-            revocation_options=revocation_options,
-            required_scopes=required_scopes,
-        )
+        self.issuer_url = issuer_url
+        self.service_documentation_url = service_documentation_url
+        self.client_registration_options = client_registration_options
+        self.revocation_options = revocation_options
+        self.required_scopes = required_scopes

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

@@ -1,25 +1,3 @@
-"""
-Simple JWT Bearer Token validation for hosted MCP servers.
-
-Uses RS256 (asymmetric) where your control plane signs with a private key
-and hosted MCP servers validate with the corresponding public key.
-
-Example usage:
-# Static public key
-provider = BearerAuthProvider(
-    public_key='''-----BEGIN PUBLIC KEY-----
-    MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEA...
-    -----END PUBLIC KEY-----''',
-    issuer="https://auth.yourservice.com"
-)
-
-# Or JWKS URI (recommended for production - allows key rotation)
-provider = Bear(
-    jwks_uri="https://auth.yourservice.com/.well-known/jwks.json",
-    issuer="https://auth.yourservice.com"
-)
-"""
-
 import time
 from dataclasses import dataclass
 from typing import Any, TypedDict
@@ -165,7 +143,6 @@ class RSAKeyPair:
             payload,
             key=self.private_key.get_secret_value(),
         )
-
         return token_bytes.decode("utf-8")
 
 
@@ -174,25 +151,28 @@ class BearerAuthProvider(OAuthProvider):
     Simple JWT Bearer Token validator for hosted MCP servers.
     Uses RS256 asymmetric encryption. Supports either static public key
     or JWKS URI for key rotation.
+
+    Note that this provider DOES NOT permit client registration or revocation, or any OAuth flows.
+    It is intended to be used with a control plane that manages clients and tokens.
     """
 
     def __init__(
         self,
-        issuer: str | None = None,
         public_key: str | None = None,
         jwks_uri: str | None = None,
+        issuer: str | None = None,
         audience: str | None = None,
         required_scopes: list[str] | None = None,
     ):
         """
-        Initialize the provider.
+        Initialize the provider. Either public_key or jwks_uri must be provided.
 
         Args:
-            issuer: Expected issuer claim (your control plane)
             public_key: RSA public key in PEM format (for static key)
             jwks_uri: URI to fetch keys from (for key rotation)
+            issuer: Expected issuer claim (optional)
             audience: Expected audience claim (optional)
-            required_scopes: List of required scopes for access
+            required_scopes: List of required scopes for access (optional)
         """
         if not (public_key or jwks_uri):
             raise ValueError("Either public_key or jwks_uri must be provided")
@@ -315,7 +295,8 @@ class BearerAuthProvider(OAuthProvider):
             if exp and exp < time.time():
                 return None
 
-            # Validate issuer
+            # Validate issuer - note we use issuer instead of issuer_url here because
+            # issuer is optional, allowing users to make this check optional
             if self.issuer:
                 if claims.get("iss") != self.issuer:
                     return None

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

@@ -0,0 +1,62 @@
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+from fastmcp.server.auth.providers.bearer import BearerAuthProvider
+
+
+# Sentinel object to indicate that a setting is not set
+class _NotSet:
+    pass
+
+
+class EnvBearerAuthProviderSettings(BaseSettings):
+    """Settings for the BearerAuthProvider."""
+
+    model_config = SettingsConfigDict(
+        env_prefix="FASTMCP_AUTH_BEARER_",
+        env_file=".env",
+        extra="ignore",
+    )
+
+    public_key: str | None = None
+    jwks_uri: str | None = None
+    issuer: str | None = None
+    audience: str | None = None
+    required_scopes: list[str] | None = None
+
+
+class EnvBearerAuthProvider(BearerAuthProvider):
+    """
+    A BearerAuthProvider that loads settings from environment variables. Any
+    providing setting will always take precedence over the environment
+    variables.
+    """
+
+    def __init__(
+        self,
+        public_key: str | None | type[_NotSet] = _NotSet,
+        jwks_uri: str | None | type[_NotSet] = _NotSet,
+        issuer: str | None | type[_NotSet] = _NotSet,
+        audience: str | None | type[_NotSet] = _NotSet,
+        required_scopes: list[str] | None | type[_NotSet] = _NotSet,
+    ):
+        """
+        Initialize the provider.
+
+        Args:
+            public_key: RSA public key in PEM format (for static key)
+            jwks_uri: URI to fetch keys from (for key rotation)
+            issuer: Expected issuer claim (optional)
+            audience: Expected audience claim (optional)
+            required_scopes: List of required scopes for access (optional)
+        """
+        kwargs = {
+            "public_key": public_key,
+            "jwks_uri": jwks_uri,
+            "issuer": issuer,
+            "audience": audience,
+            "required_scopes": required_scopes,
+        }
+        settings = EnvBearerAuthProviderSettings(
+            **{k: v for k, v in kwargs.items() if v is not _NotSet}
+        )
+        super().__init__(**settings.model_dump())

src/fastmcp/server/http.py

@@ -91,15 +91,15 @@ def setup_auth_middleware_and_routes(
         Middleware(AuthContextMiddleware),
     ]
 
-    required_scopes = auth.settings.required_scopes or []
+    required_scopes = auth.required_scopes or []
 
     auth_routes.extend(
         create_auth_routes(
             provider=auth,
-            issuer_url=auth.settings.issuer_url,
-            service_documentation_url=auth.settings.service_documentation_url,
-            client_registration_options=auth.settings.client_registration_options,
-            revocation_options=auth.settings.revocation_options,
+            issuer_url=auth.issuer_url,
+            service_documentation_url=auth.service_documentation_url,
+            client_registration_options=auth.client_registration_options,
+            revocation_options=auth.revocation_options,
         )
     )
 

src/fastmcp/server/server.py

@@ -48,6 +48,7 @@ from fastmcp.prompts.prompt import PromptResult
 from fastmcp.resources import Resource, ResourceManager
 from fastmcp.resources.template import ResourceTemplate
 from fastmcp.server.auth.auth import OAuthProvider
+from fastmcp.server.auth.providers.bearer_env import EnvBearerAuthProvider
 from fastmcp.server.http import (
     StarletteWithLifespan,
     create_sse_app,
@@ -186,6 +187,8 @@ class FastMCP(Generic[LifespanResultT]):
             lifespan=_lifespan_wrapper(self, lifespan),
         )
 
+        if auth is None and self.settings.default_auth_provider == "bearer_env":
+            auth = EnvBearerAuthProvider()
         self.auth = auth
 
         if tools:

src/fastmcp/settings.py

@@ -5,7 +5,10 @@ from pathlib import Path
 from typing import Annotated, Literal
 
 from pydantic import Field, model_validator
-from pydantic_settings import BaseSettings, SettingsConfigDict
+from pydantic_settings import (
+    BaseSettings,
+    SettingsConfigDict,
+)
 from typing_extensions import Self
 
 LOG_LEVEL = Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]
@@ -176,5 +179,24 @@ class ServerSettings(BaseSettings):
         False  # If True, uses true stateless mode (new transport per request)
     )
 
+    # Auth settings
+    default_auth_provider: Annotated[
+        Literal["bearer_env"] | None,
+        Field(
+            description=inspect.cleandoc(
+                """
+                Configure the authentication provider. This setting is intended only to
+                be used for remote confirugation of providers that fully support
+                environment variable configuration.
+
+                If None, no automatic configuration will take place.
+
+                This setting is *always* overriden by any auth provider passed to the
+                FastMCP constructor.
+                """
+            ),
+        ),
+    ] = None
+
 
 settings = Settings()

tests/auth/providers/test_bearer_env.py

@@ -0,0 +1,82 @@
+import pytest
+from pydantic import AnyHttpUrl, ValidationError
+
+from fastmcp import FastMCP
+from fastmcp.server.auth.providers.bearer import BearerAuthProvider
+from fastmcp.server.auth.providers.bearer_env import EnvBearerAuthProvider
+
+
+def test_load_bearer_env_from_env_var(monkeypatch):
+    mcp = FastMCP()
+    assert mcp.auth is None
+
+    monkeypatch.setenv("FASTMCP_SERVER_DEFAULT_AUTH_PROVIDER", "bearer_env")
+    monkeypatch.setenv("FASTMCP_AUTH_BEARER_PUBLIC_KEY", "test-public-key")
+
+    mcp_with_auth = FastMCP()
+    assert isinstance(mcp_with_auth.auth, EnvBearerAuthProvider)
+
+
+def test_load_bearer_env_from_env_var_requires_public_key_or_jwks_uri(monkeypatch):
+    mcp = FastMCP()
+    assert mcp.auth is None
+
+    monkeypatch.setenv("FASTMCP_SERVER_DEFAULT_AUTH_PROVIDER", "bearer_env")
+
+    with pytest.raises(
+        ValueError, match="Either public_key or jwks_uri must be provided"
+    ):
+        FastMCP()
+
+
+def test_configure_bearer_env_from_env_var(monkeypatch):
+    monkeypatch.setenv("FASTMCP_SERVER_DEFAULT_AUTH_PROVIDER", "bearer_env")
+    monkeypatch.setenv("FASTMCP_AUTH_BEARER_PUBLIC_KEY", "test-public-key")
+    monkeypatch.setenv("FASTMCP_AUTH_BEARER_ISSUER", "http://test-issuer")
+    monkeypatch.setenv("FASTMCP_AUTH_BEARER_AUDIENCE", "test-audience")
+    monkeypatch.setenv(
+        "FASTMCP_AUTH_BEARER_REQUIRED_SCOPES", '["test-scope1", "test-scope2"]'
+    )
+
+    mcp = FastMCP()
+    assert isinstance(mcp.auth, EnvBearerAuthProvider)
+    assert mcp.auth.public_key == "test-public-key"
+    assert mcp.auth.issuer_url == AnyHttpUrl("http://test-issuer")
+    assert mcp.auth.audience == "test-audience"
+    assert mcp.auth.required_scopes == ["test-scope1", "test-scope2"]
+
+
+def test_list_of_scopes_must_be_a_list(monkeypatch):
+    monkeypatch.setenv("FASTMCP_SERVER_DEFAULT_AUTH_PROVIDER", "bearer_env")
+    monkeypatch.setenv("FASTMCP_AUTH_BEARER_REQUIRED_SCOPES", "test-scope1")
+
+    with pytest.raises(ValidationError, match="Input should be a valid list"):
+        FastMCP()
+
+
+def test_configure_bearer_env_jwks_uri_from_env_var(monkeypatch):
+    monkeypatch.setenv("FASTMCP_SERVER_DEFAULT_AUTH_PROVIDER", "bearer_env")
+    monkeypatch.setenv("FASTMCP_AUTH_BEARER_JWKS_URI", "test-jwks-uri")
+
+    mcp = FastMCP()
+    assert isinstance(mcp.auth, EnvBearerAuthProvider)
+    assert mcp.auth.jwks_uri == "test-jwks-uri"
+
+
+def test_configure_bearer_env_public_key_and_jwks_uri_error(monkeypatch):
+    monkeypatch.setenv("FASTMCP_SERVER_DEFAULT_AUTH_PROVIDER", "bearer_env")
+    monkeypatch.setenv("FASTMCP_AUTH_BEARER_PUBLIC_KEY", "test-public-key")
+    monkeypatch.setenv("FASTMCP_AUTH_BEARER_JWKS_URI", "test-jwks-uri")
+
+    with pytest.raises(ValueError, match="Provide either public_key or jwks_uri"):
+        FastMCP()
+
+
+def test_provided_auth_takes_precedence_over_env_vars(monkeypatch):
+    monkeypatch.setenv("FASTMCP_SERVER_DEFAULT_AUTH_PROVIDER", "bearer_env")
+    monkeypatch.setenv("FASTMCP_AUTH_BEARER_PUBLIC_KEY", "test-public-key")
+
+    mcp = FastMCP(auth=BearerAuthProvider(public_key="test-public-key-2"))
+    assert isinstance(mcp.auth, BearerAuthProvider)
+    assert not isinstance(mcp.auth, EnvBearerAuthProvider)
+    assert mcp.auth.public_key == "test-public-key-2"