# Copyright (c) 2025 Dedalus Labs, Inc. and its contributors
# SPDX-License-Identifier: MIT
"""Google OAuth token management for gmail-mcp.
Supports:
- Reading stored OAuth credentials (access + refresh token)
- Refreshing access tokens automatically when expired
- Running the Installed App (Desktop) OAuth flow to obtain tokens
Environment variables:
- GOOGLE_OAUTH_CREDENTIALS: path to OAuth client JSON downloaded from Google Cloud Console
- GMAIL_TOKEN_PATH (optional): where to store tokens (defaults to ~/.config/gmail-mcp/tokens.json)
- GMAIL_SCOPES (optional): comma-separated scopes (defaults to gmail.readonly)
Optional/manual override:
- GMAIL_ACCESS_TOKEN: if set, used directly (not recommended; expires quickly)
"""
from __future__ import annotations
import json
import os
from pathlib import Path
from google.auth.transport.requests import Request
from google.oauth2.credentials import Credentials
from google_auth_oauthlib.flow import InstalledAppFlow
DEFAULT_SCOPES = ("https://www.googleapis.com/auth/gmail.readonly",)
def _default_token_path() -> Path:
return Path.home() / ".config" / "gmail-mcp" / "tokens.json"
def _get_scopes() -> tuple[str, ...]:
raw = (os.getenv("GMAIL_SCOPES") or "").strip()
if not raw:
return DEFAULT_SCOPES
scopes = [s.strip() for s in raw.split(",") if s.strip()]
return tuple(scopes) if scopes else DEFAULT_SCOPES
def _load_credentials(token_path: Path, scopes: tuple[str, ...]) -> Credentials | None:
if not token_path.exists():
return None
try:
data = json.loads(token_path.read_text(encoding="utf-8"))
return Credentials.from_authorized_user_info(data, scopes=scopes)
except Exception:
return None
def _save_credentials(token_path: Path, creds: Credentials) -> None:
token_path.parent.mkdir(parents=True, exist_ok=True)
token_path.write_text(creds.to_json(), encoding="utf-8")
def ensure_gmail_access_token(*, interactive: bool) -> str:
"""Ensure `GMAIL_ACCESS_TOKEN` exists and is fresh, returning the access token."""
existing = (os.getenv("GMAIL_ACCESS_TOKEN") or "").strip()
if existing:
return existing
scopes = _get_scopes()
token_path = Path(os.getenv("GMAIL_TOKEN_PATH") or _default_token_path())
creds = _load_credentials(token_path, scopes)
# Refresh if possible
if creds and creds.expired and creds.refresh_token:
creds.refresh(Request())
_save_credentials(token_path, creds)
# Interactive flow if needed
if (not creds) or (not creds.valid):
if not interactive:
raise RuntimeError(
"Gmail credentials not available. "
"Run `uv run python -m src.gmail_auth` (interactive browser login) "
"or set GMAIL_ACCESS_TOKEN manually."
)
creds_path = (os.getenv("GOOGLE_OAUTH_CREDENTIALS") or "").strip()
if not creds_path:
raise RuntimeError(
"Missing GOOGLE_OAUTH_CREDENTIALS. "
"Set it to the path of your downloaded OAuth client JSON (Desktop app)."
)
flow = InstalledAppFlow.from_client_secrets_file(creds_path, scopes=list(scopes))
creds = flow.run_local_server(port=0, prompt="consent", access_type="offline")
_save_credentials(token_path, creds)
token = (creds.token or "").strip() if creds else ""
if not token:
raise RuntimeError("Failed to obtain Gmail access token (empty token).")
os.environ["GMAIL_ACCESS_TOKEN"] = token
return token
