◈ API / SDK/2026-05-03Advanced

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

Designing a personal secretary AI across Gmail and Google Calendar with the Gemini API: OAuth2 scope design, Function Calling orchestration, approval gates for write tools, and ninety days of measured production cost, with full Python code.

gemini-api²⁷⁷ google-workspace⁷ function-calling²⁰ gmail⁵ google-calendar² agent¹⁰ python¹⁰⁴

✦ Premium Article

"I just want an AI that reads my Gmail and tells me what needs attention today" — you'd think that would be a weekend project. And then you spend three days fighting OAuth2 scopes, get confused by quota limits, and abandon the whole thing when the costs start climbing.

I've been there. Google Workspace APIs are genuinely powerful, but the infrastructure surrounding them — auth flows, scope management, quota types — adds enough complexity that you can exhaust yourself before writing a single line of AI logic.

This guide builds a personal secretary AI that spans Gmail, Google Calendar, and Google Drive using Gemini API Function Calling. The implementation is designed for production use: not a demo, but something you can actually run every morning and trust.

Why Secretary AI Projects Usually Fail

Before touching any code, here are three patterns that kill these projects.

Anti-pattern 1: Sequential pipeline design

Gmail → AI processes → writes to Calendar. This seems clean until you try to add Drive or Docs. The integrations become tangled, and the whole thing collapses under its own weight. The right approach from day one is to let Gemini Function Calling orchestrate tool selection — the AI decides what to call and when.

Anti-pattern 2: Mixing OAuth2 and service accounts

"OAuth2 for personal use, service accounts for teams" sounds reasonable until you need to debug a permissions error at 11pm. For personal secretary AI, pick OAuth2 and stay consistent. Service accounts are for domain-wide delegation and CI/CD pipelines.

Anti-pattern 3: Passing everything to the model without filtering

Sending your entire inbox to Gemini sounds like a good idea until you get a bill for several dollars from a single morning run. Filtering before the API call — not after — is the design constraint that keeps this usable at personal scale.

Google Workspace API Authentication Design

Two authentication approaches matter here:

OAuth2 (Authorization Code Flow): For accessing your own data — your Gmail, your Calendar, your Drive. Tokens expire hourly, so refresh token management is required. This is what you want for personal use.

Service Accounts: For domain-wide access in headless environments (CI/CD, server-to-server). Not needed for a personal secretary AI.

Here's a production-ready authentication module with automatic token refresh:

# workspace_auth.py
import logging
from pathlib import Path
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
 
logger = logging.getLogger(__name__)
 
# Principle of least privilege: only request what you actually need
SCOPES = [
    "https://www.googleapis.com/auth/gmail.readonly",
    "https://www.googleapis.com/auth/gmail.compose",
    "https://www.googleapis.com/auth/calendar.readonly",
    "https://www.googleapis.com/auth/drive.readonly",
]
 
TOKEN_PATH = Path("~/.config/secretary-ai/token.json").expanduser()
CREDENTIALS_PATH = Path("~/.config/secretary-ai/credentials.json").expanduser()
 
def get_credentials() -> Credentials:
    """
    Get or refresh OAuth2 credentials.
    First run launches a browser auth flow.
    Subsequent runs load from token file and auto-refresh if expired.
    """
    creds = None
 
    if TOKEN_PATH.exists():
        creds = Credentials.from_authorized_user_file(str(TOKEN_PATH), SCOPES)
 
    if not creds or not creds.valid:
        if creds and creds.expired and creds.refresh_token:
            try:
                creds.refresh(Request())
                logger.info("Access token refreshed successfully")
            except Exception as e:
                logger.warning(f"Token refresh failed, re-auth required: {e}")
                creds = None
 
        if not creds:
            if not CREDENTIALS_PATH.exists():
                raise FileNotFoundError(
                    f"credentials.json not found at: {CREDENTIALS_PATH}\n"
                    "Download an OAuth2 client ID from Google Cloud Console"
                )
            flow = InstalledAppFlow.from_client_secrets_file(
                str(CREDENTIALS_PATH), SCOPES
            )
            creds = flow.run_local_server(port=0)
 
        TOKEN_PATH.parent.mkdir(parents=True, exist_ok=True)
        TOKEN_PATH.write_text(creds.to_json())
        logger.info("Token saved to disk")
 
    return creds
 
def build_service(service_name: str, version: str):
    """Build a Google API service client"""
    creds = get_credentials()
    return build(service_name, version, credentials=creds)

The scope separation between gmail.readonly and gmail.compose is intentional. If a bug causes an unintended write operation, readonly prevents the damage. Apply least-privilege thinking to API scopes the same way you would to IAM policies.

✦

Thank you for reading this far.

Continue Reading

What follows includes implementation code, benchmarks, and practical content we hope you'll find useful. This site runs without ads — server and development costs are supported entirely by members like you. If it's been helpful, we'd be truly grateful for your support.

WHAT YOU'LL LEARN

✦Splitting OAuth2 scopes by read and write, and gating only the write tools behind human approval

