Google Workspace MCP Server

""" Google OAuth2 Authentication Module Handles authentication for all Google Workspace APIs """ import os import json from google.auth.transport.requests import Request from google.oauth2.credentials import Credentials from google_auth_oauthlib.flow import InstalledAppFlow from googleapiclient.discovery import build # Full Google Workspace scopes SCOPES = [ # Gmail 'https://www.googleapis.com/auth/gmail.readonly', 'https://www.googleapis.com/auth/gmail.send', 'https://www.googleapis.com/auth/gmail.modify', # Calendar 'https://www.googleapis.com/auth/calendar', 'https://www.googleapis.com/auth/calendar.events', # Sheets 'https://www.googleapis.com/auth/spreadsheets', # Docs 'https://www.googleapis.com/auth/documents', # Drive 'https://www.googleapis.com/auth/drive', 'https://www.googleapis.com/auth/drive.file', ] TOKEN_PATH = os.path.join(os.path.dirname(__file__), '..', 'tokens.json') CREDENTIALS_PATH = os.path.join(os.path.dirname(__file__), '..', 'credentials.json') def get_credentials(): """Get valid credentials, refreshing or re-authenticating as needed""" creds = None if os.path.exists(TOKEN_PATH): creds = Credentials.from_authorized_user_file(TOKEN_PATH, SCOPES) if not creds or not creds.valid: if creds and creds.expired and creds.refresh_token: creds.refresh(Request()) else: if not os.path.exists(CREDENTIALS_PATH): raise FileNotFoundError( f'credentials.json not found at {CREDENTIALS_PATH}. ' 'Please create it with your OAuth2 credentials from Google Cloud Console.' ) flow = InstalledAppFlow.from_client_secrets_file( CREDENTIALS_PATH, SCOPES) creds = flow.run_local_server(port=0) save_token(creds) return creds def save_token(creds): """Save credentials to token file""" with open(TOKEN_PATH, 'w') as token: token.write(creds.to_json()) def get_auth_url(): """Get authorization URL for OAuth2 flow""" if not os.path.exists(CREDENTIALS_PATH): raise FileNotFoundError('credentials.json not found') flow = InstalledAppFlow.from_client_secrets_file( CREDENTIALS_PATH, SCOPES) authorization_url, _ = flow.authorization_url( access_type='offline', include_granted_scopes='true', prompt='consent' # Force consent to get new scopes ) return authorization_url def get_token_from_code(code): """Exchange authorization code for credentials""" flow = InstalledAppFlow.from_client_secrets_file( CREDENTIALS_PATH, SCOPES) flow.fetch_token(code=code) creds = flow.credentials save_token(creds) return creds def is_authorized(): """Check if user is authorized with valid credentials""" try: creds = get_credentials() return creds and creds.valid except Exception: return False def revoke_credentials(): """Revoke current credentials and delete token file""" try: if os.path.exists(TOKEN_PATH): creds = Credentials.from_authorized_user_file(TOKEN_PATH, SCOPES) if creds: import requests requests.post( 'https://oauth2.googleapis.com/revoke', params={'token': creds.token}, headers={'content-type': 'application/x-www-form-urlencoded'} ) os.remove(TOKEN_PATH) return {'success': True, 'message': 'Credentials revoked and token deleted'} except Exception as error: return {'success': False, 'error': str(error)} def get_current_scopes(): """Get the scopes from current token""" try: if os.path.exists(TOKEN_PATH): with open(TOKEN_PATH, 'r') as f: token_data = json.load(f) return { 'success': True, 'scopes': token_data.get('scopes', []), 'required_scopes': SCOPES } return {'success': False, 'error': 'No token file found'} except Exception as error: return {'success': False, 'error': str(error)} def needs_reauth(): """Check if re-authentication is needed for new scopes""" try: if not os.path.exists(TOKEN_PATH): return True with open(TOKEN_PATH, 'r') as f: token_data = json.load(f) current_scopes = set(token_data.get('scopes', [])) required_scopes = set(SCOPES) # Check if all required scopes are present missing = required_scopes - current_scopes return len(missing) > 0 except Exception: return True def force_reauth(): """Force re-authentication to get new scopes""" try: # Delete existing token if os.path.exists(TOKEN_PATH): os.remove(TOKEN_PATH) # Get new credentials creds = get_credentials() return { 'success': True, 'message': 'Re-authenticated successfully with new scopes', 'scopes': SCOPES } except Exception as error: return {'success': False, 'error': str(error)}