Build Your Personal AI Chief of Staff with Gmail, Calendar, and Notion

                          You (CLI / Slack / iMessage)
                                     |
                                     v
                       +---------------------------+
                       |   Main Chief of Staff     |
                       |   (claude-sonnet-4.5)     |
                       |   + conversation_history  |
                       +---------------------------+
                          |          |          |
                          |   MCP tools attached:
                          |   - Gmail
                          |   - Google Calendar
                          |   - Notion
                          |
                          | delegates to sub-agents when work gets specialized
                          v
          +-----------------+-----------------+-----------------+
          |                 |                 |                 |
          v                 v                 v                 v
  +---------------+ +---------------+ +---------------+
  | Inbox Triager | | Meeting Prep  | | Notion Logger |
  | (haiku-4.5)   | | (gpt-4.1-mini)| | (haiku-4.5)   |
  +---------------+ +---------------+ +---------------+

export SWARMS_API_KEY="sk-..."
export GMAIL_MCP_URL="https://mcp.composio.dev/gmail/<your-account-id>"
export GCAL_MCP_URL="https://mcp.composio.dev/googlecalendar/<your-account-id>"
export NOTION_MCP_URL="https://mcp.composio.dev/notion/<your-account-id>"

import json
import os

import requests
from dotenv import load_dotenv

load_dotenv()

API_KEY = os.getenv("SWARMS_API_KEY")
BASE_URL = "https://api.swarms.world"

GMAIL_MCP_URL = os.getenv("GMAIL_MCP_URL")
GCAL_MCP_URL = os.getenv("GCAL_MCP_URL")
NOTION_MCP_URL = os.getenv("NOTION_MCP_URL")

headers = {"x-api-key": API_KEY, "Content-Type": "application/json"}

INBOX_TRIAGER = {
    "agent_name": "Inbox-Triager",
    "description": "Classifies and summarizes recent Gmail messages into priority buckets.",
    "system_prompt": (
        "You triage Gmail. For each message you read, classify it as "
        "URGENT, FOLLOW-UP, FYI, or NOISE. Summarize the actually-important "
        "ones in one line each, with sender and subject. Skip newsletters, "
        "calendar invites, and notifications unless they require a response. "
        "Return a compact digest grouped by priority bucket."
    ),
    "model_name": "claude-haiku-4.5",
    "max_loops": "auto",
    "mcp_url": GMAIL_MCP_URL,
}

MEETING_PREP = {
    "agent_name": "Meeting-Prep",
    "description": "Pulls upcoming calendar events and drafts a prep brief.",
    "system_prompt": (
        "You prepare the user for upcoming meetings. Read the next 24-72 hours "
        "of calendar events, identify the attendees, and produce a one-paragraph "
        "brief per meeting: who, what it's about, what to bring or decide, and "
        "any obvious recent email thread context. When asked, email the brief "
        "to the user."
    ),
    "model_name": "gpt-4.1-mini",
    "max_loops": "auto",
    "mcp_configs": {
        "connections": [
            {"url": GCAL_MCP_URL},
            {"url": GMAIL_MCP_URL},
        ]
    },
}

NOTION_LOGGER = {
    "agent_name": "Notion-Logger",
    "description": "Appends structured notes and updates to Notion pages and databases.",
    "system_prompt": (
        "You write to Notion. When the user mentions a deal, person, or project, "
        "find the right page or database row and append a structured note with "
        "today's date. Do not create duplicates — search first. Keep notes "
        "concise and factual; one bullet per fact."
    ),
    "model_name": "claude-haiku-4.5",
    "max_loops": "auto",
    "mcp_url": NOTION_MCP_URL,
}

