Gmail API skill for Codex

Done well means: mail is never sent, trashed or deleted without the user confirming the exact messages or recipients, replies land in the right thread, bodies decode correctly for every charset and MIME shape, and sync is incremental from a stored historyId.

Auth and scope selection live in $google-workspace-apis. Gmail-specific scope choices:

Task	Scope	Class
Send only	gmail.send	Sensitive
Read headers and labels only	gmail.metadata	Restricted (no q search allowed)
Read everything	gmail.readonly	Restricted
Read + label + trash	gmail.modify	Restricted
Manage labels only	gmail.labels	Non-sensitive
Permanent delete	https://mail.google.com/	Restricted; avoid

Use userId="me" for the authenticated user (or the impersonated user under delegation).

Search and read

import base64
from email import policy
from email.parser import BytesParser

def search(gmail, q, max_ids=None):
    token, seen = None, 0
    while True:
        resp = gmail.users().messages().list(userId="me", q=q, maxResults=500, pageToken=token).execute()
        for m in resp.get("messages", []):
            yield m["id"]
            seen += 1
            if max_ids and seen >= max_ids:
                return
        token = resp.get("nextPageToken")
        if not token:
            return

def fetch_parsed(gmail, msg_id):
    raw = gmail.users().messages().get(userId="me", id=msg_id, format="raw").execute()["raw"]
    return BytesParser(policy=policy.default).parsebytes(base64.urlsafe_b64decode(raw))

for mid in search(gmail, 'from:[email protected] has:attachment newer_than:30d -label:processed'):
    msg = fetch_parsed(gmail, mid)
    body = msg.get_body(preferencelist=("plain", "html"))
    text = body.get_content() if body else ""
    for part in msg.iter_attachments():
        save(part.get_filename(), part.get_payload(decode=True))

format="raw" plus the standard email parser is the least error-prone way to decode bodies: it handles nested multiparts, quoted-printable and charsets. Use format="metadata" with metadataHeaders=["From","Subject","Date"] when you only need headers (cheaper, faster). With format="full", large attachments arrive as attachmentId and need users.messages.attachments.get.

Useful q operators: from:, to:, subject:, label:, -label:, has:attachment, filename:pdf, newer_than:7d, after:2026/10/01, is:unread, in:anywhere, rfc822msgid:.

Send and reply

Sending is outward-facing. Show the user the final To/Cc/Subject/body (or a sample for bulk) and get explicit confirmation before calling send. Prefer creating a draft when the user has not clearly asked to send.

from email.message import EmailMessage

def build_raw(to, subject, text, html=None, reply_to_msg=None, attachments=()):
    m = EmailMessage()
    m["To"] = to
    m["Subject"] = subject
    if reply_to_msg is not None:                       # parsed original message
        m["In-Reply-To"] = reply_to_msg["Message-ID"]
        refs = reply_to_msg.get("References", "")
        m["References"] = f"{refs} {reply_to_msg['Message-ID']}".strip()
    m.set_content(text)
    if html:
        m.add_alternative(html, subtype="html")
    for path, mime in attachments:
        maintype, subtype = mime.split("/")
        m.add_attachment(open(path, "rb").read(), maintype=maintype, subtype=subtype,
                         filename=path.rsplit("/", 1)[-1])
    return base64.urlsafe_b64encode(m.as_bytes()).decode()

body = {"raw": build_raw("[email protected]", "Re: Invoice 4471", "Paid today, thanks.", reply_to_msg=orig),
        "threadId": thread_id}
gmail.users().drafts().create(userId="me", body={"message": body}).execute()   # safe default
# after confirmation:
gmail.users().messages().send(userId="me", body=body).execute()

Threading needs all three: same threadId, In-Reply-To/References headers, and a subject matching the original (a Re: prefix is fine). Omit From unless sending as a verified send-as alias. Daily sending caps exist (roughly 2,000 per day for Workspace users, 500 for consumer accounts); bulk or marketing mail belongs in a dedicated email service.

Labels and bulk changes

labels = {l["name"]: l["id"] for l in gmail.users().labels().list(userId="me").execute()["labels"]}
if "processed" not in labels:
    labels["processed"] = gmail.users().labels().create(
        userId="me", body={"name": "processed", "labelListVisibility": "labelShow",
                           "messageListVisibility": "show"}).execute()["id"]

gmail.users().messages().batchModify(userId="me", body={
    "ids": ids[:1000],                       # max 1000 per call
    "addLabelIds": [labels["processed"]],
    "removeLabelIds": ["UNREAD", "INBOX"],   # archive + mark read
}).execute()

System labels use fixed IDs (INBOX, UNREAD, STARRED, SPAM, TRASH); user labels use generated IDs, so resolve names once. Archiving is removing INBOX. Trash (messages.trash) is recoverable for 30 days; messages.delete and batchDelete are permanent and need explicit confirmation of the exact set. Label filters for the future are users.settings.filters (a standing rule: confirm before creating).

Incremental sync with history

from googleapiclient.errors import HttpError

def sync(gmail, start_history_id):
    token, latest = None, start_history_id
    try:
        while True:
            resp = gmail.users().history().list(
                userId="me", startHistoryId=start_history_id, pageToken=token,
                historyTypes=["messageAdded", "labelAdded"]).execute()
            for h in resp.get("history", []):
                for added in h.get("messagesAdded", []):
                    handle(added["message"]["id"])           # idempotent by message id
            latest = resp.get("historyId", latest)
            token = resp.get("nextPageToken")
            if not token:
                return latest
    except HttpError as e:
        if e.resp.status == 404:                             # historyId too old
            return full_resync()
        raise

Store the returned historyId only after handling succeeds. Get an initial value from users.getProfile or from the latest message you processed.

Push notifications (Pub/Sub)

- [ ] Create a Pub/Sub topic in the same Cloud project as the OAuth client
- [ ] Grant [email protected] the Pub/Sub Publisher role on the topic
- [ ] Create a push subscription to your HTTPS endpoint (with authentication) or a pull subscription
- [ ] Call users.watch with topicName and optional labelIds/labelFilterBehavior
- [ ] Persist the returned historyId; renew watch daily (it expires after 7 days)

gmail.users().watch(userId="me", body={
    "topicName": "projects/my-project/topics/gmail-inbox",
    "labelIds": ["INBOX"], "labelFilterBehavior": "include",
}).execute()

Each Pub/Sub message carries base64 JSON {"emailAddress": ..., "historyId": ...} and nothing else. Acknowledge fast, then call history.list from your stored id. Notifications can be delayed, duplicated or dropped, so also run a periodic history.list as a safety net. Verify the push request’s OIDC token when using push subscriptions.

Quota

Per-user limit is about 250 quota units per second (moving average). Costs: messages.send 100, messages.get 5, messages.list 5, history.list 2, messages.modify 5. A loop of get calls on hundreds of messages needs a small worker pool (4-8) and backoff on 429 and 403 userRateLimitExceeded.

Gotchas

• Bodies are base64url, not standard base64; padding may be missing.
• snippet is HTML-escaped plain text, not the body.
• HTML-only messages are common; fall back to stripping HTML when no text/plain part exists.
• messages.list returns IDs only; budget a get per message.
• Message IDs are per mailbox; use the Message-ID header to match across mailboxes.
• Dates: use the internalDate (ms since epoch) for ordering, not the Date header.

Verification

• Dry-run mode prints matched message IDs, subjects and planned label changes before writing.
• After sending a reply, fetch the thread and confirm the new message has the same threadId.
• After a sync run, messages.list with the same query and compare with handled IDs.