Build an email agent that reads your inbox and does the work

A scheduled script, one AI session to classify the mail, and another to actually handle it.

By George Leon on

Loading the Elevenlabs Text to Speech AudioNative Player...

My inbox runs itself now.

Most of the work went into the plumbing rather than the AI part. A dumb scheduled script decides when it’s worth spending a token, and a pile of guardrails stops a bad day turning into thirty runaway sessions.

This post is the general recipe. Swap in your own routing rules and it works for whatever you do. Hand this whole article to your coding agent and have it build the thing for you.

What You Are Building

Layer 1: the poller. A Python script on a one-minute schedule, with no AI in it and no API keys beyond Gmail. It asks the inbox for messages it hasn’t seen yet, marks them, and works out whether each one is even worth showing to a model. On a quiet minute that’s a single API call before it exits, which is why you can afford to run it constantly.

Layer 2: triage. For each message that survives the cheap filters, the poller launches a headless AI session with a single job: read this email, match it against a routing table, and output a JSON verdict. The verdict says handle, notify, or skip, and if it says handle it also says which project folder to work in and what the instructions are.

Layer 3: the handler. A second AI session, launched in the folder the verdict named, with a prompt the verdict wrote. This one has tools and permissions, so it does the real work of editing files, running scripts, drafting a reply and pinging you.

The split between triage and handling is there because triage is cheap, stateless and runs on every single message, where a handler is expensive and only fires on the small fraction that turn out to be real work. You get two other things out of it. The routing logic ends up in plain English in one file you can edit without touching code, and the handler starts with a clean context holding nothing but its own project’s instructions.

What You Need

Step 1: Gmail API Access

Skip IMAP and app passwords. The REST API is what you want, since it hands you labels, threads and attachment bytes over plain HTTPS.

In the Google Cloud console: create a project, enable the Gmail API, then create an OAuth client of type Web application with redirect URI http://127.0.0.1:8765.

If your account is on Google Workspace, set the consent screen to Internal. An External app left in Testing mode has its refresh tokens revoked after seven days, so your agent goes quiet next Tuesday for no visible reason.

Then mint a refresh token once, with a throwaway consent script:

# consent.py — run once, paste the printed token into your env
import http.server, json, urllib.parse, urllib.request, webbrowser

CLIENT_ID = "...apps.googleusercontent.com"
CLIENT_SECRET = "..."
REDIRECT = "http://127.0.0.1:8765"
SCOPE = "https://www.googleapis.com/auth/gmail.modify"

url = "https://accounts.google.com/o/oauth2/v2/auth?" + urllib.parse.urlencode({
    "client_id": CLIENT_ID, "redirect_uri": REDIRECT, "response_type": "code",
    "scope": SCOPE, "access_type": "offline", "prompt": "consent",
})
webbrowser.open(url)

class Handler(http.server.BaseHTTPRequestHandler):
    def do_GET(self):
        code = urllib.parse.parse_qs(urllib.parse.urlparse(self.path).query)["code"][0]
        body = urllib.parse.urlencode({
            "code": code, "client_id": CLIENT_ID, "client_secret": CLIENT_SECRET,
            "redirect_uri": REDIRECT, "grant_type": "authorization_code",
        }).encode()
        tok = json.load(urllib.request.urlopen("https://oauth2.googleapis.com/token", body))
        print("\nGMAIL_REFRESH_TOKEN=" + tok["refresh_token"])
        self.send_response(200)
        self.end_headers()
        self.wfile.write(b"Done, close this tab.")

http.server.HTTPServer(("127.0.0.1", 8765), Handler).handle_request()

The consent screen lets you untick individual permissions. Leave every box checked, or you’ll end up with a read-only token and every label call comes back 403.

Store GMAIL_CLIENT_ID, GMAIL_CLIENT_SECRET, and GMAIL_REFRESH_TOKEN wherever you keep secrets.

Step 2: The Gmail Helper

The gmail.modify scope covers everything the agent needs: read messages, add and remove labels, create drafts, trash mail. This helper is stdlib only, no Google packages to install:

"""gmail.py — minimal Gmail REST helper (list, read, label, trash)."""
import base64, json, os, time, urllib.parse, urllib.request

API = "https://gmail.googleapis.com/gmail/v1/users/me"
_tok = {"value": "", "expires": 0.0}


