Build a Personal AI Secretary with Gemini API — Task Automation, Email Summaries & Schedule Optimization for Solopreneurs

# Python 3.11 or higher is recommended
pip install google-generativeai google-auth google-auth-oauthlib \
    google-api-python-client slack-sdk python-dotenv \
    pydantic rich schedule

# config.py
import os
from dotenv import load_dotenv
 
load_dotenv()
 
# Gemini API — store your actual key in .env as GEMINI_API_KEY=YOUR_KEY
GEMINI_API_KEY = os.getenv("GEMINI_API_KEY")
GEMINI_MODEL_PRO = "gemini-2.5-pro-preview-03-25"
GEMINI_MODEL_FLASH = "gemini-2.5-flash-preview-04-17"
 
# Slack
SLACK_WEBHOOK_URL = os.getenv("SLACK_WEBHOOK_URL")
 
# Google OAuth
GOOGLE_CREDENTIALS_FILE = "credentials.json"
GOOGLE_TOKEN_FILE = "token.json"
SCOPES = [
    "https://www.googleapis.com/auth/gmail.readonly",
    "https://www.googleapis.com/auth/gmail.compose",
    "https://www.googleapis.com/auth/calendar",
]

# tools/definitions.py
from google.generativeai.types import FunctionDeclaration, Tool
 
# --- Gmail tools ---
get_unread_emails = FunctionDeclaration(
    name="get_unread_emails",
    description="Fetch unread emails from Gmail. Supports filtering by importance level and sender domain.",
    parameters={
        "type": "object",
        "properties": {
            "max_results": {
                "type": "integer",
                "description": "Maximum number of emails to return (default: 10)",
            },
            "sender_filter": {
                "type": "string",
                "description": "Filter by a specific sender domain, e.g. '@client.com'",
            },
            "importance": {
                "type": "string",
                "enum": ["high", "normal", "low", "all"],
                "description": "Filter by Gmail importance level",
            },
        },
        "required": [],
    },
)
 
create_draft_reply = FunctionDeclaration(
    name="create_draft_reply",
    description="Save a reply draft to Gmail for a specified message. Does NOT send — user reviews before sending.",
    parameters={
        "type": "object",
        "properties": {
            "message_id": {
                "type": "string",
                "description": "Gmail message ID of the email being replied to",
            },
            "draft_body": {
                "type": "string",
                "description": "Body of the reply (Markdown formatting accepted)",
            },
            "tone": {
                "type": "string",
                "enum": ["formal", "casual", "technical"],
                "description": "Writing tone for the draft",
            },
        },
        "required": ["message_id", "draft_body"],
    },
)
 
# --- Calendar tools ---
get_today_schedule = FunctionDeclaration(
    name="get_today_schedule",
    description="Fetch today's Google Calendar events. Optionally include tomorrow's events as well.",
    parameters={
        "type": "object",
        "properties": {
            "include_tomorrow": {
                "type": "boolean",
                "description": "Whether to also include tomorrow's events",
            },
        },
        "required": [],
    },
)
 
add_focus_block = FunctionDeclaration(
    name="add_focus_block",
    description="Add a deep-work focus block to Google Calendar in an available time slot",
    parameters={
        "type": "object",
        "properties": {
            "title": {
                "type": "string",
                "description": "Event title, e.g. 'Deep Work: Article Writing'",
            },
            "start_time": {
                "type": "string",
                "description": "Start time in ISO 8601 format, e.g. '2026-03-31T09:00:00+09:00'",
            },
            "duration_minutes": {
                "type": "integer",
                "description": "Duration of the focus block in minutes",
            },
            "description": {
                "type": "string",
                "description": "Optional detailed description",
            },
        },
        "required": ["title", "start_time", "duration_minutes"],
    },
)
 
# --- Task management tools ---
get_task_list = FunctionDeclaration(
    name="get_task_list",
    description="Retrieve the current task list, including priority scores and deadlines",
    parameters={
        "type": "object",
        "properties": {
            "status_filter": {
                "type": "string",
                "enum": ["todo", "in_progress", "done", "all"],
                "description": "Filter tasks by status",
            },
        },
        "required": [],
    },
)
 
