Подключение к Gmail через API

Почему вместо SMTP следует использовать Gmail API?

# Install required libraries:
# pip install google-api-python-client google-auth google-auth-oauthlib

# The Gmail API lets agents:
# - List and search messages (labels, queries)
# - Read full message content and attachments
# - Send messages via OAuth (no password needed)
# - Manage labels and threads
# - Watch for new messages via push notifications

print('Gmail API is part of Google Workspace APIs')

Аутентификация: сервисный аккаунт и аутентификация пользователя

# Service Account approach:
# 1. Go to Google Cloud Console -> APIs & Services -> Credentials
# 2. Create a Service Account
# 3. Download the JSON key file
# 4. In Google Workspace Admin: enable domain-wide delegation
# 5. Grant required scopes to the service account

# User OAuth approach:
# 1. Create OAuth 2.0 Client ID (Desktop or Web App type)
# 2. Download credentials.json
# 3. First run: user sees consent screen and grants access
# 4. Agent stores token.json with refresh token for subsequent runs

print('Choose service account for org automation, OAuth for personal Gmail')

Структура JSON учётных данных OAuth2

# credentials.json structure (User OAuth — Desktop app type):
# {
#   "installed": {
#     "client_id": "123456789.apps.googleusercontent.com",
#     "client_secret": "GOCSPX-abc123xyz",
#     "redirect_uris": ["urn:ietf:wg:oauth:2.0:oob", "http://localhost"],
#     "auth_uri": "https://accounts.google.com/o/oauth2/auth",
#     "token_uri": "https://oauth2.googleapis.com/token"
#   }
# }

# service-account.json structure:
# {
#   "type": "service_account",
#   "project_id": "my-project",
#   "private_key_id": "abc123",
#   "private_key": "-----BEGIN PRIVATE KEY-----\n...",
#   "client_email": "agent@my-project.iam.gserviceaccount.com",
#   "client_id": "..."
# }

print('Store credential files outside your git repository')

Области доступа Gmail API

# Gmail API scope constants
SCOPE_READONLY = 'https://www.googleapis.com/auth/gmail.readonly'
SCOPE_SEND = 'https://www.googleapis.com/auth/gmail.send'
SCOPE_MODIFY = 'https://www.googleapis.com/auth/gmail.modify'
SCOPE_COMPOSE = 'https://www.googleapis.com/auth/gmail.compose'

# Calendar scopes (often used alongside Gmail)
SCOPE_CALENDAR_READ = 'https://www.googleapis.com/auth/calendar.readonly'
SCOPE_CALENDAR_EVENTS = 'https://www.googleapis.com/auth/calendar.events'

# Combine scopes your agent actually needs
AGENT_SCOPES = [
    SCOPE_READONLY,
    SCOPE_SEND,
    SCOPE_CALENDAR_EVENTS
]
print(f'Using {len(AGENT_SCOPES)} scopes')

OAuth пользователя: поток первой аутентификации

from google_auth_oauthlib.flow import InstalledAppFlow
from google.auth.transport.requests import Request
from google.oauth2.credentials import Credentials
import os

SCOPES = ['https://www.googleapis.com/auth/gmail.readonly']

def get_credentials(token_file='token.json', creds_file='credentials.json'):
    creds = None

    # Load existing token if available
    if os.path.exists(token_file):
        creds = Credentials.from_authorized_user_file(token_file, SCOPES)

    # Refresh or re-authenticate if needed
    if not creds or not creds.valid:
        if creds and creds.expired and creds.refresh_token:
            creds.refresh(Request())  # auto-refresh
        else:
            # Opens browser for user consent (first time only)
            flow = InstalledAppFlow.from_client_secrets_file(
                creds_file, SCOPES
            )
            creds = flow.run_local_server(port=0)
        # Save token for next run
        with open(token_file, 'w') as f:
            f.write(creds.to_json())

    return creds

Аутентификация сервисного аккаунта

from google.oauth2 import service_account
import os

SCOPES = [
    'https://www.googleapis.com/auth/gmail.readonly',
    'https://www.googleapis.com/auth/gmail.send'
]

