A Personal Secretary AI on Gemini API and Google Workspace — Auth, Orchestration, and Approval Gates

# gmail_agent.py
import base64
from datetime import datetime, timedelta
from workspace_auth import build_service
 
def fetch_recent_emails(max_results: int = 20, days_back: int = 1) -> list[dict]:
    """
    Fetch recent unread emails — metadata and snippets only.
    Full body is explicitly not fetched here to control token costs.
    """
    service = build_service("gmail", "v1")
 
    after_date = (datetime.now() - timedelta(days=days_back)).strftime("%Y/%m/%d")
    # Server-side filter: skip promotions and social notifications
    query = f"is:unread after:{after_date} -category:promotions -category:social"
 
    result = service.users().messages().list(
        userId="me",
        q=query,
        maxResults=max_results
    ).execute()
 
    messages = result.get("messages", [])
    email_data = []
 
    for msg in messages:
        # format="metadata" means no message body in the response
        detail = service.users().messages().get(
            userId="me",
            id=msg["id"],
            format="metadata",
            metadataHeaders=["Subject", "From", "Date"]
        ).execute()
 
        headers = {h["name"]: h["value"] for h in detail.get("payload", {}).get("headers", [])}
        email_data.append({
            "id": msg["id"],
            "subject": headers.get("Subject", "(no subject)"),
            "from": headers.get("From", ""),
            "date": headers.get("Date", ""),
            "snippet": detail.get("snippet", ""),
        })
 
    return email_data
 
def get_email_body(message_id: str) -> str:
    """
    Fetch full email body — only call this for high-priority messages.
    Handles multipart emails recursively and caps at 2000 chars.
    """
    service = build_service("gmail", "v1")
    detail = service.users().messages().get(
        userId="me",
        id=message_id,
        format="full"
    ).execute()
 
    def extract_text(payload: dict) -> str:
        """Recursively extract plain text from multipart email payloads"""
        if payload.get("mimeType") == "text/plain":
            data = payload.get("body", {}).get("data", "")
            if data:
                return base64.urlsafe_b64decode(data).decode("utf-8", errors="replace")
        for part in payload.get("parts", []):
            text = extract_text(part)
            if text:
                return text
        return ""
 
    body = extract_text(detail.get("payload", {}))
    return body[:2000] if len(body) > 2000 else body

# calendar_agent.py
from datetime import datetime
from zoneinfo import ZoneInfo
from workspace_auth import build_service
 
def fetch_todays_events(timezone: str = "Asia/Tokyo") -> list[dict]:
    """
    Fetch today's calendar events, normalized to the specified timezone.
    Uses zoneinfo (Python 3.9+ stdlib) instead of pytz to avoid
    the counter-intuitive localize() behavior.
    """
    service = build_service("calendar", "v3")
    tz = ZoneInfo(timezone)
 
    now_local = datetime.now(tz)
    day_start = now_local.replace(hour=0, minute=0, second=0, microsecond=0)
    day_end = now_local.replace(hour=23, minute=59, second=59, microsecond=0)
 
    events_result = service.events().list(
        calendarId="primary",
        timeMin=day_start.isoformat(),
        timeMax=day_end.isoformat(),
        singleEvents=True,
        orderBy="startTime",
        maxResults=20,
    ).execute()
 
    formatted = []
    for event in events_result.get("items", []):
        start_raw = event.get("start", {})
        end_raw = event.get("end", {})
 
        if "dateTime" in start_raw:
            start_dt = datetime.fromisoformat(start_raw["dateTime"]).astimezone(tz)
            end_dt = datetime.fromisoformat(end_raw["dateTime"]).astimezone(tz)
            time_str = f"{start_dt.strftime('%H:%M')} - {end_dt.strftime('%H:%M')}"
        else:
            time_str = "All day"
 
        formatted.append({
            "title": event.get("summary", "(untitled)"),
            "time": time_str,
            "location": event.get("location", ""),
            "description": event.get("description", "")[:300],
            "attendees": [a.get("email", "") for a in event.get("attendees", [])],
        })
 
    return formatted

# secretary_agent.py
import json
import google.generativeai as genai
from gmail_agent import fetch_recent_emails, get_email_body
from calendar_agent import fetch_todays_events
 