update_task_priority = FunctionDeclaration(
    name="update_task_priority",
    description="Update the priority of a task. Used when the AI determines a reprioritization is warranted based on context.",
    parameters={
        "type": "object",
        "properties": {
            "task_id": {"type": "string", "description": "Task ID"},
            "priority": {
                "type": "string",
                "enum": ["critical", "high", "medium", "low"],
            },
            "reason": {
                "type": "string",
                "description": "AI-generated explanation for the priority change",
            },
        },
        "required": ["task_id", "priority"],
    },
)
 
# Assembled tool set
SECRETARY_TOOLS = Tool(
    function_declarations=[
        get_unread_emails,
        create_draft_reply,
        get_today_schedule,
        add_focus_block,
        get_task_list,
        update_task_priority,
    ]
)

# tools/executor.py
import json
from datetime import datetime, timedelta, timezone
from googleapiclient.discovery import build
from google.oauth2.credentials import Credentials
from config import GOOGLE_TOKEN_FILE
 
JST = timezone(timedelta(hours=9))
 
def _get_gmail_service():
    creds = Credentials.from_authorized_user_file(GOOGLE_TOKEN_FILE)
    return build("gmail", "v1", credentials=creds)
 
def _get_calendar_service():
    creds = Credentials.from_authorized_user_file(GOOGLE_TOKEN_FILE)
    return build("calendar", "v3", credentials=creds)
 
def execute_tool(tool_name: str, tool_args: dict) -> dict:
    """Execute the tool requested by Gemini and return the result."""
    try:
        if tool_name == "get_unread_emails":
            return _get_unread_emails(**tool_args)
        elif tool_name == "create_draft_reply":
            return _create_draft_reply(**tool_args)
        elif tool_name == "get_today_schedule":
            return _get_today_schedule(**tool_args)
        elif tool_name == "add_focus_block":
            return _add_focus_block(**tool_args)
        elif tool_name == "get_task_list":
            return _get_task_list(**tool_args)
        elif tool_name == "update_task_priority":
            return _update_task_priority(**tool_args)
        else:
            return {"error": f"Unknown tool: {tool_name}"}
    except Exception as e:
        return {"error": str(e), "tool": tool_name}
 
def _get_unread_emails(max_results=10, sender_filter=None, importance="all"):
    """Retrieve unread Gmail messages."""
    service = _get_gmail_service()
    query = "is:unread"
    if sender_filter:
        query += f" from:{sender_filter}"
    if importance == "high":
        query += " is:important"
 
    results = service.users().messages().list(
        userId="me", q=query, maxResults=max_results
    ).execute()
 
    messages = results.get("messages", [])
    email_list = []
 
    for msg in messages:
        detail = service.users().messages().get(
            userId="me", messageId=msg["id"], format="metadata",
            metadataHeaders=["Subject", "From", "Date"]
        ).execute()
 
        headers = {h["name"]: h["value"] for h in detail["payload"]["headers"]}
        snippet = detail.get("snippet", "")
 
        email_list.append({
            "id": msg["id"],
            "subject": headers.get("Subject", "(no subject)"),
            "from": headers.get("From", ""),
            "date": headers.get("Date", ""),
            "snippet": snippet[:200],
            "importance": "high" if "IMPORTANT" in detail.get("labelIds", []) else "normal",
        })
 
    return {"emails": email_list, "total": len(email_list)}
 
def _add_focus_block(title: str, start_time: str, duration_minutes: int, description: str = ""):
    """Add a focus block event to Google Calendar."""
    service = _get_calendar_service()
 
    start_dt = datetime.fromisoformat(start_time)
    end_dt = start_dt + timedelta(minutes=duration_minutes)
 
    event = {
        "summary": f"🎯 {title}",
        "description": description or "Focus block created by AI Secretary",
        "start": {"dateTime": start_dt.isoformat(), "timeZone": "Asia/Tokyo"},
        "end": {"dateTime": end_dt.isoformat(), "timeZone": "Asia/Tokyo"},
        "colorId": "9",  # Blueberry for visibility
        "reminders": {
            "useDefault": False,
            "overrides": [{"method": "popup", "minutes": 5}],
        },
    }
 
    created_event = service.events().insert(
        calendarId="primary", body=event
    ).execute()
 
    return {
        "status": "created",
        "event_id": created_event["id"],
        "title": title,
        "start": start_time,
        "duration_minutes": duration_minutes,
        "link": created_event.get("htmlLink", ""),
    }