def get_service_account_credentials(impersonate_user):
    service_account_file = os.environ.get(
        'GOOGLE_SERVICE_ACCOUNT_JSON',
        'service-account.json'
    )

    credentials = service_account.Credentials.from_service_account_file(
        service_account_file,
        scopes=SCOPES
    )

    # Impersonate a real user (requires domain-wide delegation in Admin)
    delegated = credentials.with_subject(impersonate_user)
    return delegated

creds = get_service_account_credentials('agent@yourcompany.com')
print('Service account credentials ready')

Создание объекта сервиса Gmail

from googleapiclient.discovery import build

def build_gmail_service(credentials):
    service = build(
        'gmail',
        'v1',
        credentials=credentials,
        cache_discovery=False  # avoid file warnings in some environments
    )
    return service

# Full setup: credentials -> service
creds = get_credentials()          # or get_service_account_credentials()
gmail = build_gmail_service(creds)

# Test: get user profile
profile = gmail.users().getProfile(userId='me').execute()
print('Email:', profile['emailAddress'])
print('Total messages:', profile['messagesTotal'])

Создание объекта сервиса Календаря

from googleapiclient.discovery import build

def build_google_services(credentials):
    gmail = build(
        'gmail', 'v1',
        credentials=credentials,
        cache_discovery=False
    )
    calendar = build(
        'calendar', 'v3',
        credentials=credentials,
        cache_discovery=False
    )
    return gmail, calendar

# Use both in one agent
creds = get_credentials()
gmail_service, calendar_service = build_google_services(creds)

# Test calendar access
cal_list = calendar_service.calendarList().list().execute()
for cal in cal_list.get('items', []):
    print(f'Calendar: {cal["summary"]}')

Обработка ошибок API Google

from googleapiclient.errors import HttpError
import json

def safe_gmail_call(service, user_id='me'):
    try:
        profile = service.users().getProfile(userId=user_id).execute()
        return profile
    except HttpError as e:
        status = e.resp.status
        try:
            error_body = json.loads(e.content.decode())
            message = error_body.get('error', {}).get('message', str(e))
        except Exception:
            message = str(e)

        if status == 401:
            print('AUTH ERROR: Credentials invalid or expired')
        elif status == 403:
            print(f'PERMISSION ERROR: {message}')
            print('Check scopes and domain-wide delegation settings')
        elif status == 429:
            print('QUOTA EXCEEDED: Gmail API rate limit hit')
        else:
            print(f'Gmail API error {status}: {message}')
        return None

Квоты и ограничения частоты запросов API Google

import time
from googleapiclient.errors import HttpError

def gmail_call_with_retry(func, max_retries=5):
    for attempt in range(max_retries):
        try:
            return func()
        except HttpError as e:
            if e.resp.status in (429, 500, 503):
                wait = (2 ** attempt) + 1
                print(f'Quota/server error. Waiting {wait}s (attempt {attempt+1})')
                time.sleep(wait)
            elif e.resp.status == 403:
                # Check if it's a quota exceeded vs permission error
                import json
                body = json.loads(e.content.decode())
                reason = body.get('error', {}).get('errors', [{}])[0].get('reason', '')
                if reason == 'rateLimitExceeded':
                    time.sleep(2 ** attempt)
                else:
                    raise  # real permission error, don't retry
            else:
                raise
    raise Exception(f'Gmail API call failed after {max_retries} attempts')

Безопасное хранение учётных данных

import json
import os
from google.oauth2 import service_account

SCOPES = ['https://www.googleapis.com/auth/gmail.readonly']

def get_credentials_from_env():
    # Load service account JSON from environment variable
    sa_json = os.environ.get('GOOGLE_SERVICE_ACCOUNT_JSON')
    if not sa_json:
        raise EnvironmentError(
            'GOOGLE_SERVICE_ACCOUNT_JSON env var not set. '
            'Set it to the contents of your service-account.json'
        )

    sa_info = json.loads(sa_json)
    credentials = service_account.Credentials.from_service_account_info(
        sa_info,
        scopes=SCOPES
    )
    return credentials

# In production: export GOOGLE_SERVICE_ACCOUNT_JSON=$(cat service-account.json)
creds = get_credentials_from_env()
print('Service account loaded from env var')

Быстрая проверка: сервисный аккаунт и OAuth пользователя

Итоги подключения к Gmail API