def build_cos_config():
    return {
        "agent_name": "Chief-of-Staff",
        "description": (
            "Personal Chief of Staff with live access to Gmail, Google Calendar, "
            "and Notion. Delegates specialized work to a team of sub-agents."
        ),
        "system_prompt": (
            "You are the user's personal Chief of Staff. You have live access to "
            "their Gmail, Google Calendar, and Notion via MCP tools, and you can "
            "delegate to three specialist sub-agents: Inbox-Triager, Meeting-Prep, "
            "and Notion-Logger.\n\n"
            "Rules:\n"
            "- Default to action. If the user says 'schedule X', schedule it; "
            "  don't ask which calendar unless there's genuine ambiguity.\n"
            "- Delegate when the work is bulk or specialized. A single 'reply to "
            "  Sarah' you handle yourself. 'Triage my inbox' goes to Inbox-Triager.\n"
            "- Remember context from earlier in the conversation. If the user "
            "  mentioned the Acme deal at 9am, that's the same Acme at 4pm.\n"
            "- For calendar slots, return them as structured ISO datetimes.\n"
            "- Talk like a competent human CoS, not a chatbot. No 'I'd be happy to'."
        ),
        "model_name": "claude-sonnet-4.5",
        "max_loops": "auto",
        "max_tokens": 8192,
        "temperature": 0.2,
        "mcp_configs": {
            "connections": [
                {"url": GMAIL_MCP_URL},
                {"url": GCAL_MCP_URL},
                {"url": NOTION_MCP_URL},
            ]
        },
        "handoffs": [INBOX_TRIAGER, MEETING_PREP, NOTION_LOGGER],
    }


def call_cos(task: str, history: list | None = None) -> dict:
    payload = {"agent_config": build_cos_config(), "task": task}
    if history:
        payload["history"] = history

    response = requests.post(
        f"{BASE_URL}/v1/agent/completions",
        headers=headers,
        json=payload,
        timeout=300,
    )
    response.raise_for_status()
    return response.json()


def extract_reply(result: dict) -> str:
    outputs = result.get("outputs")
    if isinstance(outputs, list):
        for item in reversed(outputs):
            if not isinstance(item, dict):
                continue
            role = (item.get("role") or "").lower()
            if role in ("user", "system"):
                continue
            content = item.get("content")
            if isinstance(content, list):
                return " ".join(str(c) for c in content)
            if content:
                return str(content)
    return json.dumps(outputs or result, indent=2)

def cos_repl():
    history: list[dict] = []
    print("Chief of Staff online. Ctrl-C to quit.\n")
    while True:
        try:
            user_msg = input("you > ").strip()
        except (EOFError, KeyboardInterrupt):
            print()
            break
        if not user_msg:
            continue

        result = call_cos(task=user_msg, history=history or None)
        reply = extract_reply(result)
        print(f"cos > {reply}\n")

        history.append({"role": "user", "content": user_msg})
        history.append({"role": "assistant", "content": reply})


if __name__ == "__main__":
    cos_repl()

you > what did Sarah email me about yesterday?

cos > Sarah Chen sent two threads yesterday. The big one is the Q3 partnership
      proposal — she's asking for a call this week to walk through the revised
      terms (she dropped the rev-share ask from 30% to 22%). The other is a
      quick heads-up that she'll be OOO Friday. Want me to propose times for
      the call?

you > yeah, schedule contract review with Sarah Thursday after 3

cos > Booked. Thursday 3:30-4:30pm with Sarah Chen, titled "Q3 Partnership —
      Contract Review." Invite sent to sarah@acmecorp.com with the revised
      terms PDF from yesterday's thread attached. Calendar link is in your
      inbox.

you > add a note to the Acme deal in notion: they want net-60 terms

cos > Logged. Appended to "Acme Corp — Q3 Partnership" in your Deals database
      under today's date: "Acme requested net-60 payment terms (vs. our
      standard net-30). Surfaced during contract review prep, 5/28."

# digest.py — run via cron: 0 7 * * * /usr/bin/python3 /path/to/digest.py

def morning_digest():
    payload = {
        "agent_config": build_cos_config(),
        "task": (
            "It's 7am. Run my morning routine:\n"
            "1. Delegate to Inbox-Triager: read the last 18 hours of email and "
            "   give me the URGENT and FOLLOW-UP buckets only.\n"
            "2. Delegate to Meeting-Prep: brief me on every meeting today and "
            "   tomorrow morning.\n"
            "3. Compose the combined digest as a single email, subject "
            "   \"Morning Brief — <today's date>\", and send it to me via Gmail.\n"
            "Be terse. I'm reading this with one eye open."
        ),
    }
    response = requests.post(
        f"{BASE_URL}/v1/agent/completions",
        headers=headers,
        json=payload,
        timeout=600,
    )
    response.raise_for_status()
    print(extract_reply(response.json()))


if __name__ == "__main__":
    morning_digest()

crontab -e
# add: 0 7 * * * /usr/bin/python3 /Users/you/cos/digest.py >> /tmp/cos.log 2>&1