Build an Autonomous AI Agent to Manage Your Inbox in 30 Minutes

pip install google-auth google-auth-oauthlib google-auth-httplib2 google-api-python-client

from google.auth.transport.requests import Request
from google.oauth2.credentials import Credentials
from google_auth_oauthlib.flow import InstalledAppFlow
from googleapiclient.discovery import build
import os

SCOPES = [
    "https://www.googleapis.com/auth/gmail.readonly",
    "https://www.googleapis.com/auth/gmail.modify",   # needed for labels
    "https://www.googleapis.com/auth/gmail.compose",  # needed for drafts
]

def get_gmail_service():
    creds = None
    if os.path.exists("token.json"):
        creds = Credentials.from_authorized_user_file("token.json", SCOPES)

    if not creds or not creds.valid:
        if creds and creds.expired and creds.refresh_token:
            creds.refresh(Request())
        else:
            flow = InstalledAppFlow.from_client_secrets_file("credentials.json", SCOPES)
            creds = flow.run_local_server(port=0)
        with open("token.json", "w") as f:
            f.write(creds.to_json())

    return build("gmail", "v1", credentials=creds)

from langchain.tools import tool
from langchain_core.messages import BaseMessage
import base64
import json

service = get_gmail_service()

@tool
def list_unread_emails(max_results: int = 10) -> str:
    """
    Returns a JSON list of unread emails with id, sender, subject, and snippet.
    Call this first to see what needs attention.
    """
    results = service.users().messages().list(
        userId="me",
        labelIds=["UNREAD"],
        maxResults=max_results
    ).execute()

    messages = results.get("messages", [])
    emails = []
    for msg in messages:
        detail = service.users().messages().get(
            userId="me", id=msg["id"], format="metadata",
            metadataHeaders=["From", "Subject"]
        ).execute()
        headers = {h["name"]: h["value"] for h in detail["payload"]["headers"]}
        emails.append({
            "id": msg["id"],
            "from": headers.get("From", ""),
            "subject": headers.get("Subject", ""),
            "snippet": detail.get("snippet", "")
        })

    return json.dumps(emails, indent=2)


@tool
def read_email_body(message_id: str) -> str:
    """
    Returns the full plain-text body of a single email given its message_id.
    Use this when the snippet isn't enough to understand the email.
    """
    msg = service.users().messages().get(
        userId="me", id=message_id, format="full"
    ).execute()

    def extract_text(payload):
        # Walk MIME parts to find plain text
        if payload.get("mimeType") == "text/plain":
            data = payload["body"].get("data", "")
            return base64.urlsafe_b64decode(data).decode("utf-8", errors="ignore")
        for part in payload.get("parts", []):
            result = extract_text(part)
            if result:
                return result
        return ""

    return extract_text(msg["payload"]) or "(no plain text body)"


@tool
def apply_label(message_id: str, label_name: str) -> str:
    """
    Applies a Gmail label to a message. Creates the label if it doesn't exist.
    label_name should be one of: 'needs-reply', 'waiting', 'newsletter', 'archive'.
    """
    # Find or create label
    labels = service.users().labels().list(userId="me").execute().get("labels", [])
    label_id = next((l["id"] for l in labels if l["name"] == label_name), None)

    if not label_id:
        new_label = service.users().labels().create(
            userId="me", body={"name": label_name}
        ).execute()
        label_id = new_label["id"]

    service.users().messages().modify(
        userId="me",
        id=message_id,
        body={"addLabelIds": [label_id], "removeLabelIds": ["UNREAD"]}
    ).execute()

    return f"Label '{label_name}' applied and marked as read."


@tool
def create_draft_reply(message_id: str, reply_body: str) -> str:
    """
    Creates a Gmail draft reply to a given message. Does NOT send it.
    The user will review and send it manually. reply_body is plain text.
    """
    original = service.users().messages().get(
        userId="me", id=message_id, format="metadata",
        metadataHeaders=["From", "Subject", "Message-ID", "To"]
    ).execute()

    headers = {h["name"]: h["value"] for h in original["payload"]["headers"]}
    to = headers.get("From", "")
    subject = headers.get("Subject", "")
    if not subject.startswith("Re:"):
        subject = f"Re: {subject}"
    thread_id = original["threadId"]

    # Encode the MIME message
    mime_message = f"To: {to}\nSubject: {subject}\nContent-Type: text/plain\n\n{reply_body}"
    encoded = base64.urlsafe_b64encode(mime_message.encode()).decode()

    draft = service.users().drafts().create(
        userId="me",
        body={"message": {"raw": encoded, "threadId": thread_id}}
    ).execute()

    return f"Draft created with id {draft['id']}. Subject: '{subject}'"

from langchain_anthropic import ChatAnthropic
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate

llm = ChatAnthropic(model="claude-sonnet-4-6", temperature=0)

tools = [list_unread_emails, read_email_body, apply_label, create_draft_reply]

prompt = ChatPromptTemplate.from_messages([
    ("system", """You are an inbox management assistant. Your job:

1. List unread emails
2. Read any that need more context than the snippet provides
3. Apply one label per email: 'needs-reply', 'waiting', 'newsletter', or 'archive'
4. For emails labeled 'needs-reply', create a draft reply

Rules:
- Never send emails. Only create drafts.
- Apply exactly one label per email. 
- Keep draft replies professional and under 150 words.
- If unsure about something, label it 'needs-reply' and leave a draft asking for clarification."""),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}"),
])

agent = create_tool_calling_agent(llm, tools, prompt)
executor = AgentExecutor(agent=agent, tools=tools, verbose=True, max_iterations=30)

result = executor.invoke({
    "input": "Process my unread emails. Label each one appropriately and draft replies where needed."
})

print(result["output"])

python inbox_agent.py