WORKSPACE_TOOLS = [
    genai.protos.Tool(
        function_declarations=[
            genai.protos.FunctionDeclaration(
                name="get_todays_schedule",
                description="Fetch today's Google Calendar events. Use for schedule review and meeting prep.",
                parameters=genai.protos.Schema(
                    type=genai.protos.Type.OBJECT,
                    properties={},
                    required=[],
                ),
            ),
            genai.protos.FunctionDeclaration(
                name="get_unread_emails",
                description="Get unread emails with metadata (subject, sender, snippet). Use for priority assessment.",
                parameters=genai.protos.Schema(
                    type=genai.protos.Type.OBJECT,
                    properties={
                        "max_results": genai.protos.Schema(
                            type=genai.protos.Type.INTEGER,
                            description="Number of emails to fetch. Default 15.",
                        ),
                    },
                    required=[],
                ),
            ),
            genai.protos.FunctionDeclaration(
                name="read_email_body",
                description="Fetch the full body of a specific email. Only use for high-priority messages.",
                parameters=genai.protos.Schema(
                    type=genai.protos.Type.OBJECT,
                    properties={
                        "message_id": genai.protos.Schema(
                            type=genai.protos.Type.STRING,
                            description="Email ID from get_unread_emails response",
                        ),
                    },
                    required=["message_id"],  # Must be explicit — Gemini won't infer it
                ),
            ),
        ]
    )
]
 
def handle_function_call(function_name: str, function_args: dict) -> str:
    """
    Execute the function Gemini requested and return JSON result.
    Errors are returned as JSON too — Gemini handles them gracefully.
    """
    try:
        if function_name == "get_todays_schedule":
            events = fetch_todays_events()
            return json.dumps({"events": events, "count": len(events)}, ensure_ascii=False)
 
        elif function_name == "get_unread_emails":
            max_results = function_args.get("max_results", 15)
            emails = fetch_recent_emails(max_results=max_results)
            return json.dumps({"emails": emails, "count": len(emails)}, ensure_ascii=False)
 
        elif function_name == "read_email_body":
            message_id = function_args["message_id"]
            body = get_email_body(message_id)
            return json.dumps({"body": body}, ensure_ascii=False)
 
        else:
            return json.dumps({"error": f"Unknown function: {function_name}"})
 
    except Exception as e:
        return json.dumps({"error": str(e), "function": function_name})
 
def run_secretary_agent(user_query: str) -> str:
    """
    Main agent loop. Gemini autonomously selects which Workspace tools
    to call based on the user's query.
    """
    genai.configure(api_key="YOUR_GEMINI_API_KEY")
 
    model = genai.GenerativeModel(
        model_name="gemini-2.5-pro",
        tools=WORKSPACE_TOOLS,
        system_instruction="""You are a personal secretary AI assistant.
Use Gmail, Google Calendar, and Google Drive tools to answer
the user's questions accurately and concisely.
 
Guidelines:
- Call multiple tools sequentially when the request requires it
- Prioritize emails that need replies, have deadlines, or are from important contacts
- Summarize with appropriate discretion about personal information
- Format times in the user's local timezone""",
    )
 
    chat = model.start_chat()
    response = chat.send_message(user_query)
 
    max_iterations = 5  # Critical: prevents runaway tool-calling loops
    iteration = 0
 
    while response.candidates[0].content.parts and iteration < max_iterations:
        function_calls = [
            part for part in response.candidates[0].content.parts
            if hasattr(part, "function_call") and part.function_call.name
        ]
 
        if not function_calls:
            break  # Text response only — we're done
 
        function_responses = []
        for part in function_calls:
            fc = part.function_call
            result = handle_function_call(fc.name, dict(fc.args))
 
            # Use name-based mapping, not index-based
            # Gemini may return function calls in any order
            function_responses.append(
                genai.protos.Part(
                    function_response=genai.protos.FunctionResponse(
                        name=fc.name,
                        response={"result": result},
                    )
                )
            )
 
        response = chat.send_message(function_responses)
        iteration += 1
 
    final_text = ""
    for part in response.candidates[0].content.parts:
        if hasattr(part, "text"):
            final_text += part.text
 
    return final_text

# cost_tracker.py
import json
import datetime
from pathlib import Path
 
def log_usage(response, query: str):
    """
    Append token usage to a JSONL log file.
    Expected output:
    {"timestamp": "2026-05-03T04:45:00", "query_preview": "Summarize today's schedule",
     "prompt_tokens": 3240, "response_tokens": 512, "total_tokens": 3752}
    """
    usage = response.usage_metadata
    log_entry = {
        "timestamp": datetime.datetime.now().isoformat(),
        "query_preview": query[:50],
        "prompt_tokens": usage.prompt_token_count,
        "response_tokens": usage.candidates_token_count,
        "total_tokens": usage.total_token_count,
    }
 
    log_path = Path("~/.config/secretary-ai/usage.jsonl").expanduser()
    log_path.parent.mkdir(parents=True, exist_ok=True)
    with log_path.open("a") as f:
        f.write(json.dumps(log_entry) + "\n")
 
