# Gmail > Keep the email conversations each user takes part in, in their memory, with their own Google sign-in. Each thread is one source; a new reply teaches only what it adds, a thread that is gone takes what it taught with it, and disconnecting forgets everything from Gmail. Keep the email conversations each of your users takes part in, in their memory. Your app already holds an access token for each user, from their Google sign-in with the `gmail.readonly` scope. This sync adds each thread the user wrote in as one source, under its own id with the label `channel: gmail`: who said what, and when, in each message's own words, without the quoted copies of earlier messages. Newsletters and mail no one answered stay out. The first run reads the past year; after that, Gmail's history says which threads changed, so a new reply teaches only what it adds, a thread that is gone takes what it taught with it, and when the user disconnects Gmail, one call forgets everything that came from it. ## Install ```bash pip install httpx geniffy ``` ```bash uv add httpx geniffy ``` Set `GENIFFY_API_KEY` from **API keys** in the Geniffy app. ## The sync ```python import base64 import html import re from datetime import datetime, timezone import httpx from geniffy import Geniffy, NotFoundError geniffy = Geniffy() # reads GENIFFY_API_KEY LABELS = {"channel": "gmail"} QUERY = "in:sent newer_than:1y" # the first run: every thread the user wrote in, this past year QUOTED = re.compile(r"^(On .+ wrote:|-+ ?Original Message ?-+|_{20,})$") # where a reply starts quoting def gmail(token: str) -> httpx.Client: """One user's mailbox, with the access token from their Google sign-in.""" return httpx.Client(base_url="https://gmail.googleapis.com/gmail/v1/users/me", timeout=60, headers={"Authorization": f"Bearer {token}"}) def plain(part: dict) -> str: """A message's plain text, from whichever of its parts holds it.""" data = part.get("body", {}).get("data") if part.get("mimeType") == "text/plain" and data and not part.get("filename"): return base64.urlsafe_b64decode(data + "=" * (-len(data) % 4)).decode("utf-8", "replace") for sub in part.get("parts", []): if text := plain(sub): return text return "" def headers(message: dict) -> dict: return {h["name"].lower(): h["value"] for h in message["payload"].get("headers", [])} def own_words(text: str) -> str: """What a message adds, without the earlier messages it quotes.""" kept = [] for line in text.splitlines(): if QUOTED.match(line.strip()): break if not line.startswith(">"): kept.append(line) return "\n".join(kept).strip() def conversation(thread: dict) -> tuple[str, str, datetime] | None: """A thread as one note: who wrote each message, when, and its own words; with its subject and the time of its latest message. None when the user never wrote in it.""" messages = [m for m in thread.get("messages", []) if not {"TRASH", "SPAM", "DRAFT"} & set(m.get("labelIds", []))] if not any("SENT" in m.get("labelIds", []) for m in messages): return None parts, when = [], None for m in messages: words = own_words(plain(m["payload"]) or html.unescape(m.get("snippet", ""))) when = datetime.fromtimestamp(int(m["internalDate"]) / 1000, timezone.utc) if words: parts.append(f"From {headers(m).get('from', 'someone')}, {when:%d %b %Y}:\n{words}") if not parts: return None return headers(messages[0]).get("subject") or "(no subject)", "\n\n".join(parts), when def keep(mem, api: httpx.Client, thread_id: str) -> bool: """Add one thread under its Gmail id. False when it is gone, or the user never wrote in it.""" got = api.get(f"/threads/{thread_id}", params={"format": "full"}) found = None if got.status_code == 404 else conversation(got.raise_for_status().json()) if found is None: return False subject, text, when = found mem.memories.add(text, title=subject, said_at=when, external_id=f"gmail:{thread_id}", labels=LABELS) return True def forget(mem, thread_id: str) -> None: try: mem.sources.delete(external_id=f"gmail:{thread_id}") except NotFoundError: # never added: no one answered it pass def read_all(mem, api: httpx.Client) -> None: """Every thread QUERY finds; and what memory holds from Gmail that is no longer there goes.""" seen, params = set(), {"q": QUERY, "maxResults": 100} while True: out = api.get("/threads", params=params).raise_for_status().json() seen |= {f"gmail:{t['id']}" for t in out.get("threads", []) if keep(mem, api, t["id"])} if "nextPageToken" not in out: break params["pageToken"] = out["nextPageToken"] mem.sources.delete_labelled(LABELS, keep=seen) def changed(api: httpx.Client, history_id: str) -> tuple[set, str] | None: """The threads that changed since history_id, and the id to pass next time. None when Gmail no longer keeps its history that far back.""" threads, params = set(), {"startHistoryId": history_id, "maxResults": 500} while True: got = api.get("/history", params=params) if got.status_code == 404: return None out = got.raise_for_status().json() for record in out.get("history", []): for item in record.get("messagesAdded", []) + record.get("messagesDeleted", []): threads.add(item["message"]["threadId"]) for item in record.get("labelsAdded", []) + record.get("labelsRemoved", []): if {"TRASH", "SPAM"} & set(item.get("labelIds", [])): # binned or marked as spam, or back threads.add(item["message"]["threadId"]) if "nextPageToken" not in out: return threads, out["historyId"] params["pageToken"] = out["nextPageToken"] def sync(user_id: str, token: str, history_id: str | None = None) -> str: """Bring one user's email conversations into their memory. Returns the id to pass next time: pass None the first time and QUERY is read; after that, only threads that changed since the last run are fetched.""" mem = geniffy.space(f"user_{user_id}") with gmail(token) as api: found = changed(api, history_id) if history_id else None if found is None: # the first run, or history too old to ask for start = api.get("/profile").raise_for_status().json()["historyId"] read_all(mem, api) return start threads, latest = found for thread_id in threads: if not keep(mem, api, thread_id): forget(mem, thread_id) return latest def disconnect(user_id: str) -> int: """The user disconnected Gmail: everything that came from it goes. Returns how many threads.""" return geniffy.space(f"user_{user_id}").sources.delete_labelled(LABELS) ``` Run it on a schedule, and keep the id each run returns with the user for the next run. Google's access tokens last an hour, so refresh the user's before each run, as your Google sign-in library does. `gmail.readonly` is one of Google's restricted scopes: before your app reads the mail of people outside your own Google Workspace, Google's verification asks for a security assessment. Plan for it before you launch. ## How it behaves - **Each thread is one source,** under its Gmail id and dated by its latest message. A reply sends the thread again, and only what the reply adds is learned. See [Your own ids](https://docs.geniffy.com/add-memories/your-own-ids). - **Only conversations the user wrote in,** so newsletters, receipts and mail no one answered stay out. The first run reads the past year; change `QUERY` to read further back. After that, every thread the user writes in comes in. - **Each message in its own words.** The quoted copy of the message before it is left out, so nothing is learned twice. A message with no plain text comes in as Gmail's short preview of it, and attachments are left out. - **Only what changed is fetched.** After the first run, Gmail's history names the threads that changed. Gmail keeps that history for a week or more; when a run asks for more than it keeps, it reads everything again and puts memory right. - **A thread that is gone goes from memory too,** with what it taught: deleted, in the bin, or marked as spam. - **Recall can keep to email,** or leave it out: `mem.context(question, labels={"channel": "gmail"})`. See [Labels](https://docs.geniffy.com/add-memories/labels). - **Disconnecting forgets it all** in one call, and nothing the user added another way. Source: https://docs.geniffy.com/integrations/gmail