def _token() -> str:
    if _tok["value"] and time.time() < _tok["expires"]:
        return _tok["value"]
    body = urllib.parse.urlencode({
        "client_id": os.environ["GMAIL_CLIENT_ID"],
        "client_secret": os.environ["GMAIL_CLIENT_SECRET"],
        "refresh_token": os.environ["GMAIL_REFRESH_TOKEN"],
        "grant_type": "refresh_token",
    }).encode()
    data = json.load(urllib.request.urlopen("https://oauth2.googleapis.com/token", body))
    _tok.update(value=data["access_token"], expires=time.time() + data["expires_in"] - 60)
    return _tok["value"]


def _api(path: str, method: str = "GET", payload: dict | None = None) -> dict:
    req = urllib.request.Request(
        f"{API}{path}", method=method,
        data=json.dumps(payload).encode() if payload is not None else None,
        headers={"Authorization": f"Bearer {_token()}",
                 "Content-Type": "application/json"})
    with urllib.request.urlopen(req) as r:
        return json.loads(r.read() or b"{}")


def list_recent(query: str, limit: int = 50) -> list[dict]:
    """Message metadata, newest first. `query` is Gmail search syntax."""
    q = urllib.parse.urlencode({"q": query, "maxResults": limit})
    out = []
    for m in _api(f"/messages?{q}").get("messages", []):
        full = _api(f"/messages/{m['id']}?format=metadata")
        h = {x["name"].lower(): x["value"] for x in full["payload"]["headers"]}
        out.append({"id": m["id"], "thread_id": full["threadId"],
                    "from": h.get("from", ""), "subject": h.get("subject", ""),
                    "date": h.get("date", "")})
    return out


def _walk(part: dict, out: dict) -> None:
    if part.get("filename"):
        out["attachments"].append(part["filename"])
    elif part.get("mimeType") == "text/plain":
        data = part.get("body", {}).get("data")
        if data:
            out["body_text"] += base64.urlsafe_b64decode(data + "==").decode("utf-8", "replace")
    for p in part.get("parts", []):
        _walk(p, out)


def get_message(msg_id: str) -> dict:
    """Full message: headers, plain-text body, attachment filenames."""
    full = _api(f"/messages/{msg_id}")
    h = {x["name"].lower(): x["value"] for x in full["payload"]["headers"]}
    out = {"id": msg_id, "thread_id": full["threadId"], "body_text": "",
           "attachments": [], **{k: h.get(k, "") for k in
                                 ("from", "to", "cc", "date", "subject")}}
    _walk(full["payload"], out)
    return out


def ensure_label(name: str) -> str:
    for label in _api("/labels").get("labels", []):
        if label["name"] == name:
            return label["id"]
    return _api("/labels", "POST", {"name": name})["id"]


def modify_labels(msg_id: str, add: list[str] = (), remove: list[str] = ()) -> None:
    _api(f"/messages/{msg_id}/modify", "POST",
         {"addLabelIds": list(add), "removeLabelIds": list(remove)})

A bare Gmail query searches All Mail. If you mean the inbox, say in:inbox explicitly, or you’ll pull in your own sent replies and archived threads from four years ago.

Step 3: The Poller

The poller is what decides when you spend money. Its state lives in a Gmail label: every message it has looked at gets agent-seen, and its query asks for inbox messages lacking that label. There’s no database and no cursor file to corrupt, and when it goes wrong you can see the state sitting in Gmail and fix it by hand.

"""poller.py — the dumb layer. Runs every minute from a scheduler."""
import json, os, re, subprocess, time
from datetime import datetime, timezone
from email.utils import parsedate_to_datetime
from pathlib import Path

from gmail import ensure_label, get_message, list_recent, modify_labels

AGENT_DIR = Path(__file__).resolve().parent
LABEL, HANDLED_LABEL = "agent-seen", "agent-handled"
LOCK, LOG = AGENT_DIR / "poller.lock", AGENT_DIR / "log.txt"
LOCK_STALE_SECS = 90 * 60     # evict a wedged run older than this
SCAN_LIMIT = 60               # how deep to look each poll
MAX_PER_RUN = 5               # threads per run; the next poll gets the rest
MAX_AGE_DAYS = 14             # ignore resurfaced old threads
HANDLER_TIMEOUT = 1800
BODY_EXCERPT_CHARS = 4000
# Senders that will never be worth an LLM call. Add to this list often.
SKIP_SENDERS = {"noreply@example.com", "notifications@stripe.com"}
OWN_ADDRESSES = {"you@example.com"}


