Gmail Read-Only Agent Gateway — v1 Implementation Plan (Python)

Audience: an implementing agent (and its sub-agents). This document is self-contained: it defines the architecture, exact contracts, module boundaries, a non-overlapping task breakdown, and machine-verifiable success criteria for each task so a main agent can verify completion without human judgment.

1. Project summary

A personal, self-hosted service that exposes a Gmail mailbox to AI agents in a strictly read-only fashion. The service syncs new mail from Gmail into a local SQLite store using an OAuth token with only the gmail.readonly scope, and serves it to a single polling agent over a small authenticated HTTP API. No write capability of any kind exists anywhere in the codebase — no send, no delete, no modify, no label changes. Attachments are never stored or served.

Security invariants (must hold in every task)

1. The only Google OAuth scope requested anywhere is https://www.googleapis.com/auth/gmail.readonly.
2. Google credentials (client secret, refresh token) are never readable via any HTTP endpoint and never logged.
3. The HTTP API registers only GET routes. No POST/PUT/PATCH/DELETE handlers exist (they must return 405).
4. Email bodies are stored and served as sanitized plain text only (see §7). Attachments and raw MIME are discarded at ingest.
5. API auth is a single static bearer token compared in constant time (secrets.compare_digest). The comparison lives in one function (app/auth.py:verify_token) so it can later be swapped for multi-key lookup.

2. Stack and constraints

The sync worker runs as an asyncio background task inside the same FastAPI process (started via lifespan). No queue, no Postgres, no Redis, no rate limiting, no multi-key management — explicitly out of scope for v1.

3. Repository layout (create exactly this)

.
├── pyproject.toml
├── README.md
├── Dockerfile
├── .env.example
├── app/
│   ├── __init__.py
│   ├── config.py          # Settings: env parsing, paths, validation
│   ├── db.py              # connection factory, schema DDL, migrations
│   ├── models.py          # dataclasses: Message, HealthStatus, UpdatesPage
│   ├── auth.py            # verify_token(header_value) -> None | raises 401
│   ├── sanitize.py        # html_to_text, strip_dangerous_unicode, normalize
│   ├── cursor.py          # encode_cursor / decode_cursor (opaque, urlsafe)
│   ├── store.py           # insert_message, get_updates(cursor), get_since(ts), sync metadata
│   ├── gmail/
│   │   ├── __init__.py
│   │   ├── client.py      # GmailClient protocol (interface)
│   │   ├── real.py        # RealGmailClient: OAuth + history.list + messages.get
│   │   └── fake.py        # FakeGmailClient: reads JSON fixtures from a directory
│   ├── sync.py            # sync loop: cursor mgmt, fetch, normalize, persist
│   ├── api.py             # FastAPI app factory, routes, lifespan
│   └── main.py            # uvicorn entrypoint
├── scripts/
│   └── authorize.py       # one-time OAuth consent flow, writes token file
├── tests/
│   ├── conftest.py        # tmp DB fixture, app fixture with FakeGmailClient
│   ├── fixtures/emails/   # JSON fixture emails for fake mode
│   ├── test_sanitize.py
│   ├── test_cursor.py
│   ├── test_store.py
│   ├── test_sync.py
│   ├── test_api_auth.py
│   ├── test_api_updates.py
│   └── test_api_readonly.py
└── .github/ (none needed for v1)

4. Configuration contract (app/config.py)

All config from environment variables. .env.example must list every one with a comment.

5. Database schema (app/db.py)

CREATE TABLE IF NOT EXISTS messages (
  id            TEXT PRIMARY KEY,        -- Gmail message id
  thread_id     TEXT NOT NULL,
  sender_name   TEXT NOT NULL DEFAULT '',
  sender_email  TEXT NOT NULL,
  subject       TEXT NOT NULL DEFAULT '',
  body_text     TEXT NOT NULL,           -- sanitized plain text only
  internal_date_ms INTEGER NOT NULL,     -- Gmail internalDate (epoch ms)
  synced_at_ms  INTEGER NOT NULL         -- when THIS service ingested it (epoch ms, monotonic per §6)
);
CREATE INDEX IF NOT EXISTS idx_messages_synced ON messages (synced_at_ms, id);

CREATE TABLE IF NOT EXISTS sync_state (
  key   TEXT PRIMARY KEY,   -- 'history_id', 'last_sync_ok_ms', 'last_sync_error'
  value TEXT NOT NULL
);

Rules:

