How to Automate Gmail Customer Support using the API

1#!/usr/bin/env python3
2"""Gmail Customer Support Automation — Push-based triage and auto-reply."""
3import os
4import base64
5import json
6import logging
7from email.mime.text import MIMEText
8from flask import Flask, request, jsonify
9from google.oauth2 import service_account
10from googleapiclient.discovery import build
11
12logging.basicConfig(level=logging.INFO)
13log = logging.getLogger(__name__)
14app = Flask(__name__)
15
16SCOPES = ['https://www.googleapis.com/auth/gmail.modify', 'https://www.googleapis.com/auth/gmail.send']
17SERVICE_ACCOUNT_FILE = os.environ.get('GOOGLE_APPLICATION_CREDENTIALS', 'service-account-key.json')
18SUPPORT_EMAIL = os.environ.get('SUPPORT_EMAIL', 'support@yourcompany.com')
19GCP_PROJECT = os.environ.get('GCP_PROJECT_ID', 'my-project')
20PUBSUB_TOPIC = 'gmail-support-notifications'
21
22# State (use a database in production)
23processed_message_ids = set()
24last_history_id = None
25auto_replied_label_id = None
26escalated_label_id = None
27
28def get_service(user_email):
29    creds = service_account.Credentials.from_service_account_file(
30        SERVICE_ACCOUNT_FILE, scopes=SCOPES
31    ).with_subject(user_email)
32    return build('gmail', 'v1', credentials=creds)
33
34def setup_labels(service):
35    global auto_replied_label_id, escalated_label_id
36    labels = service.users().labels().list(userId=SUPPORT_EMAIL).execute().get('labels', [])
37    names = {l['name']: l['id'] for l in labels}
38    if 'Support/Auto-Replied' not in names:
39        r = service.users().labels().create(userId=SUPPORT_EMAIL,
40            body={'name': 'Support/Auto-Replied'}).execute()
41        auto_replied_label_id = r['id']
42    else:
43        auto_replied_label_id = names['Support/Auto-Replied']
44    if 'Support/Escalated' not in names:
45        r = service.users().labels().create(userId=SUPPORT_EMAIL,
46            body={'name': 'Support/Escalated'}).execute()
47        escalated_label_id = r['id']
48    else:
49        escalated_label_id = names['Support/Escalated']
50
51def get_email_body(message):
52    payload = message.get('payload', {})
53    if payload.get('mimeType') == 'text/plain' and payload.get('body', {}).get('data'):
54        return base64.urlsafe_b64decode(payload['body']['data'] + '==').decode('utf-8', errors='ignore')
55    for part in payload.get('parts', []):
56        if part['mimeType'] == 'text/plain' and part.get('body', {}).get('data'):
57            return base64.urlsafe_b64decode(part['body']['data'] + '==').decode('utf-8', errors='ignore')
58    return ''
59
60def process_message(service, message_id):
61    global processed_message_ids
62    if message_id in processed_message_ids:
63        log.info(f'Skipping already-processed message: {message_id}')
64        return
65    msg = service.users().messages().get(userId=SUPPORT_EMAIL, id=message_id, format='full').execute()
66    labels = msg.get('labelIds', [])
67    if auto_replied_label_id in labels or escalated_label_id in labels:
68        return
69    headers = {h['name']: h['value'] for h in msg['payload']['headers']}
70    subject = headers.get('Subject', '')
71    sender = headers.get('From', '')
72    msg_id_header = headers.get('Message-ID', '')
73    body = get_email_body(msg)
74    combined = (subject + ' ' + body).lower()
75    escalate = any(kw in combined for kw in ['urgent', 'broken', 'not working', 'outage'])
76    if escalate:
77        service.users().messages().modify(userId=SUPPORT_EMAIL, id=message_id,
78            body={'addLabelIds': [escalated_label_id], 'removeLabelIds': ['UNREAD']}).execute()
79        log.info(f'Escalated: {subject}')
80    else:
81        if 'password' in combined:
82            reply_body = 'To reset your password: https://app.example.com/reset'
83        elif 'billing' in combined:
84            reply_body = 'For billing questions: https://app.example.com/billing'
85        else:
86            reply_body = 'Thank you for contacting support. We will respond within 24 hours.'
87        reply = MIMEText(reply_body, 'plain', 'utf-8')
88        reply['From'] = SUPPORT_EMAIL
89        reply['To'] = sender
90        reply['Subject'] = f'Re: {subject}'
91        reply['In-Reply-To'] = msg_id_header
92        reply['References'] = msg_id_header
93        raw = base64.urlsafe_b64encode(reply.as_bytes()).decode()
94        service.users().messages().send(userId=SUPPORT_EMAIL,
95            body={'raw': raw, 'threadId': msg['threadId']}).execute()
96        service.users().messages().modify(userId=SUPPORT_EMAIL, id=message_id,
97            body={'addLabelIds': [auto_replied_label_id], 'removeLabelIds': ['UNREAD', 'INBOX']}).execute()
98        log.info(f'Auto-replied: {subject}')
99    processed_message_ids.add(message_id)
100
101@app.route('/gmail-webhook', methods=['POST'])
102def webhook():
103    global last_history_id
104    res = jsonify({})
105    res.status_code = 204
106    envelope = request.get_json(silent=True)
107    if not envelope or 'message' not in envelope:
108        return res
109    try:
110        data = base64.urlsafe_b64decode(envelope['message']['data'] + '==').decode()
111        notification = json.loads(data)
112        history_id = notification.get('historyId')
113        service = get_service(SUPPORT_EMAIL)
114        start_id = last_history_id or history_id
115        result = service.users().history().list(
116            userId=SUPPORT_EMAIL, startHistoryId=start_id,
117            historyTypes=['messageAdded'], labelId='INBOX'
118        ).execute()
119        for record in result.get('history', []):
120            for added in record.get('messagesAdded', []):
121                if 'UNREAD' in added['message'].get('labelIds', []):
122                    process_message(service, added['message']['id'])
123        last_history_id = result.get('historyId', history_id)
124    except Exception as e:
125        log.error(f'Webhook error: {e}')
126    return res
127
128@app.route('/setup', methods=['POST'])
129def setup():
130    global last_history_id
131    service = get_service(SUPPORT_EMAIL)
132    setup_labels(service)
133    result = service.users().watch(userId=SUPPORT_EMAIL, body={
134        'topicName': f'projects/{GCP_PROJECT}/topics/{PUBSUB_TOPIC}',
135        'labelIds': ['INBOX'], 'labelFilterBehavior': 'INCLUDE'
136    }).execute()
137    last_history_id = result['historyId']
138    return jsonify({'status': 'ok', 'historyId': last_history_id})
139
140if __name__ == '__main__':
141    app.run(port=8080)