def log(line: str) -> None:
    with LOG.open("a", encoding="utf-8") as f:
        f.write(f"{datetime.now():%Y-%m-%d %H:%M:%S}  {line}\n")


def run_claude(prompt: str, cwd: Path, timeout: int, label: str) -> str | None:
    """Run a headless session, prompt over stdin so there is no shell quoting."""
    cmd = ["claude", "-p", "--output-format", "json", "--dangerously-skip-permissions"]
    try:
        proc = subprocess.run(cmd, input=prompt, capture_output=True, text=True,
                              encoding="utf-8", cwd=str(cwd), timeout=timeout)
    except subprocess.TimeoutExpired:
        log(f"{label}: timed out after {timeout}s")
        return None
    if proc.returncode != 0:
        log(f"{label}: exited {proc.returncode}: {proc.stdout[-400:]!r} {proc.stderr[-400:]!r}")
        return None
    try:
        return json.loads(proc.stdout)["result"]
    except Exception:
        log(f"{label}: unparseable output: {proc.stdout[-400:]!r}")
        return None


def sender(from_header: str) -> str:
    m = re.search(r"[\w.+-]+@[\w.-]+", from_header or "")
    return m.group(0).lower() if m else ""


def age_days(date_header: str) -> float | None:
    try:
        dt = parsedate_to_datetime(date_header)
    except Exception:
        return None
    if dt is None:
        return None
    if dt.tzinfo is None:
        dt = dt.replace(tzinfo=timezone.utc)
    return (datetime.now(timezone.utc) - dt).total_seconds() / 86400


def group_by_thread(msgs: list[dict]) -> list[tuple[dict, list[dict]]]:
    """Collapse a newest-first list into [(newest of thread, older siblings)]."""
    groups: dict[str, list[dict]] = {}
    order: list[str] = []
    for m in msgs:
        tid = m.get("thread_id") or m["id"]
        if tid not in groups:
            groups[tid] = []
            order.append(tid)
        groups[tid].append(m)
    return [(groups[t][0], groups[t][1:]) for t in order]


def triage_prompt(full: dict) -> str:
    return (
        "You are the email triage session described in this folder's CLAUDE.md. "
        "Classify the email below per the routing table and output ONLY the JSON verdict.\n\n"
        f"Gmail message id: {full['id']}\nThread id: {full['thread_id']}\n"
        f"From: {full['from']}\nTo: {full['to']}\nCc: {full['cc']}\n"
        f"Date: {full['date']}\nSubject: {full['subject']}\n"
        f"Attachments: {full['attachments'] or 'none'}\n\n"
        f"Body (first {BODY_EXCERPT_CHARS} chars):\n{full['body_text'][:BODY_EXCERPT_CHARS]}\n")


def parse_verdict(text: str) -> dict | None:
    m = re.search(r"\{.*\}", text, re.S)
    if not m:
        return None
    try:
        v = json.loads(m.group(0))
    except json.JSONDecodeError:
        return None
    return v if v.get("action") in ("handle", "notify", "skip") else None


def process(msg: dict, label_id: str, handled_id: str) -> None:
    desc = f"{msg['from']!r} / {msg['subject']!r}"
    modify_labels(msg["id"], add=[label_id])       # label first: at-most-once
    if sender(msg["from"]) in SKIP_SENDERS | OWN_ADDRESSES:
        log(f"pre-skip (denylisted): {desc}")
        return
    full = get_message(msg["id"])
    log(f"triage start: {desc}")
    result = run_claude(triage_prompt(full), AGENT_DIR, 600, "triage")
    verdict = parse_verdict(result) if result else None
    if not verdict:
        modify_labels(msg["id"], remove=[label_id])  # hand it back to the queue
        log(f"triage FAILED, un-labeled for retry: {desc}")
        return
    log(f"verdict: {json.dumps({k: v for k, v in verdict.items() if k != 'handler_prompt'})}")

    action = verdict["action"]
    if action in ("handle", "notify"):
        modify_labels(msg["id"], add=[handled_id])
    if action == "handle":
        repo = Path(verdict["repo"])
        if not repo.is_dir():
            notify(f"triage routed to a missing folder: {repo}")
            return
        log(f"handler start in {repo}")
        out = run_claude(verdict["handler_prompt"], repo, HANDLER_TIMEOUT, "handler")
        log(f"handler done: {(out or 'FAILED')[:300]}")
        if out is None:
            notify(f"handler FAILED for {msg['subject']!r}")
    elif action == "notify":
        notify(verdict.get("summary") or desc)
    else:
        log(f"skipped ({verdict.get('category', '?')})")