def estimate_monthly_cost(log_path: Path, price_per_1m_input: float = 1.25) -> dict:
    """
    Calculate estimated monthly cost from usage logs.
    price_per_1m_input: USD per million input tokens (check current Gemini API pricing)
    """
    current_month = datetime.datetime.now().strftime("%Y-%m")
    total_input = 0
    total_output = 0
    count = 0
 
    if log_path.exists():
        for line in log_path.read_text().splitlines():
            entry = json.loads(line)
            if entry["timestamp"].startswith(current_month):
                total_input += entry.get("prompt_tokens", 0)
                total_output += entry.get("response_tokens", 0)
                count += 1
 
    estimated_usd = (total_input / 1_000_000) * price_per_1m_input
    return {
        "run_count": count,
        "total_input_tokens": total_input,
        "total_output_tokens": total_output,
        "estimated_usd": round(estimated_usd, 4),
    }

import time
import random
from googleapiclient.errors import HttpError
 
def with_retry(func, max_retries: int = 3, base_delay: float = 1.0):
    """
    Apply exponential backoff retry to Google API calls.
    Only retries on 429 (rate limit) and 503 (temporary server errors).
    404 (not found) and 403 (permission denied) fail immediately — retrying won't help.
    """
    for attempt in range(max_retries):
        try:
            return func()
        except HttpError as e:
            if e.resp.status in (429, 503) and attempt < max_retries - 1:
                delay = base_delay * (2 ** attempt) + random.uniform(0, 1)
                print(f"Rate limited — retrying in {delay:.1f}s ({attempt + 1}/{max_retries})")
                time.sleep(delay)
            else:
                raise

# preflight.py — verify auth before the real work starts
from datetime import datetime, timezone
from google.auth.exceptions import RefreshError
from google.auth.transport.requests import Request
 
from workspace_auth import get_credentials
 
 
class ReauthRequired(Exception):
    """Auth needs human re-consent. Notify instead of running the job."""
 
 
def preflight(notify) -> "Credentials":
    """
    Check auth once, before any Gemini call.
    Failing here avoids the worse ordering: pay for inference, then die on auth.
    Pass a notify callable that reaches somewhere you actually read in the morning.
    """
    creds = get_credentials()
 
    if creds.expired and creds.refresh_token:
        try:
            creds.refresh(Request())
        except RefreshError as e:
            # invalid_grant lands here: revocation, the 7-day expiry, or a password change
            notify(f"[secretary] re-authorization required: {e}")
            raise ReauthRequired(str(e)) from e
 
    # Speak up just before a silent failure rather than after it
    if creds.expiry:
        remaining = creds.expiry.replace(tzinfo=timezone.utc) - datetime.now(timezone.utc)
        if remaining.total_seconds() < 300:
            notify("[secretary] access token has under five minutes left")
 
    return creds
 
 
if __name__ == "__main__":
    preflight(notify=print)

# config.py — one place for every model identifier
from dataclasses import dataclass
 
@dataclass(frozen=True)
class ModelConfig:
    """
    Model identifiers for the secretary agent.
    Scheduled runs pin a version; migration happens by editing this file.
    """
    # Triage and priority scoring — called often, so cost dominates
    triage: str = "gemini-3.5-flash"
    # Summaries and reply drafts — quality dominates, so Pro earns its price
    compose: str = "gemini-3.5-pro"
    # Migration canary. Point this at latest and diff outputs before switching
    canary: str = "gemini-flash-latest"
 
MODELS = ModelConfig()

# approval.py
import json
from enum import Enum
 
class RiskTier(str, Enum):
    """Tool risk. READ runs automatically; anything else asks first."""
    READ = "read"      # Reads only. Nothing to undo
    WRITE = "write"    # Changes state, but reversibly (creating a draft)
    SEND = "send"      # Leaves the building. Cannot be undone
 
TOOL_RISK = {
    "get_todays_schedule": RiskTier.READ,
    "get_unread_emails": RiskTier.READ,
    "read_email_body": RiskTier.READ,
    "create_draft_reply": RiskTier.WRITE,
}
 
def requires_approval(function_name: str) -> bool:
    """Anything that isn't READ needs approval. Unknown tools fail safe."""
    return TOOL_RISK.get(function_name, RiskTier.SEND) != RiskTier.READ
 
def request_approval(function_name: str, function_args: dict) -> bool:
    """
    Confirm before running a write tool.
    Under cron there is no stdin — swap this for a pending-approval queue.
    """
    print(f"\n[confirm] about to run {function_name}")
    print(json.dumps(function_args, ensure_ascii=False, indent=2))
    answer = input("Run it? [y/N]: ").strip().lower()
    return answer == "y"

def handle_function_call(function_name: str, function_args: dict) -> str:
    if requires_approval(function_name):
        if not request_approval(function_name, function_args):
            # A refusal is a result, not an exception
            return json.dumps(
                {"status": "declined", "reason": "user did not approve execution"},
                ensure_ascii=False,
            )
    # existing branches follow (get_todays_schedule / get_unread_emails / ...)