• Open connections with PRAGMA journal_mode=WAL; and PRAGMA foreign_keys=ON;.
• synced_at_ms must be strictly increasing across inserts within one process (if time.time_ns()//1e6 ties, bump by 1ms) — this makes the cursor total-order well-defined together with id as tiebreaker.
• Inserts are idempotent: INSERT OR IGNORE on primary key (re-sync must not duplicate).

6. Cursor format (app/cursor.py)

Opaque, URL-safe token encoding (synced_at_ms: int, last_id: str):

• encode_cursor(synced_at_ms, last_id) → base64.urlsafe_b64encode(json.dumps({"v":1,"t":synced_at_ms,"i":last_id})) without padding.
• decode_cursor(s) → tuple; raises InvalidCursor on any malformed input (bad base64, bad JSON, wrong version, missing keys, non-int t). API maps InvalidCursor to HTTP 400 with {"error":"invalid_cursor"}.
• Pagination predicate: (synced_at_ms, id) > (t, i) lexicographically, ordered by synced_at_ms ASC, id ASC.

7. Sanitization spec (app/sanitize.py)

Applied at ingest, before storage:

1. Prefer the text/plain MIME part; if absent, convert text/html via BeautifulSoup get_text(separator="\n").
2. Strip zero-width characters: U+200B, U+200C, U+200D, U+2060, U+FEFF.
3. Strip bidirectional override/embedding controls: U+202A–U+202E, U+2066–U+2069.
4. Normalize to NFC. Collapse runs of 3+ newlines to 2. Strip leading/trailing whitespace.
5. Truncate stored body at 100,000 characters (append \n[truncated] when cut).

No attachments: at MIME walk time, skip any part with filename set or Content-Disposition: attachment; never store or return part bytes other than the chosen text body.

8. HTTP API contract (app/api.py)

All routes require Authorization: Bearer <GATEWAY_API_KEY>; failures return 401 {"error":"unauthorized"} (same response for missing and wrong key). Only GET routes exist.

GET /v1/updates?cursor=<opaque>&limit=<1..500, default 100>

• Without cursor: returns no messages, only a next_cursor positioned at the current end (bootstrap call — agents start from "now").
• With cursor: returns messages after the cursor position ordered by (synced_at_ms, id), up to limit.
• Response:

{
  "messages": [{
    "id": "...", "thread_id": "...",
    "sender_name": "...", "sender_email": "...",
    "subject": "...",
    "body_untrusted": "...",
    "received_at": "2026-08-28T11:00:00Z",
    "synced_at": "2026-08-28T11:02:11Z"
  }],
  "next_cursor": "...",
  "has_more": false
}

• next_cursor always present (equals input cursor position advanced past returned rows; unchanged when zero rows). The body field is deliberately named body_untrusted.

GET /v1/messages?since=<ISO8601>&limit=<1..500, default 100>

• Timestamp convenience variant filtering on synced_at_ms > since. Missing/invalid since → 400 {"error":"invalid_since"}. Same message shape, no cursor in response.

GET /v1/health

• Returns 200 {"status":"ok"|"degraded","last_sync_ok":"<ISO8601>|null","last_sync_error":"<string>|null","message_count":<int>,"mode":"real"|"fake"}. degraded when last successful sync is older than 3× the sync interval. Health requires auth too (it leaks mailbox metadata).

9. Gmail client boundary (app/gmail/)

client.py defines the protocol both implementations satisfy:

class GmailClient(Protocol):
    def list_new_message_ids(self, history_id: str | None) -> tuple[list[str], str]:
        """Returns (new_message_ids, new_history_id). history_id None => bootstrap:
        return ([], current_history_id) — v1 starts from 'now', no backfill.
        Raises HistoryExpired if the stored history_id is too old."""
    def fetch_message(self, message_id: str) -> RawMessage:
        """RawMessage: dataclass with headers dict, mime parts (text/plain, text/html), internal_date_ms."""

• real.py: loads token via google.oauth2.credentials.Credentials.from_authorized_user_file, refreshes if expired, builds the service with cache_discovery=False. Uses users.history.list (with startHistoryId, historyTypes=messageAdded) and users.messages.get(format='full'). On HistoryExpired (HTTP 404 from history.list), sync resets: store the profile's current historyId and continue (accepting a gap; log a warning).
• fake.py: reads *.json files from GATEWAY_FAKE_MAILDIR; each file is {"id","thread_id","headers":{"From","Subject","Date"},"text_plain":...,"text_html":...,"internal_date_ms":...}. list_new_message_ids returns files not yet reported this process lifetime, sorted by filename, so dropping a new JSON file simulates mail arrival. This makes the entire system runnable and testable with zero Google setup.
• scripts/authorize.py: standalone InstalledAppFlow.run_local_server consent flow requesting only the readonly scope; writes GATEWAY_GOOGLE_TOKEN. Never imported by the app.

10. Sync loop (app/sync.py)

async def run_sync_loop(client, store, interval): every interval seconds → read history_id from sync_state → list_new_message_ids → for each id not already in DB: fetch_message, extract sender via email.utils.parseaddr(headers['From']), sanitize body per §7, insert → persist new history_id and last_sync_ok_ms. On any exception: record last_sync_error, log, sleep, continue (loop never dies). Also expose async def sync_once(...) used by tests and by lifespan for an immediate first sync.

11. Task breakdown for sub-agents

Tasks are designed to be non-overlapping by file: no two tasks in the same wave touch the same file. Interfaces above are the contract; implement to them exactly.

Wave 1 (parallel, no dependencies)

T1 — Skeleton & config. Files: pyproject.toml, .env.example, README.md, Dockerfile, app/__init__.py, app/config.py, app/models.py, app/main.py (stub that imports app factory lazily). Success criteria: uv sync succeeds; uv run python -c "from app.config import Settings" works; instantiating settings without GATEWAY_API_KEY raises a validation error whose message names the variable; ruff check . clean.

T2 — Storage core. Files: app/db.py, app/cursor.py, app/store.py, tests/test_cursor.py, tests/test_store.py. Success criteria: uv run pytest tests/test_cursor.py tests/test_store.py green. Tests must cover: cursor round-trip; decode_cursor raising InvalidCursor on ≥4 malformed inputs; idempotent insert (same id twice → 1 row); monotonic synced_at_ms under two same-millisecond inserts; get_updates pagination returning correct pages and has_more across a 250-row dataset with limit 100.

T3 — Sanitization. Files: app/sanitize.py, tests/test_sanitize.py. Success criteria: uv run pytest tests/test_sanitize.py green. Tests must cover: HTML→text of a nested HTML email; removal of every codepoint listed in §7 items 2–3 (assert none remain in output); NFC normalization; newline collapsing; 100k truncation with marker; plain-text passthrough unchanged apart from whitespace rules.

T4 — Gmail clients. Files: app/gmail/__init__.py, app/gmail/client.py, app/gmail/real.py, app/gmail/fake.py, scripts/authorize.py, tests/fixtures/emails/ (≥5 varied fixtures: plain-text, HTML-only, both-parts, attachment-bearing, unicode-tricks). Success criteria: uv run python -c "import app.gmail.real" succeeds (imports must not require credentials); FakeGmailClient over the fixtures returns all ids once and then ([], hid) on the next call; grep gate: grep -r "gmail" app/ scripts/ | grep -i scope shows only gmail.readonly; no other Gmail scope string appears anywhere in the repo.

Wave 2 (parallel, depends on Wave 1)

T5 — Sync engine. Files: app/sync.py, tests/test_sync.py, tests/conftest.py (owns conftest). Success criteria: uv run pytest tests/test_sync.py green. Tests must cover: sync_once with FakeGmailClient ingests all fixtures with sanitized bodies (assert the unicode-tricks fixture is cleaned and the attachment fixture stored without attachment data); second sync_once inserts nothing; a client whose fetch_message raises records last_sync_error without raising out of sync_once's loop wrapper run_sync_loop iteration; history_id persisted.

T6 — API layer. Files: app/auth.py, app/api.py, tests/test_api_auth.py, tests/test_api_updates.py, tests/test_api_readonly.py; finalize app/main.py. Success criteria: uv run pytest tests/test_api_auth.py tests/test_api_updates.py tests/test_api_readonly.py green. Tests must cover: 401 on missing/wrong bearer for all three routes (identical body); verify_token uses secrets.compare_digest (assert via source inspection or monkeypatch); bootstrap /v1/updates without cursor returns 0 messages + usable next_cursor; full poll cycle: bootstrap → insert 3 rows → poll returns exactly 3 with correct field names including body_untrusted → next poll returns 0; invalid cursor → 400; /v1/messages?since= valid & invalid; /v1/health shape and degraded transition; read-only gate: parametrized test asserting POST/PUT/PATCH/DELETE on every route path returns 405, and a route-table introspection test asserting every registered route allows only GET (and HEAD).

Wave 3 (single agent, integration)

T7 — Integration & docs polish. May touch any file. Run the full suite, run the service end-to-end in fake mode: start server, bootstrap cursor via curl, drop a new JSON fixture into the maildir, wait one sync interval, poll and confirm the new message appears; verify Docker build; finalize README (setup incl. Google Cloud OAuth steps, fake-mode quickstart, API reference, security notes). Success criteria: uv run pytest fully green; ruff check . clean; the curl end-to-end transcript shows the dropped fixture returned by /v1/updates; docker build . succeeds; README contains a copy-pasteable fake-mode quickstart that works.

Main-agent verification checklist (after all waves)

1. uv run pytest -q → exit 0, zero skipped security tests.
2. ruff check . → exit 0.
3. Grep gates: only gmail.readonly scope string in repo; no users().messages().send, modify, trash, or delete calls anywhere; body_untrusted present in API responses.
4. Live fake-mode smoke test per T7 transcript.
5. Confirm .env.example documents every variable in §4.

12. Explicitly out of scope for v1

Attachments, HTML bodies in responses, multiple API keys, rate limiting, MCP adapter, Gmail push (Pub/Sub), backfill of pre-existing mail, label filtering, TLS termination (run behind a reverse proxy or on localhost/tailnet).

13. Future-proofing notes for the implementer

Keep store.get_updates and sync_once free of FastAPI imports — they are the durable core. The HTTP layer and any future MCP adapter are skins over those functions. app/auth.py:verify_token is the single seam for a future multi-key upgrade.