# agent/secretary.py
import google.generativeai as genai
from config import GEMINI_API_KEY, GEMINI_MODEL_PRO
from tools.definitions import SECRETARY_TOOLS
from tools.executor import execute_tool
 
genai.configure(api_key=GEMINI_API_KEY)
 
SYSTEM_PROMPT = """You are a highly capable AI secretary supporting an independent professional.
 
## Your Role
- Maintain a holistic view of the user's tasks, emails, and schedule
- Proactively suggest the best course of action for the day
- Call available tools as needed to read and write calendar, email, and task data
- Briefly explain your reasoning so the user can review and adjust
 
## Priority Framework
- Deadline within 24 hours: critical
- Messages from clients or partners: high
- Revenue-generating tasks: high
- Everything else: medium or low
 
## Operating Principles
- Never overload a single day. Keep deep-work blocks to 90 minutes maximum
- Draft replies should be concise — communicate the key point in 3 sentences or fewer
- Adapt to the user's personal working style
"""
 
def run_secretary_session(user_request: str, verbose: bool = False) -> str:
    """
    Execute an AI secretary session with automatic Function Calling loop.
 
    Args:
        user_request: Natural language request from the user
        verbose: Whether to print tool call logs
 
    Returns:
        Gemini's final text response
    """
    model = genai.GenerativeModel(
        model_name=GEMINI_MODEL_PRO,
        system_instruction=SYSTEM_PROMPT,
        tools=[SECRETARY_TOOLS],
    )
 
    chat = model.start_chat()
    response = chat.send_message(user_request)
 
    # Function Calling loop — up to 10 iterations
    max_iterations = 10
    iteration = 0
 
    while iteration < max_iterations:
        iteration += 1
 
        if not response.candidates[0].content.parts:
            break
 
        has_function_call = False
        tool_results = []
 
        for part in response.candidates[0].content.parts:
            if hasattr(part, "function_call") and part.function_call:
                has_function_call = True
                fc = part.function_call
 
                if verbose:
                    print(f"🔧 Tool call: {fc.name}({dict(fc.args)})")
 
                result = execute_tool(fc.name, dict(fc.args))
 
                if verbose:
                    print(f"✅ Result: {str(result)[:200]}...")
 
                tool_results.append(
                    genai.protos.Part(
                        function_response=genai.protos.FunctionResponse(
                            name=fc.name,
                            response={"result": result},
                        )
                    )
                )
 
        if not has_function_call:
            break
 
        response = chat.send_message(tool_results)
 
    for part in response.candidates[0].content.parts:
        if hasattr(part, "text") and part.text:
            return part.text
 
    return "(Failed to generate a response)"

# jobs/morning_briefing.py
from datetime import datetime, timezone, timedelta
from agent.secretary import run_secretary_session
from slack_sdk.webhook import WebhookClient
from config import SLACK_WEBHOOK_URL
 
JST = timezone(timedelta(hours=9))
 
def run_morning_briefing():
    today = datetime.now(JST).strftime("%B %d, %Y")
 
    request = f"""
    Today is {today}. Please complete the following tasks in order:
 
    1. Fetch today's calendar and insert one 90-minute deep work block in the best available slot
    2. Retrieve up to 5 high-importance unread emails and summarize each in 1–2 sentences
    3. Pull the task list and tell me the top 3 tasks I should tackle today (critical and high priority only)
    4. Combine all of this into a concise "today's battle plan" in plain English
 
    Use the most important task's name as the title of the deep work block.
    """
 
    result = run_secretary_session(request, verbose=False)
 
    if SLACK_WEBHOOK_URL:
        client = WebhookClient(SLACK_WEBHOOK_URL)
        client.send(text=f"🌅 *Morning Briefing — {today}*\n\n{result}")
        print("✅ Sent to Slack")
 
    return result
 
if __name__ == "__main__":
    print(run_morning_briefing())

# jobs/evening_report.py
from agent.secretary import run_secretary_session
from slack_sdk.webhook import WebhookClient
from config import SLACK_WEBHOOK_URL
from datetime import datetime, timezone, timedelta
 