✦What changed in output and cost when the model behind gemini-flash-latest was swapped

✦How to decide whether to keep your own orchestrator now that Managed Agents are in preview

Secure payment via Stripe · Cancel anytime

✦

Unlock This Article

Get full access to the rest of this article. Buy once, read anytime. This site is ad-free — your support goes directly toward keeping it running.

Unlock all articles with Membership →

Gmail Agent: Two-Stage Filtering for Cost-Effective Email Analysis

The naive approach — fetch all unread emails, send to Gemini — fails fast. A hundred unread messages, even just snippets, can exceed ten thousand tokens per API call.

The production pattern is two-stage filtering:

Server-side filtering with Gmail's q parameter (date, labels, sender)
Send only metadata to Gemini for priority scoring
Fetch full body only for high-priority messages

# 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

The format="metadata" parameter is the key cost control here. Gmail API quota consumption stays low, and you're only sending the data Gemini actually needs for priority scoring.

Calendar Agent: Timezone-Safe Event Fetching

Calendar API is simpler than Gmail, but timezone handling trips up almost everyone at least once. The API returns times in UTC or whatever timezone each event was created with — and getting garbled times in your morning summary is annoying enough to debug.

# 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

The datetime.fromisoformat(...).astimezone(ZoneInfo("Asia/Tokyo")) pattern is safer than pytz.localize() because it handles DST transitions correctly without the "fold" ambiguity issues.

Gemini Function Calling Orchestration — The Core Architecture

This is where the pieces come together. Rather than hardcoding which API to call when, we define the Workspace tools as Gemini function declarations and let the model decide what to invoke based on the user's request.

# 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

Production Operations: Cost Tracking, Token Refresh, and Retry Logic

This section is what separates a weekend hack from something you can actually rely on.

Token Usage Logging

The usage_metadata in every Gemini response is your cost visibility. Log it:

# 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),
    }

Running this agent twice daily on a typical inbox should cost well under a dollar per month, assuming you keep the metadata-only filtering in place.

Exponential Backoff for Google API Rate Limits

Gmail API and Calendar API have different quota types — user-level daily limits vs. per-second limits. Simple retry doesn't work reliably. Use exponential backoff with jitter:

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

Three Production Pitfalls and How to Avoid Them

These cost me time. Hopefully they don't cost you the same.

Pitfall 1: Scope additions don't apply to existing tokens

If you add a new OAuth2 scope after initial authentication, the saved token.json still has only the original scopes. You'll get insufficient authentication scopes errors even though your code looks correct. The fix is simple — delete token.json and re-authenticate — but the error message doesn't tell you that. Anytime you modify SCOPES, delete the token file first.

Pitfall 2: Omitting required in Function Declaration causes missing arguments

When required is not specified in a FunctionDeclaration, Gemini may call the function without providing the argument. For read_email_body, which needs message_id, this causes a KeyError in handle_function_call. Always explicitly declare required fields. This behavior differs from OpenAI's function calling — something to watch if you're migrating from OpenAI.

Pitfall 3: Index-based function response mapping breaks with parallel calls

When Gemini returns multiple function calls simultaneously (e.g., "check my schedule AND my emails"), the order isn't guaranteed. Building FunctionResponse objects using part.function_call.name rather than list indices is the correct approach. The implementation above does this correctly — the name-based mapping means response order doesn't matter.

Pin the Model, or Let `gemini-flash-latest` Decide?

The code above hardcodes model_name="gemini-2.5-pro". That was deliberate.

Alias identifiers like gemini-flash-latest are lighter to write and always track the newest model. But they track it on Google's schedule, not yours. When Gemini 3.5 Flash reached general availability in July 2026, the model that gemini-flash-latest resolves to changed underneath everyone using it. If your nightly job pointed at the alias, a different model started answering one morning.

For a secretary agent — where the whole premise is that the same shape of summary lands in your inbox every day — that quiet swap matters. As an indie developer running several jobs unattended overnight, I diffed outputs before and after, and the summaries carried the same information, but the priority ordering had shifted. Nothing broke. It just wasn't the same.

Identifier style	Good for	Watch out for
Pinned, e.g. `gemini-3.5-flash`	Scheduled jobs that produce a consistent output shape	You own the migration timing, and you have to track deprecations yourself
Alias, e.g. `gemini-flash-latest`	Interactive work where you read every result	The underlying model changes without notice, which can break anything downstream that depends on output shape

My recommendation: pin for scheduled agents, use aliases for experiments you watch by hand. Then keep every model name in one place so migration is a one-line change.

# 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()

Splitting triage onto Flash and drafting onto Pro pays off on both axes. Ranking unread mail by subject, sender, and snippet is a shallow task — there's little reason to spend Pro tokens on it. Drafting a reply is different: the quality of the prose is what actually saves you time.

For where the Flash/Pro line actually falls once retries are counted, see measuring retry amplification on Gemini 3.5 Flash — unit prices alone don't tell the whole story.

Put an Approval Gate in Front of Write Tools

