"""Google OAuth provider for FastMCP. This module provides a complete Google OAuth integration that's ready to use with just a client ID and client secret. It handles all the complexity of Google's OAuth flow, token validation, and user management. Example: ```python from fastmcp import FastMCP from fastmcp.server.auth.providers.google import GoogleProvider # Simple Google OAuth protection auth = GoogleProvider( client_id="your-google-client-id.apps.googleusercontent.com", client_secret="your-google-client-secret" ) mcp = FastMCP("My Protected Server", auth=auth) ``` """ from __future__ import annotations import contextlib import time 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__) GOOGLE_SCOPE_ALIASES: dict[str, str] = { "email": "https://www.googleapis.com/auth/userinfo.email", "profile": "https://www.googleapis.com/auth/userinfo.profile", } def _normalize_google_scope(scope: str) -> str: """Normalize a Google scope shorthand to its canonical full URI. Google accepts shorthand scopes like "email" and "profile" in authorization requests, but returns the full URI form in token responses. This normalizes to the full URI so comparisons work regardless of which form was used. """ return GOOGLE_SCOPE_ALIASES.get(scope, scope) class GoogleTokenVerifier(TokenVerifier): """Token verifier for Google OAuth tokens. Google OAuth tokens are opaque (not JWTs), so we verify them by calling Google's tokeninfo endpoint with the access token as a query parameter. This returns the OAuth app ID (``aud``), granted scopes, and expiry time. User profile data (name, picture, etc.) is fetched separately from the v2 userinfo endpoint when the token is valid. """ def __init__( self, *, 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. """ normalized = ( [_normalize_google_scope(s) for s in required_scopes] if required_scopes else required_scopes ) super().__init__(required_scopes=normalized) self.timeout_seconds = timeout_seconds self._http_client = http_client async def verify_token(self, token: str) -> AccessToken | None: """Verify a Google OAuth token using the tokeninfo endpoint. Calls ``https://oauth2.googleapis.com/tokeninfo?access_token=TOKEN`` to validate the token and retrieve the OAuth app ID (``aud``), granted scopes, and expiry time. On success, fetches user profile data from the v2 userinfo endpoint to populate name, picture, and locale claims. """ 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: Verify token via tokeninfo endpoint. # Returns aud (OAuth app ID), scope (space-separated), expires_in, sub, email. response = await client.get( "https://oauth2.googleapis.com/tokeninfo", params={"access_token": token}, headers={"User-Agent": "FastMCP-Google-OAuth"}, ) if response.status_code != 200: logger.debug( "Google token verification failed: %d", response.status_code, ) return None token_data = response.json() # aud is the OAuth app ID (client_id / audience) aud = token_data.get("aud") if not aud: logger.debug("Google tokeninfo missing 'aud' claim") return None # sub is required (unique Google user ID) sub = token_data.get("sub") if not sub: logger.debug("Google tokeninfo missing 'sub' claim") return None # Parse scopes directly from the tokeninfo response (space-separated) scope_str = token_data.get("scope", "") token_scopes = scope_str.split() if scope_str else [] # Check required scopes if self.required_scopes: token_scopes_set = set(token_scopes) required_scopes_set = set(self.required_scopes) if not required_scopes_set.issubset(token_scopes_set): logger.debug( "Google token missing required scopes. Has %d, needs %d", len(token_scopes_set), len(required_scopes_set), ) return None # Compute expiry from expires_in (seconds until expiry) expires_at: int | None = None expires_in = token_data.get("expires_in") if expires_in is not None: with contextlib.suppress(ValueError, TypeError): expires_at = int(time.time()) + int(expires_in) # Step 2: Fetch user profile from v2 userinfo endpoint. # tokeninfo provides auth data; userinfo provides name, picture, locale. user_data: dict = {} try: userinfo_response = await client.get( "https://www.googleapis.com/oauth2/v2/userinfo", headers={ "Authorization": f"Bearer {token}", "User-Agent": "FastMCP-Google-OAuth", }, ) if userinfo_response.status_code == 200: user_data = userinfo_response.json() except Exception as e: logger.debug("Failed to fetch Google user profile: %s", e) access_token = AccessToken( token=token, client_id=sub, scopes=token_scopes, expires_at=expires_at, claims={ "sub": sub, "aud": aud, "email": token_data.get("email") or user_data.get("email"), "email_verified": token_data.get("email_verified") or user_data.get("verified_email"), "name": user_data.get("name"), "picture": user_data.get("picture"), "given_name": user_data.get("given_name"), "family_name": user_data.get("family_name"), "locale": user_data.get("locale"), "google_user_data": user_data or None, }, ) logger.debug("Google token verified successfully") return access_token except httpx.RequestError as e: logger.debug("Failed to verify Google token: %s", e) return None except Exception as e: logger.debug("Google token verification error: %s", e) return None class GoogleProvider(OAuthProxy): """Complete Google OAuth provider for FastMCP. This provider makes it trivial to add Google OAuth protection to any FastMCP server. Just provide your Google OAuth app credentials and a base URL, and you're ready to go. Features: - Transparent OAuth proxy to Google - Automatic token validation via Google's tokeninfo API - User information extraction from Google APIs - Minimal configuration required Example: ```python from fastmcp import FastMCP from fastmcp.server.auth.providers.google import GoogleProvider auth = GoogleProvider( client_id="123456789.apps.googleusercontent.com", client_secret="GOCSPX-abc123...", base_url="https://my-server.com" ) mcp = FastMCP("My App", auth=auth) ``` """ def __init__( self, *, client_id: str, client_secret: str | None = None, base_url: AnyHttpUrl | str, resource_base_url: AnyHttpUrl | str | None = None, 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["remember", "external"] = True, consent_csp_policy: str | None = None, forward_resource: bool = True, fallback_refresh_token_expiry_seconds: int | None = None, fastmcp_access_token_expiry_seconds: int | None = None, token_expiry_threshold_seconds: int = 0, extra_authorize_params: dict[str, str] | None = None, http_client: httpx.AsyncClient | None = None, enable_cimd: bool = True, ): """Initialize Google OAuth provider. Args: client_id: Google OAuth client ID (e.g., "123456789.apps.googleusercontent.com") client_secret: Google OAuth client secret (e.g., "GOCSPX-abc123..."). Optional for PKCE public clients (e.g., native apps). When omitted, jwt_signing_key must be provided. base_url: Public URL where OAuth endpoints will be accessible (includes any mount path) resource_base_url: Optional public base URL for the protected resource metadata and token audience. Defaults to ``base_url``. 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 Google OAuth app (defaults to "/auth/callback") required_scopes: Required Google scopes (defaults to ["openid"]). Common scopes include: - "openid" for OpenID Connect (default) - "https://www.googleapis.com/auth/userinfo.email" for email access - "https://www.googleapis.com/auth/userinfo.profile" for profile info Google scope shorthands like "email" and "profile" are automatically normalized to their full URI forms for token verification. valid_scopes: All scopes that clients are allowed to request, advertised through well-known endpoints. Defaults to required_scopes if not provided. Use this when you want clients to be able to request additional scopes beyond the required minimum. Shorthands are normalized to full URI forms. timeout_seconds: HTTP request timeout for Google 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 True, users see a consent screen before being redirected to Google. When False, authorization proceeds directly without user confirmation. When "external", the built-in consent screen is skipped but no warning is logged, indicating that consent is handled externally (e.g. by Google's own consent). SECURITY WARNING: Only set to False for local development or testing environments. extra_authorize_params: Additional parameters to forward to Google's authorization endpoint. 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. enable_cimd: Enable CIMD (Client ID Metadata Document) support for URL-based client IDs (default True). Set to False to disable. fallback_refresh_token_expiry_seconds: Lifetime for the FastMCP-issued refresh token when the upstream provider omits `refresh_expires_in` (e.g. Cognito, GitHub, many OIDC IdPs). Defaults to 1 year. The upstream refresh remains the source of truth. See `OAuthProxy` for details. fastmcp_access_token_expiry_seconds: Lifetime for the FastMCP-issued access token, decoupling it from the upstream provider's `expires_in`. Defaults to None (mirror the upstream lifetime). Set this for bridges whose upstream issues short-lived access tokens that some MCP clients can't refresh gracefully (e.g. `mcp-remote`). See `OAuthProxy` for details. token_expiry_threshold_seconds: Number of seconds before actual expiry to treat a token as expired, refreshing early to avoid races. Defaults to 0. """ # Parse scopes if provided as string # Google requires at least one scope - openid is the minimal OIDC scope required_scopes_final = ( parse_scopes(required_scopes) if required_scopes is not None else ["openid"] ) # Normalize valid_scopes if provided parsed_valid_scopes = ( parse_scopes(valid_scopes) if valid_scopes is not None else None ) valid_scopes_final = ( [_normalize_google_scope(s) for s in parsed_valid_scopes] if parsed_valid_scopes is not None else None ) # Create Google token verifier # Normalization of shorthand scopes (e.g. "email" -> full URI) happens # inside GoogleTokenVerifier so required_scopes match what Google returns. token_verifier = GoogleTokenVerifier( required_scopes=required_scopes_final, timeout_seconds=timeout_seconds, http_client=http_client, ) # Set Google-specific defaults for extra authorize params # access_type=offline ensures refresh tokens are returned # prompt=consent forces consent screen to get refresh token (Google only issues on first auth otherwise) google_defaults = { "access_type": "offline", "prompt": "consent", } # User-provided params override defaults if extra_authorize_params: google_defaults.update(extra_authorize_params) extra_authorize_params_final = google_defaults # Initialize OAuth proxy with Google endpoints super().__init__( upstream_authorization_endpoint="https://accounts.google.com/o/oauth2/v2/auth", upstream_token_endpoint="https://oauth2.googleapis.com/token", upstream_client_id=client_id, upstream_client_secret=client_secret, token_verifier=token_verifier, base_url=base_url, resource_base_url=resource_base_url, redirect_path=redirect_path, issuer_url=issuer_url or base_url, # Default to base_url if not specified 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, forward_resource=forward_resource, fallback_refresh_token_expiry_seconds=fallback_refresh_token_expiry_seconds, fastmcp_access_token_expiry_seconds=fastmcp_access_token_expiry_seconds, token_expiry_threshold_seconds=token_expiry_threshold_seconds, extra_authorize_params=extra_authorize_params_final, valid_scopes=valid_scopes_final, enable_cimd=enable_cimd, ) logger.debug( "Initialized Google OAuth provider for client %s with scopes: %s", client_id, required_scopes_final, )