def main() -> None:
    label_id = ensure_label(LABEL)
    msgs = list_recent(f"in:inbox -label:{LABEL}", limit=SCAN_LIMIT)
    if not msgs:
        return                                  # quiet poll: one API call, no log spam
    if LOCK.exists():
        if time.time() - LOCK.stat().st_mtime < LOCK_STALE_SECS:
            return                              # previous run still working
        log("evicting stale lock")
    handled_id = ensure_label(HANDLED_LABEL)
    LOCK.write_text(str(datetime.now()))
    try:
        for newest, siblings in reversed(group_by_thread(msgs)[:MAX_PER_RUN]):
            LOCK.write_text(str(datetime.now()))   # re-stamp per thread
            for s in siblings:                     # older messages are history
                modify_labels(s["id"], add=[label_id])
            age = age_days(newest.get("date", ""))
            if age is not None and age > MAX_AGE_DAYS:
                modify_labels(newest["id"], add=[label_id])
                log(f"pre-skip (stale, {age:.0f}d old)")
                continue
            process(newest, label_id, handled_id)
    finally:
        LOCK.unlink(missing_ok=True)


if __name__ == "__main__":
    main()

Add your own notify(). Mine is six lines of urllib against the Telegram bot API:

def notify(text: str) -> None:
    tok, chat = os.environ["TELEGRAM_BOT_TOKEN"], os.environ["TELEGRAM_CHAT_ID"]
    body = urllib.parse.urlencode({"chat_id": chat, "text": text}).encode()
    try:
        urllib.request.urlopen(f"https://api.telegram.org/bot{tok}/sendMessage", body, timeout=30)
    except Exception as e:
        log(f"notify failed: {e}")

Baseline the inbox before the first real run. Label everything currently sitting there as agent-seen without processing it, or your first poll cheerfully starts triaging two thousand old emails:

for m in list_recent(f"in:inbox -label:{LABEL}", limit=500):
    modify_labels(m["id"], add=[label_id])

Step 4: The Routing Table

The triage session runs in the agent folder, so that folder’s CLAUDE.md is its entire brief. Write it as instructions to a person doing one job.

Make skip the default for everything you haven’t explicitly routed. Set it to notify instead and you’ve rebuilt email notifications with a token bill attached, which helps nobody, because you can already see your inbox. What you’re paying for is a thing that goes and does the job.

# Email agent — triage

A poller watches the inbox for messages without the `agent-seen` label and launches a headless session here for each one. You are that session.

You receive one email (metadata plus a body excerpt) in your prompt. Decide what to do with it using the routing table below, then output ONLY a JSON object, no prose and no markdown fences, with this shape:

```
{
  "category": "<short slug>",
  "action": "handle" | "notify" | "skip",
  "repo": "<absolute path for the handler session's cwd; null unless handle>",
  "handler_prompt": "<full prompt for the handler session; null unless handle>",
  "summary": "<one plain line: who it's from and what it wants>"
}
```

`handle` — the poller runs `claude -p "<handler_prompt>"` with cwd `repo`. The handler starts with zero context beyond that folder's own CLAUDE.md, so put everything it needs in the prompt: message id, sender, subject, what to do.

`skip` — a log line, nothing else. THIS IS THE DEFAULT for everything that is not one of the routed rows below, no matter how interesting it looks.

`notify` — the poller sends `summary` to my phone. Reserved for the single ambiguous row below. Never use it to forward news, leads, invites, or FYIs.

## Routing table

Check these in order. First match wins.

**1. agent-test** — subject contains `[agent-test]`.
Action: `handle`, repo `<agent folder>`, handler_prompt "Send the notification `POC handler ran` and stop."