JST = timezone(timedelta(hours=9))
 
def run_evening_report():
    today = datetime.now(JST).strftime("%B %d, %Y")
 
    request = f"""
    Generate an end-of-day reflection report for {today}.
 
    1. Fetch the task list and count completed vs. remaining tasks
    2. Get today's calendar and summarize completed events plus tomorrow's schedule
    3. Flag any important unread emails that still need attention (max 3)
 
    Finish with:
    - Completion rate for today (done / total tasks)
    - One thing to tackle first thing tomorrow morning
    - A one-sentence summary (under 30 words) wrapping it all up
    """
 
    result = run_secretary_session(request)
 
    if SLACK_WEBHOOK_URL:
        client = WebhookClient(SLACK_WEBHOOK_URL)
        client.send(text=f"🌙 *Evening Report — {today}*\n\n{result}")
 
    return result

# Dockerfile
FROM python:3.11-slim
 
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
 
COPY . .
 
ENV GOOGLE_APPLICATION_CREDENTIALS="/app/service-account.json"
 
CMD ["python", "-m", "uvicorn", "api.main:app", "--host", "0.0.0.0", "--port", "8080"]

# api/main.py — Cloud Run entry point
from fastapi import FastAPI, HTTPException, Header
from jobs.morning_briefing import run_morning_briefing
from jobs.evening_report import run_evening_report
import os
 
app = FastAPI()
INTERNAL_TOKEN = os.getenv("INTERNAL_TOKEN", "")
 
def verify_token(x_internal_token: str = Header(None)):
    if x_internal_token != INTERNAL_TOKEN:
        raise HTTPException(status_code=403, detail="Forbidden")
 
@app.post("/jobs/morning")
async def morning(x_internal_token: str = Header(None)):
    verify_token(x_internal_token)
    result = run_morning_briefing()
    return {"status": "ok", "summary": result[:200]}
 
@app.post("/jobs/evening")
async def evening(x_internal_token: str = Header(None)):
    verify_token(x_internal_token)
    result = run_evening_report()
    return {"status": "ok", "summary": result[:200]}

# Build and deploy to Cloud Run
gcloud run deploy ai-secretary \
  --source . \
  --region asia-northeast1 \
  --memory 512Mi \
  --timeout 300 \
  --set-env-vars "GEMINI_API_KEY=YOUR_GEMINI_API_KEY,SLACK_WEBHOOK_URL=YOUR_SLACK_WEBHOOK_URL" \
  --no-allow-unauthenticated
 
# Schedule morning briefing at 8 AM on weekdays
gcloud scheduler jobs create http morning-briefing \
  --schedule="0 8 * * 1-5" \
  --uri="https://YOUR_CLOUD_RUN_URL/jobs/morning" \
  --http-method=POST \
  --headers="X-Internal-Token=YOUR_INTERNAL_TOKEN" \
  --time-zone="Asia/Tokyo"

def run_simple_task(request: str) -> str:
    """Use Flash for lightweight tasks like summarization or classification."""
    import google.generativeai as genai
    from config import GEMINI_MODEL_FLASH
 
    model = genai.GenerativeModel(GEMINI_MODEL_FLASH)
    response = model.generate_content(request)
    return response.text
 
def run_complex_task(request: str) -> str:
    """Reserve Pro for multi-step reasoning and autonomous decision-making."""
    return run_secretary_session(request)

from google.oauth2.credentials import Credentials
from google.auth.transport.requests import Request
 
def get_valid_credentials():
    creds = None
    if os.path.exists(GOOGLE_TOKEN_FILE):
        creds = Credentials.from_authorized_user_file(GOOGLE_TOKEN_FILE, SCOPES)
 
    if not creds or not creds.valid:
        if creds and creds.expired and creds.refresh_token:
            creds.refresh(Request())
        # else: run OAuth flow again
 
        with open(GOOGLE_TOKEN_FILE, "w") as token:
            token.write(creds.to_json())
 
    return creds

import time, random
 
def call_with_retry(func, max_retries=3, base_delay=2):
    for attempt in range(max_retries):
        try:
            return func()
        except Exception as e:
            if "429" in str(e) 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