How to Automate Gmail Sales Outreach using the API

1#!/usr/bin/env python3
2"""Gmail Sales Outreach Automator — personalized sends with daily limit tracking."""
3import os
4import csv
5import base64
6import json
7import time
8import logging
9from datetime import date
10from email.mime.text import MIMEText
11from email.header import Header
12from googleapiclient.discovery import build
13from google.oauth2.credentials import Credentials
14from google_auth_oauthlib.flow import InstalledAppFlow
15from google.auth.transport.requests import Request
16from googleapiclient.errors import HttpError
17
18logging.basicConfig(level=logging.INFO, format='%(asctime)s %(message)s')
19log = logging.getLogger(__name__)
20
21SCOPES = ['https://www.googleapis.com/auth/gmail.send', 'https://www.googleapis.com/auth/gmail.readonly',
22          'https://www.googleapis.com/auth/gmail.labels']
23SENDER_EMAIL = os.environ.get('SENDER_EMAIL', 'me@gmail.com')
24DAILY_LIMIT = 450
25STATE_FILE = 'outreach_state.json'
26LEADS_CSV = 'leads.csv'
27
28TEMPLATE_SUBJECT = 'Quick question for {company}'
29TEMPLATE_BODY = ('Hi {first_name},\n\n'
30    'I noticed {company} {observation}. Teams at this stage often struggle with {pain_point}.\n\n'
31    'We help companies solve this in days, not months. Open to a quick 15-min chat?\n\nBest, Alex')
32
33def get_service():
34    creds = None
35    if os.path.exists('token.json'):
36        creds = Credentials.from_authorized_user_file('token.json', SCOPES)
37    if not creds or not creds.valid:
38        if creds and creds.expired and creds.refresh_token:
39            creds.refresh(Request())
40        else:
41            flow = InstalledAppFlow.from_client_secrets_file('credentials.json', SCOPES)
42            creds = flow.run_local_server(port=0)
43        with open('token.json', 'w') as f: f.write(creds.to_json())
44    return build('gmail', 'v1', credentials=creds)
45
46def get_or_create_label(service, name):
47    labels = service.users().labels().list(userId='me').execute().get('labels', [])
48    for l in labels:
49        if l['name'] == name: return l['id']
50    r = service.users().labels().create(userId='me', body={'name': name}).execute()
51    return r['id']
52
53def load_state():
54    try:
55        with open(STATE_FILE) as f: return json.load(f)
56    except FileNotFoundError:
57        return {'sent': {}, 'daily_counts': {}}
58
59def save_state(state):
60    with open(STATE_FILE, 'w') as f: json.dump(state, f, indent=2)
61
62def encode_message(to, subject, body):
63    msg = MIMEText(body, 'plain', 'utf-8')
64    msg['From'] = SENDER_EMAIL
65    msg['To'] = to
66    msg['Subject'] = str(Header(subject, 'utf-8'))
67    return {'raw': base64.urlsafe_b64encode(msg.as_bytes()).decode()}
68
69def check_replies(service, state):
70    for email, s in state.get('sent', {}).items():
71        if s.get('replied'): continue
72        res = service.users().messages().list(userId='me', q=f'from:{email} in:anywhere', maxResults=1).execute()
73        if res.get('messages'):
74            state['sent'][email]['replied'] = True
75            log.info(f'Reply from: {email}')
76        time.sleep(0.1)
77
78def main():
79    service = get_service()
80    sent_label_id = get_or_create_label(service, 'Outreach/Sent')
81    state = load_state()
82    check_replies(service, state)
83    save_state(state)
84    today = str(date.today())
85    today_count = state['daily_counts'].get(today, 0)
86    with open(LEADS_CSV) as f:
87        for lead in csv.DictReader(f):
88            if today_count >= DAILY_LIMIT:
89                log.warning('Daily limit reached')
90                break
91            email = lead['email']
92            if email in state['sent']:
93                continue
94            try:
95                subject = TEMPLATE_SUBJECT.format(**lead)
96                body = TEMPLATE_BODY.format(**lead)
97                msg = encode_message(email, subject, body)
98                result = service.users().messages().send(userId='me', body=msg).execute()
99                state['sent'][email] = {'message_id': result['id'], 'thread_id': result['threadId'], 'sent_at': today}
100                service.users().messages().modify(userId='me', id=result['id'],
101                    body={'addLabelIds': [sent_label_id]}).execute()
102                today_count += 1
103                state['daily_counts'][today] = today_count
104                log.info(f'Sent to {email} ({today_count}/{DAILY_LIMIT})')
105                save_state(state)
106                time.sleep(1.5)
107            except HttpError as e:
108                if '429' in str(e) or 'Daily' in str(e): break
109                log.error(f'Failed {email}: {e}')
110
111if __name__ == '__main__':
112    main()