**2. work-request** — <your main work request: describe exactly who it comes from and what it looks like>.
Action: `handle`, repo `<project folder>`, handler_prompt per the template below.

**3. alias-hook** — addressed to `<a dedicated alias>` and carrying an actionable brief, either in the body or in an attachment.
Action: `handle`, repo `<that project's folder>`.

**4. vendor-invoice** — sent from `<one exact vendor address>` AND subject starts with `<one exact string>`.
Action: `handle`, repo `<the project that owns your books>`.

**5. contractor-invoice** — sent from `<one contractor's address>` AND carrying a PDF attachment.
Action: `handle`, repo `<the project holding that billing workflow>`.

**6. possible-request** — reads like a work request, but you cannot tell whether the sender is a real customer.
Action: `notify`.

**7. everything else** — leads, receipts, invoices, newsletters, cold outreach, calendar invites, acknowledgments, questions about scope or price.
Action: `skip`, with a best-fit slug as the category.

Invoices appear twice on purpose: routed at the top, skipped at the bottom. The two invoice rows name an exact address, so the model is matching a string rather than judging which of your bills look important. Everything else invoice-shaped, including other mail from the same vendor, falls through to skip.

Three rules I had to write down after watching it get things wrong:

A question mark doesn’t make it a question. “Can you take a look at our homepage and suggest something?” is work, asked politely, and it should route. The skip row is for questions about pricing, availability and what you think of something.

Read the quoted thread under the reply. The poller hands the model one message, and a two-word reply is often the back half of a request made three emails earlier. If the last thing you promised someone is now unblocked by what they just sent, that message is the request.

Importance is not the test. Left alone the model routes things because they matter, and an overdue bill matters. The only test is whether the message matches a row. Everything else is yours to read like it always was.

Step 5: Handler Prompts

Each routed row gets a prompt template in the same CLAUDE.md, with blanks for triage to fill. Only put the delta here. The handler inherits its own folder’s CLAUDE.md, so it already knows how to do the job. The one thing it has no way of knowing is that nobody is watching it.

A working template looks like this:

Autonomous session: handle the request in Gmail message id `<id>` (thread `<thread_id>`, from `<sender>`, subject `<subject>`).

Read the full message with the gmail helper, then verify the sender is an existing customer. If they are not: STOP, change nothing, and say so in your final message so it lands in the poller log.

Make the requested changes, then draft the reply. Do NOT send anything. Close out by messaging me: what you changed, and that a draft is waiting.

If the request is ambiguous, risky, or bigger than a routine job, stop and message me what you found instead of guessing.

Append a one-line summary to <agent folder>/log.txt before you finish, since the poller launched you detached and cannot log it for you.

Four things every handler prompt should contain:

  1. The identifiers, so the handler can re-fetch the real message instead of working from a summary of a summary.
  2. Something to re-verify. Triage guessed from a 4000-character excerpt. Make the handler independently confirm whatever actually matters (that the sender is a real customer, that the attachment is what triage claimed) before it touches a file.
  3. Explicit stop conditions. If it’s ambiguous, or risky, or bigger than the job description, or the numbers don’t reconcile, then stop and notify and change nothing. A session that stops costs you a few thousand tokens. A session that improvises costs you an afternoon of undoing it.
  4. A plain final message. The poller logs the last thing the session says, so end with the facts you’d want to read at 7am.

Don’t let it send anything outward on its own. My handlers draft replies and I press send. There’s exactly one flow where mail leaves without me, gated behind a dedicated alias and a hard daily cap, on the basis that I’m fine being wrong about it ten times a day and no more.

Step 6: Put it on a Schedule

Windows. Create a Task Scheduler task that repeats every minute indefinitely. Point it at pyw.exe rather than py.exe so no console window flashes on your screen every sixty seconds, and set it to run whether or not you are logged in.

