Google Workspace API Integration
Integrates with Google Workspace APIs — Gmail, Google Drive, Google Calendar, Google Docs, Google Sheets, and Admin SDK — using the official google-api-python-client library for automating productivity workflows across the Google ecosystem.
TL;DR for Code Generation
- Use
google.oauth2.service_account.Credentialsfor server-to-server apps orgoogle_auth_oauthlib.flowfor user delegation - Each Google API has its own
discovery.build()service endpoint (e.g.,"gmail", "v1","drive", "v3") - Always specify
fieldsparameter to reduce response size and latency - Use service accounts with domain-wide delegation for admin operations
- Batch Google Sheets updates with
batchUpdate()for atomic changes - Handle
googleapiclient.errors.HttpErrorwith status code differentiation
When to Use
Use this skill when:
- Reading, sending, or filtering Gmail messages programmatically
- Creating, listing, or sharing files and folders in Google Drive
- Inserting, updating, or querying Google Calendar events
- Reading or writing cell ranges in Google Sheets (CRUD)
- Creating or editing Google Docs content
- Managing users, groups, and devices via Admin SDK
- Automating cross-Google-app workflows (e.g., "email me the sheet data")
When NOT to Use
- Accessing Google Workspace data for consumer (free) Gmail accounts with service accounts (use OAuth 2.0 user tokens instead)
- High-volume SMTP sending (use the Gmail SMTP relay or a dedicated email service)
- Replacing Google Apps Script for trivial spreadsheet-bound automation (keep it in Sheets)
- Real-time Drive change tracking (use
drive.changes.watch()with webhooks, not polling)
Core Workflow
Set Up Google Cloud Project — Enable the required APIs (Gmail, Drive, Calendar, Sheets, etc.) in the Google Cloud Console. Checkpoint: Verify each API shows "Enabled" in the Google Cloud Console dashboard.
Create Credentials — For server apps, create a service account and download the JSON key. For user-delegated access, configure an OAuth 2.0 consent screen and download
credentials.json. Checkpoint: Test the credential file with a token generation call.Build the Service Object — Call
googleapiclient.discovery.build(service_name, version, credentials=creds)for each API. Cache the service object. Checkpoint: Execute a lightweight list call (e.g.,drive.files().list(pageSize=1).execute()).Construct the Request — Use the service's fluent method chain (e.g.,
service.users().messages().list(userId="me")). Applyqsearch queries andfieldsfor sparse responses. Checkpoint: Verify the request executes withoutHttpError.Handle Pagination — Check the response for
nextPageToken. Loop withpageTokenset to the previous response's token until it isNone. Checkpoint: Confirm at least one complete page was returned.Parse and Process — Extract fields from the response dict. Handle
HttpErrorwith specific handling for 403 (insufficient permissions), 404 (not found), and 429 (rate limit). Checkpoint: Log warning for 403, retry for 429, raise for others.
Implementation Patterns
Pattern 1: Send Email via Gmail API
import os
import base64
from email.message import EmailMessage
from google.oauth2 import service_account
from googleapiclient.discovery import build
SCOPES = ["https://www.googleapis.com/auth/gmail.send"]
SERVICE_ACCOUNT_FILE = os.environ["GOOGLE_SERVICE_ACCOUNT_PATH"]
DELEGATED_USER = "bot@example.com"
creds = service_account.Credentials.from_service_account_file(
SERVICE_ACCOUNT_FILE, scopes=SCOPES
).with_subject(DELEGATED_USER)
service = build("gmail", "v1", credentials=creds)
msg = EmailMessage()
msg.set_content("The weekly report is ready for review.")
msg["To"] = "manager@example.com"
msg["From"] = DELEGATED_USER
msg["Subject"] = "Weekly Report Ready"
encoded = base64.urlsafe_b64encode(msg.as_bytes()).decode()
service.users().messages().send(userId="me", body={"raw": encoded}).execute()
Pattern 2: Read Google Sheets Data
def read_sheet_range(spreadsheet_id: str, range_name: str) -> list[list[str | float | None]]:
"""Read a range from a Google Sheet and return rows as lists."""
creds = service_account.Credentials.from_service_account_file(
os.environ["GOOGLE_SERVICE_ACCOUNT_PATH"],
scopes=["https://www.googleapis.com/auth/spreadsheets.readonly"],
)
service = build("sheets", "v4", credentials=creds)
result = service.spreadsheets().values().get(
spreadsheetId=spreadsheet_id,
range=range_name,
).execute()
return result.get("values", [])
# Usage
data = read_sheet_range("1BxiMVs0Xw_b9HVMu3HIQ", "Sheet1!A1:E10")
for row in data:
print(f"{row[0]}: {row[4]}")
Pattern 3: Create Calendar Event
from datetime import datetime, timedelta
from googleapiclient.discovery import build
def create_calendar_event(
summary: str,
description: str,
start_time: datetime,
end_time: datetime,
attendees: list[str],
) -> dict:
"""Create a Google Calendar event and return the event resource."""
service = build("calendar", "v3", credentials=creds)
event = {
"summary": summary,
"description": description,
"start": {"dateTime": start_time.isoformat(), "timeZone": "America/New_York"},
"end": {"dateTime": end_time.isoformat(), "timeZone": "America/New_York"},
"attendees": [{"email": a} for a in attendees],
}
return service.events().insert(calendarId="primary", body=event, sendUpdates="all").execute()
Pattern 4: BAD vs GOOD — Error Handling
# ❌ BAD — catches all errors, no fields specified
try:
result = service.files().list(q="mimeType='application/vnd.google-apps.folder'").execute()
except Exception as e:
print("Error:", e)
# ✅ GOOD — typed error handling, fields projection, pagination
from googleapiclient.errors import HttpError
def list_folders(drive_service) -> list[dict]:
"""List all folders the service account has access to."""
folders = []
page_token = None
while True:
try:
response = drive_service.files().list(
q="mimeType='application/vnd.google-apps.folder' and trashed=false",
fields="nextPageToken, files(id, name, createdTime)",
pageToken=page_token,
pageSize=100,
).execute()
folders.extend(response.get("files", []))
page_token = response.get("nextPageToken")
if not page_token:
break
except HttpError as e:
if e.resp.status == 403:
print(f"Permission denied: {e}")
break
if e.resp.status == 429:
print("Rate limited — backing off.")
raise
raise
return folders
Constraints
MUST DO
- Use service account with domain-wide delegation for admin operations across the domain
- Always specify
fieldsparameter to reduce bandwidth and latency - Implement exponential backoff retry for 429 and 5xx errors
- Use
qsearch queries for server-side filtering (Gmail, Drive) - Store service account JSON keys encrypted or in a secrets manager
MUST NOT DO
- Hardcode OAuth client secrets in application code
- Use the same service account for end-user data without
.with_subject()delegation - Poll Drive for changes — use
drive.changes.watch()push notifications - Exceed Google API quota limits (check
quotaUseranduserIpheaders)
Output Template
Every integration function should expose:
- Authentication — Service account or OAuth 2.0 credential initialization
- API Service —
discovery.build()for the target API - Request — Fluent method chain with typed parameters
- Pagination — Loop with
pageToken - Error Handling —
try/except HttpErrorwith status-specific logic
Related Skills
| Skill | Purpose | |