Support configuring bearer auth from env vars

This commit is contained in:
Jeremiah Lowin 2025-06-01 16:24:45 -04:00
commit db18baa24f
7 changed files with 160 additions and 34 deletions

View file

@ -39,7 +39,7 @@ class OAuthProvider(
if isinstance(service_documentation_url, str):
service_documentation_url = AnyHttpUrl(service_documentation_url)
self.settings = AuthSettings(
self.auth_settings = AuthSettings(
issuer_url=issuer_url,
service_documentation_url=service_documentation_url,
client_registration_options=client_registration_options,

View file

@ -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")

View file

@ -0,0 +1,54 @@
from enum import Enum
from pydantic_settings import BaseSettings, SettingsConfigDict
from fastmcp.server.auth.providers.bearer import BearerAuthProvider
class NotSet(Enum):
sentinel = 0
NOTSET = NotSet.sentinel
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.
"""
def __init__(
self,
public_key: str | None | NotSet = NOTSET,
jwks_uri: str | None | NotSet = NOTSET,
issuer: str | None | NotSet = NOTSET,
audience: str | None | NotSet = NOTSET,
required_scopes: list[str] | None | NotSet = NOTSET,
):
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())

View file

@ -91,15 +91,15 @@ def setup_auth_middleware_and_routes(
Middleware(AuthContextMiddleware),
]
required_scopes = auth.settings.required_scopes or []
required_scopes = auth.auth_settings.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.auth_settings.issuer_url,
service_documentation_url=auth.auth_settings.service_documentation_url,
client_registration_options=auth.auth_settings.client_registration_options,
revocation_options=auth.auth_settings.revocation_options,
)
)

View file

@ -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.auth_provider == "bearer_env":
auth = EnvBearerAuthProvider()
self.auth = auth
if tools:

View file

@ -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,8 @@ class ServerSettings(BaseSettings):
False # If True, uses true stateless mode (new transport per request)
)
# Auth settings
auth_provider: Literal["bearer_env"] | None = None
settings = Settings()

View file

@ -0,0 +1,83 @@
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_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_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_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 == "http://test-issuer"
assert mcp.auth.auth_settings.issuer_url == AnyHttpUrl("http://test-issuer")
assert mcp.auth.audience == "test-audience"
assert mcp.auth.auth_settings.required_scopes == ["test-scope1", "test-scope2"]
def test_list_of_scopes_must_be_a_list(monkeypatch):
monkeypatch.setenv("FASTMCP_SERVER_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_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_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_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"