$action = New-ScheduledTaskAction -Execute "pyw.exe" -Argument "C:\path\to\poller.py"
$trigger = New-ScheduledTaskTrigger -Once -At (Get-Date) `
  -RepetitionInterval (New-TimeSpan -Minutes 1)
Register-ScheduledTask -TaskName "email-agent-poller" -Action $action -Trigger $trigger

Mac or Linux. A cron entry does the same job:

* * * * * cd /path/to/agent && /usr/bin/python3 poller.py >> cron.log 2>&1

Every minute sounds aggressive until you look at what a quiet poll does, which is one Gmail call and an exit in under a second.

The Guards That Matter

Every one of these came out of something going wrong.

Label before you triage, and un-label on failure. Marking the message first is what makes handling at-most-once, so a crash mid-handler can’t re-run the work. The cost of that ordering is that a session dying before it produces a verdict drops the email forever, which ate a real client request of mine. So a failed triage removes the label again and lets the next poll retry, capped at three attempts per message. On the third it keeps the label and messages you, since that’s the one situation where the script knows it dropped mail it never managed to read.

One thread, one triage. Group the batch by thread id and triage only the newest message, labeling the older ones as seen. Skip this and a thread that resurfaces in your inbox arrives as thirty separate live requests. Mine opened three handler windows off one old chain before I fixed it.

An age gate. Anything older than about two weeks gets labeled without triaging. It’s the backstop for archived threads that find their way back into the inbox.

A lock file, re-stamped per message. One run at a time. Give the lock a staleness timeout so a wedged run gets evicted, then re-stamp it as each message starts, or a busy batch blows past the timeout and the next poll starts a second run on top of the live one.

A denylist you maintain. Read your log weekly. Any sender that’s been triaged five times and skipped five times is a sender you’re paying a model to ignore. Support the domain form too, for the ones that rotate the local part.

Caps on anything that reaches the outside world. A daily limit on outbound sends, a limit on concurrent handler sessions. Set them at the level where a bad day costs you something you’d notice, and have the script notify you rather than act once a cap binds.

One log file. Detached handlers append their own closeouts to it. Rotate it on size. It’s your only record of what the agent did, so it’s the first place to look when something is off.

When the Job Touches Money

A bad edit to a homepage takes a minute to revert. A bad invoice is wrong numbers in your books that you find six weeks later. So the billing flows get four extra rules.

Make it idempotent. Dup-check every write against the invoice number, so a re-run books nothing twice. You will re-run one by hand at some point.

Tie out the numbers, and stop dead when they don’t. Items plus tax plus fees has to equal the invoice total. When it doesn’t, the handler writes nothing and texts me the discrepancy. No rounding, no best guess.

Stop at the draft. The re-invoicing handler builds the Stripe invoice and leaves it unsent. Same rule as client replies, same reason.

Never let it move money. Paying my writer is a step the agent cannot run. Look for that shape in your own flows and write the exclusion into the handler prompt.

One more if you want the source email out of your inbox afterwards: verify against the system of record, not the session. A handler saying it’s done isn’t evidence. Mine only trashes the email once a separate read-only script confirms the database rows and the filed PDF exist. Trash, not delete, so a wrong call is recoverable.

Headless Versus Interactive Handlers

This one cost me a real reply. A headless claude -p session exits the instant it prints its final message, killing any backgrounded listener along with it. So a handler that finishes by messaging you and waiting for a “send it” or “log 20 minutes” is waiting in a room you can’t get into.

There are two ways out. The handler doesn’t wait at all, and you approve the drafted work yourself later on. Or you launch that category in a visible terminal window rather than headless, where the session sits at its prompt after finishing, the listener survives, and a reply from your phone wakes it back up. I run the categories that need an answer in visible windows and leave the fire-and-forget ones headless.

If you go the visible-window route, count the windows and refuse to open more than a handful. They never close themselves.

What it Costs

Very little to have running, since the expensive layer only fires on real work. A quiet minute is one Gmail API call. Your bill is then one small triage session per message that gets past the denylist, plus however many handlers the day’s actual work demands. Work out those two numbers for your own inbox before you turn it on: a few dozen messages a day disappears inside a subscription plan with generous limits, and several hundred does not.

What you pay in the first two weeks is attention. You read the log every morning and discover your routing table has opinions you didn’t intend to give it. Every guard listed above is a line I added after one of those mornings.

Where to Take it Next

The inbox is the least interesting part of this. The pattern underneath is a cheap scheduled watcher, a classifier that emits structured routing decisions, and an executor that runs in the right folder with a written brief and permission to stop.

Point those three layers at whatever queue you spend your day in. Form submissions, a support inbox, a Slack channel, GitHub issues, purchase orders, calendar changes. You rewrite the poller’s source and the routing table. Everything else carries over untouched.