This is the part that decided whether I'd run the agent for real.

The moment gmail.compose entered the scope list, this agent could create drafts. Function Calling means the model chooses the tools — so in principle, the AI decides on its own when to draft an email. A read-only mistake leaks information at worst. A write is different.

I underrated that difference at first. During testing I asked it to "prepare drafts for anything that needs a reply," and it dutifully wrote five. Two were addressed to people I had no intention of replying to. Nothing was sent, so there was no harm done. Still, opening that drafts folder stopped me in my tracks.

The fix is to tier tools by risk and require a human in the loop only on the writing side.

# 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"

Drop the gate at the top of handle_function_call(). The key detail: a denial is not an error. Return it to Gemini as JSON and the model understands it wasn't approved, then falls back to summarizing. Raise an exception instead and you take the whole conversation down with it.

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 / ...)

Running headless under cron, input() is off the table. I queue pending approvals to a JSONL file and surface them in the morning notification instead. The difference is between "the AI wrote five drafts" and "the AI would like to write five drafts." That one step back is what let me sleep through the job.

If you want to go further with tiering, adding risk-tiered approval gates to Gemini Function Calling covers the design in more depth.

Now That Managed Agents Exist, Should You Keep Your Own Orchestrator?

In July 2026, Managed Agents entered public preview on the Gemini API. You can run stateful, autonomous agents inside an isolated Linux sandbox that Google hosts — planning, reasoning, code execution, file operations, and web browsing all handled on their side.

So is the while loop we built in this article obsolete?

My answer: for a secretary agent, keep your own — for now. Three reasons.

Where the credentials live. The heart of this agent is a refresh token for your own Gmail and Calendar. Keeping it on your machine isn't a feature here; it's closer to a precondition
Room for the approval gate. The gate above only exists because your code sits directly in front of tool execution. Hand execution off and you go hunting for a place to insert that step
A visible ceiling. Having max_iterations = 5 in code you own is what makes the cost side sit still

The opposite case is equally clear. Managed Agents are the natural fit for disposable work that shouldn't depend on your machine, for research tasks involving code execution and file manipulation, and for agents you want to call from several environments. Not having to build a sandbox is a real saving.

Concern	Your own orchestrator	Managed Agents (public preview)
Personal OAuth tokens	Stay on your machine	Need a delivery design into the sandbox
Pre-execution approval	Sits right before the tool call	Requires a separate insertion point
Environment setup	You build it	None
Code execution, file ops	You isolate it yourself	Included in the sandbox
Best suited to	Long-lived jobs over personal data	Disposable research and automation

For getting artifacts back out of that sandbox safely, see scoped credentials and egress design for Managed Agents. It's a preview feature, so waiting for GA before committing production work to it seems reasonable.

Ninety Days of Real Numbers, and What Surprised Me

Here's the record from running this once every morning at 6:00 for roughly three months. Each run does: fetch today's schedule → fetch unread metadata → score priority → fetch bodies for the top three → generate the summary. Figures come from aggregating the JSONL that log_usage() writes.

Measure	Observed
Input tokens per run (median)	~4,100
Output tokens per run (median)	~480
Function calls per run	3–5, including three body fetches
Runs per month	30 (once each morning)
Monthly spend after moving triage to Flash	Settled into cents, not dollars

What surprised me wasn't the absolute cost — it was the distribution. The median held steady around 4,100 tokens, but two or three days a month blew past 20,000. Tracing those days, the cause was mail threads with long quoted histories: every one of the top three bodies hit the [:2000] ceiling on the same morning.

Design against the average and you miss the tail. I moved my spend alert from "three times the mean" to "anchored on the observed maximum."

One more thing. Before with_retry() went in, the job died silently on a 429 about once a month. Cron fails quietly. A morning summary can simply not arrive, and if you're busy you won't notice. I added backoff not to save money, but to remove a failure I couldn't see.

For the underlying phenomenon of tool results inflating the conversation, passing handles instead of giant tool responses goes deeper.

Start With One Morning Routine

The full integration — Gmail, Calendar, Drive, Docs — is easier to build when you start with a working single-tool version. Pick one that would save you the most time today. "Send me a Slack message at 8am with today's calendar summary" is a twenty-line script using the calendar agent code above.

Once that's running reliably, adding Gmail priority scoring takes an hour. After that, Drive search adds another hour. The architecture is designed so each tool is independent — adding new ones doesn't require touching the existing code.

The auth, cost tracking, and retry logic from this guide should go in from day one. These aren't optimizations you add later — they're what makes the difference between "works on my machine" and "runs every morning without attention."

For more on Gemini API Function Calling fundamentals, see Gemini API Function Calling: tool integration in practice. For keeping multi-turn tool calls intact, preserving Gemini 3 thought signatures in production is a useful companion.

Thank You for Reading

Gemini Lab is ad-free, supported entirely by members like you. We publish practical guides daily with implementation code, benchmarks, and production-ready patterns. If you've found it useful, we'd love to have you on board.