Antigravity × Google Workspace API で構築するビジネス自動化エージェント — Gmail・Drive・Sheets・Docs 完全連携実装ガイド

# requirements.txt
google-auth==2.29.0
google-auth-oauthlib==1.2.0
google-auth-httplib2==0.2.0
google-api-python-client==2.125.0
google-generativeai==0.5.3
 
# setup.py: 認証モジュール
import os
import json
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
 
# 必要なスコープを全て宣言する（後から追加するとトークン再認証が必要）
SCOPES = [
    "https://www.googleapis.com/auth/gmail.modify",
    "https://www.googleapis.com/auth/drive",
    "https://www.googleapis.com/auth/spreadsheets",
    "https://www.googleapis.com/auth/documents",
]
 
TOKEN_PATH = Path("token.json")
CREDENTIALS_PATH = Path("credentials.json")  # Google Cloud Consoleからダウンロード
 
def get_credentials() -> Credentials:
    """OAuth 2.0 認証情報を取得する。token.jsonが存在する場合は再利用する。"""
    creds = None
    
    if TOKEN_PATH.exists():
        creds = Credentials.from_authorized_user_file(str(TOKEN_PATH), SCOPES)
    
    # 期限切れの場合はリフレッシュトークンで更新
    if creds and creds.expired and creds.refresh_token:
        try:
            creds.refresh(Request())
        except Exception as e:
            print(f"トークン更新失敗: {e} — 再認証が必要です")
            creds = None
    
    # 有効な認証情報がない場合のみブラウザ認証フローを起動
    if not creds or not creds.valid:
        if not CREDENTIALS_PATH.exists():
            raise FileNotFoundError(
                "credentials.json が見つかりません。"
                "Google Cloud Console からOAuth 2.0クライアントIDをダウンロードしてください。"
            )
        flow = InstalledAppFlow.from_client_secrets_file(str(CREDENTIALS_PATH), SCOPES)
        creds = flow.run_local_server(port=0)
        
        # リフレッシュトークンをファイルに保存（次回以降は無人実行可能）
        TOKEN_PATH.write_text(creds.to_json())
        print(f"✅ 認証成功。{TOKEN_PATH} に保存しました。")
    
    return creds
 
def build_services():
    """全Workspace APIのサービスオブジェクトを一括で構築する。"""
    creds = get_credentials()
    return {
        "gmail": build("gmail", "v1", credentials=creds),
        "drive": build("drive", "v3", credentials=creds),
        "sheets": build("sheets", "v4", credentials=creds),
        "docs": build("docs", "v1", credentials=creds),
    }

import base64
from email.mime.text import MIMEText
from typing import Optional
import google.generativeai as genai
 
class GmailAgent:
    """Gmail の受信メールを AI で分類・処理するエージェント。"""
    
    # 自動処理しない信頼スコアの下限（これ以下はDraftに保存）
    AUTO_REPLY_THRESHOLD = 0.85
    
    def __init__(self, services: dict, gemini_api_key: str):
        self.gmail = services["gmail"]
        genai.configure(api_key=gemini_api_key)
        self.model = genai.GenerativeModel("gemini-2.0-flash")
        
    def fetch_unread_emails(self, query: str = "is:unread -from:me", max_results: int = 10) -> list[dict]:
        """未読メールを取得する。queryでフィルタリング可能。"""
        try:
            result = self.gmail.users().messages().list(
                userId="me",
                q=query,
                maxResults=max_results
            ).execute()
            
            messages = result.get("messages", [])
            if not messages:
                return []
            
            emails = []
            for msg in messages:
                detail = self.gmail.users().messages().get(
                    userId="me",
                    id=msg["id"],
                    format="full"
                ).execute()
                
                headers = {h["name"]: h["value"] for h in detail["payload"]["headers"]}
                body = self._extract_body(detail["payload"])
                
                emails.append({
                    "id": msg["id"],
                    "from": headers.get("From", ""),
                    "subject": headers.get("Subject", ""),
                    "body": body[:3000],  # 長すぎる場合は先頭3000文字
                    "thread_id": detail["threadId"],
                })
            
            return emails
            
        except Exception as e:
            print(f"❌ メール取得エラー: {e}")
            return []
    
    def _extract_body(self, payload: dict) -> str:
        """メールペイロードからテキスト本文を抽出する。"""
        body = ""
        if "parts" in payload:
            for part in payload["parts"]:
                if part["mimeType"] == "text/plain":
                    data = part["body"].get("data", "")
                    body = base64.urlsafe_b64decode(data).decode("utf-8", errors="ignore")
                    break
        elif "body" in payload:
            data = payload["body"].get("data", "")
            if data:
                body = base64.urlsafe_b64decode(data).decode("utf-8", errors="ignore")
        return body
    
    def analyze_and_reply(self, email: dict) -> dict:
        """
        Gemini API でメールの意図を分析し、返信文を生成する。
        戻り値: {"action": "send"|"draft"|"skip", "reply": str, "confidence": float, "category": str}
        """
        prompt = f"""
あなたはメール対応アシスタントです。以下のメールを分析してください。
 
送信者: {email['from']}
件名: {email['subject']}
本文:
{email['body']}
 
次のJSON形式で回答してください（コードブロックなし）:
{{
  "category": "inquiry|support|spam|notification|other",
  "confidence": 0.0から1.0の数値,
  "should_reply": true or false,
  "reply_draft": "返信文（日本語）。should_replyがfalseなら空文字",
  "reason": "判断理由を1文で"
}}
"""
        try:
            response = self.model.generate_content(prompt)
            import json
            result = json.loads(response.text.strip())
            
            # 信頼スコアに基づいて自動送信するか Draft 保存するか決定
            if result["should_reply"] and result["confidence"] >= self.AUTO_REPLY_THRESHOLD:
                action = "send"
            elif result["should_reply"]:
                action = "draft"  # 人間のレビューが必要
            else:
                action = "skip"
            
            return {
                "action": action,
                "reply": result.get("reply_draft", ""),
                "confidence": result["confidence"],
                "category": result["category"],
                "reason": result.get("reason", ""),
            }
            
        except Exception as e:
            print(f"❌ Gemini API エラー: {e}")
            # エラー時は必ず draft に回す（自動送信しない）
            return {"action": "draft", "reply": "", "confidence": 0.0, "category": "unknown", "reason": str(e)}
    
    def send_or_save_draft(self, original_email: dict, analysis: dict) -> Optional[str]:
        """分析結果に基づいてメールを送信または Draft として保存する。"""
        if analysis["action"] == "skip" or not analysis["reply"]:
            print(f"  ↳ スキップ（カテゴリ: {analysis['category']}）")
            return None
        
        # 返信メールを構築
        reply_text = analysis["reply"]
        message = MIMEText(reply_text, "plain", "utf-8")
        message["to"] = original_email["from"]
        message["subject"] = f"Re: {original_email['subject']}"
        message["In-Reply-To"] = original_email["id"]
        
        raw = base64.urlsafe_b64encode(message.as_bytes()).decode("utf-8")
        body = {
            "raw": raw,
            "threadId": original_email["thread_id"]
        }
        
        try:
            if analysis["action"] == "send":
                sent = self.gmail.users().messages().send(userId="me", body=body).execute()
                print(f"  ✅ 送信完了 (信頼スコア: {analysis['confidence']:.2f})")
                return sent["id"]
            else:  # draft
                draft = self.gmail.users().drafts().create(
                    userId="me", body={"message": body}
                ).execute()
                print(f"  📝 Draft保存 (信頼スコア: {analysis['confidence']:.2f} — 要レビュー)")
                return draft["id"]
        except Exception as e:
            print(f"  ❌ 送信エラー: {e}")
            return None

from googleapiclient.errors import HttpError
import copy
 
class DocumentAgent:
    """Google Drive と Docs を使って文書を自動生成するエージェント。"""
    
    def __init__(self, services: dict, gemini_api_key: str):
        self.drive = services["drive"]
        self.docs = services["docs"]
        genai.configure(api_key=gemini_api_key)
        self.model = genai.GenerativeModel("gemini-2.0-flash")
    
    def create_from_template(
        self,
        template_id: str,
        replacements: dict[str, str],
        output_folder_id: str,
        title: str
    ) -> str:
        """
        Googleドキュメントのテンプレートをコピーして変数を置換し、文書を生成する。
        
        Args:
            template_id: テンプレートドキュメントのID
            replacements: {"{{変数名}}": "置換後の値"} の形式
            output_folder_id: 生成ファイルを保存するフォルダのID
            title: 生成するファイルのタイトル
        Returns:
            生成されたドキュメントのID
        """
        # テンプレートをコピー
        try:
            copied = self.drive.files().copy(
                fileId=template_id,
                body={
                    "name": title,
                    "parents": [output_folder_id]
                }
            ).execute()
            doc_id = copied["id"]
            print(f"  📄 テンプレートをコピーしました: {title}")
        except HttpError as e:
            raise RuntimeError(f"テンプレートのコピーに失敗しました: {e}")
        
        # 変数を一括置換するリクエストを構築
        requests = [
            {
                "replaceAllText": {
                    "containsText": {
                        "text": placeholder,
                        "matchCase": True
                    },
                    "replaceText": value
                }
            }
            for placeholder, value in replacements.items()
        ]
        
        if requests:
            try:
                self.docs.documents().batchUpdate(
                    documentId=doc_id,
                    body={"requests": requests}
                ).execute()
                print(f"  ✅ {len(requests)}個の変数を置換しました")
            except HttpError as e:
                # 置換失敗でもファイルは残す（部分的な成果は保持）
                print(f"  ⚠️ 変数置換中にエラー: {e}")
        
        return doc_id
    
    def generate_release_notes(self, app_name: str, version: str, changes: list[str]) -> str:
        """
        アプリのリリースノートを Gemini で自動生成してドキュメントにまとめる。
        戻り値: 生成した Google Doc の URL
        """
        # Gemini でリリースノートの文章を生成
        prompt = f"""
以下の変更点を元に、App Store / Google Play 向けのリリースノートを日本語で作成してください。
ユーザーフレンドリーな表現で、技術的な専門用語は避けてください。
200文字以内でまとめてください。
 
アプリ名: {app_name}
バージョン: {version}
変更内容:
{chr(10).join(f"- {c}" for c in changes)}
"""
        response = self.model.generate_content(prompt)
        release_notes = response.text.strip()
        
        # テンプレートの変数
        replacements = {
            "{{APP_NAME}}": app_name,
            "{{VERSION}}": version,
            "{{RELEASE_DATE}}": "2026-04-20",
            "{{RELEASE_NOTES}}": release_notes,
            "{{RAW_CHANGES}}": "\n".join(f"• {c}" for c in changes),
        }
        
        # ここではテンプレートIDと保存先フォルダIDを環境変数から取得する
        import os
        template_id = os.environ.get("RELEASE_NOTE_TEMPLATE_ID", "")
        folder_id = os.environ.get("RELEASE_NOTES_FOLDER_ID", "")
        
        if not template_id or not folder_id:
            # テンプレートなしの場合は新規ドキュメントを作成
            doc = self.docs.documents().create(
                body={"title": f"{app_name} v{version} リリースノート"}
            ).execute()
            doc_id = doc["documentId"]
            # 本文を挿入
            self.docs.documents().batchUpdate(
                documentId=doc_id,
                body={"requests": [{
                    "insertText": {
                        "location": {"index": 1},
                        "text": f"{app_name} v{version}\n\n{release_notes}\n\n変更詳細:\n{chr(10).join(f'• {c}' for c in changes)}"
                    }
                }]}
            ).execute()
        else:
            doc_id = self.create_from_template(
                template_id=template_id,
                replacements=replacements,
                output_folder_id=folder_id,
                title=f"{app_name} v{version} リリースノート"
            )
        
        url = f"https://docs.google.com/document/d/{doc_id}/edit"
        print(f"  ✅ リリースノート生成: {url}")
        return url

from datetime import datetime, timezone, timedelta
from typing import Any
 
class SheetsAgent:
    """Google Sheets へのデータ書き込みとレポート生成を担うエージェント。"""
    
    JST = timezone(timedelta(hours=9))
    
    def __init__(self, services: dict, gemini_api_key: str):
        self.sheets = services["sheets"]
        genai.configure(api_key=gemini_api_key)
        self.model = genai.GenerativeModel("gemini-2.0-flash")
    
    def append_daily_metrics(
        self,
        spreadsheet_id: str,
        sheet_name: str,
        metrics: dict[str, Any]
    ) -> None:
        """
        日次メトリクスをスプレッドシートに追記する。
        metrics は {"ダウンロード数": 100, "収益": 5000, ...} の形式。
        """
        today = datetime.now(self.JST).strftime("%Y-%m-%d")
        row = [today] + list(metrics.values())
        
        try:
            self.sheets.spreadsheets().values().append(
                spreadsheetId=spreadsheet_id,
                range=f"{sheet_name}\!A:Z",
                valueInputOption="USER_ENTERED",
                insertDataOption="INSERT_ROWS",
                body={"values": [row]}
            ).execute()
            print(f"  ✅ {today} のデータを追記しました ({len(row)-1}項目)")
        except HttpError as e:
            print(f"  ❌ スプレッドシート書き込みエラー: {e}")
            raise
    
    def generate_weekly_summary(
        self,
        spreadsheet_id: str,
        data_sheet: str,
        summary_sheet: str
    ) -> str:
        """
        過去7日分のデータを読み込み、Gemini で週次サマリーを生成してシートに書き込む。
        戻り値: 生成されたインサイトテキスト
        """
        # 過去7行のデータを取得
        try:
            result = self.sheets.spreadsheets().values().get(
                spreadsheetId=spreadsheet_id,
                range=f"{data_sheet}\!A:Z"
            ).execute()
            all_rows = result.get("values", [])
        except HttpError as e:
            print(f"  ❌ データ読み込みエラー: {e}")
            return ""
        
        if len(all_rows) < 2:
            print("  ⚠️ データが不足しています（ヘッダー除いて最低1行必要）")
            return ""
        
        header = all_rows[0]
        recent_rows = all_rows[-7:]  # 最新7行
        
        # データをテキスト形式に変換してGeminiへ
        data_text = "\n".join(
            ", ".join(f"{h}: {v}" for h, v in zip(header, row))
            for row in recent_rows
        )
        
        prompt = f"""
以下は過去7日間のアプリパフォーマンスデータです。
このデータを分析して、以下の観点でインサイトを日本語で生成してください:
 
1. 今週のハイライト（最も良かった点）
2. 注意が必要なトレンド
3. 来週に試すべきアクション（具体的に1つ）
 
データ:
{data_text}
 
300文字以内で簡潔にまとめてください。
"""
        response = self.model.generate_content(prompt)
        insight = response.text.strip()
        
        # サマリーシートに書き込む
        today = datetime.now(self.JST).strftime("%Y-%m-%d")
        try:
            self.sheets.spreadsheets().values().append(
                spreadsheetId=spreadsheet_id,
                range=f"{summary_sheet}\!A:B",
                valueInputOption="USER_ENTERED",
                insertDataOption="INSERT_ROWS",
                body={"values": [[today, insight]]}
            ).execute()
            print(f"  ✅ 週次サマリーを {summary_sheet} に書き込みました")
        except HttpError as e:
            print(f"  ❌ サマリー書き込みエラー: {e}")
        
        return insight

# Business Automation Agents
 
## Gmail Agent
Role: メール受信監視と自動返信
Triggers:
  - 新着未読メール（15分間隔でポーリング）
  - subject_keywords: ["お問い合わせ", "サポート", "inquiry"]
Actions:
  - fetch_unread_emails(query="is:unread is:inbox")
  - analyze_and_reply(email)
  - send_or_save_draft(email, analysis)
Error handling:
  - API error → retry 3 times with 30s backoff
  - Gemini timeout → save as draft, log error
 
## Document Agent
Role: リリースノートとレポートの自動生成
Triggers:
  - Git tag push (via webhook)
  - Drive 指定フォルダへのファイル追加
Actions:
  - generate_release_notes(app_name, version, changes)
Dependencies: Gmail Agent（完成通知メールを送信）
 
## Sheets Agent
Role: 日次/週次データ収集とレポート生成
Triggers:
  - 毎日 09:00 JST（日次メトリクス追記）
  - 毎週月曜 08:00 JST（週次サマリー生成）
Actions:
  - append_daily_metrics(spreadsheet_id, "Daily", metrics)
  - generate_weekly_summary(spreadsheet_id, "Daily", "Weekly Summary")

# Pub/Sub を使ったプッシュ通知設定（ポーリングより推奨）
def setup_push_notification(self, pubsub_topic: str) -> None:
    """Gmail の新着メールを Pub/Sub 経由でプッシュ受信する設定。"""
    try:
        self.gmail.users().watch(
            userId="me",
            body={
                "topicName": pubsub_topic,
                "labelIds": ["INBOX"],
                "labelFilterAction": "include"
            }
        ).execute()
        print(f"✅ Pub/Sub 通知を設定しました: {pubsub_topic}")
        print("⚠️ この設定は7日後に失効します。定期的に更新してください。")
    except HttpError as e:
        print(f"❌ Push 通知設定エラー: {e}")

def refresh_if_needed(creds: Credentials) -> Credentials:
    """必要に応じてトークンをリフレッシュする。バッチ処理の各ループで呼ぶ。"""
    if creds.expired and creds.refresh_token:
        creds.refresh(Request())
        TOKEN_PATH.write_text(creds.to_json())
    return creds

# .github/workflows/workspace-automation.yml
name: Google Workspace Automation
 
on:
  schedule:
    - cron: "0 0 * * *"   # 毎日 09:00 JST（UTC 00:00）
    - cron: "0 23 * * 0"  # 毎週月曜 08:00 JST（UTC 日曜 23:00）
  workflow_dispatch:  # 手動実行も可能
 
jobs:
  daily-metrics:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      
      - name: Python セットアップ
        uses: actions/setup-python@v5
        with:
          python-version: "3.12"
          cache: "pip"
      
      - name: 依存関係インストール
        run: pip install -r requirements.txt
      
      - name: OAuth トークンを Secrets から復元
        env:
          GOOGLE_TOKEN_JSON: ${{ secrets.GOOGLE_TOKEN_JSON }}
        run: echo "$GOOGLE_TOKEN_JSON" > token.json
      
      - name: 日次メトリクス収集
        env:
          GOOGLE_CREDENTIALS_JSON: ${{ secrets.GOOGLE_CREDENTIALS_JSON }}
          GEMINI_API_KEY: ${{ secrets.GEMINI_API_KEY }}
          SPREADSHEET_ID: ${{ secrets.SPREADSHEET_ID }}
        run: |
          echo "$GOOGLE_CREDENTIALS_JSON" > credentials.json
          python scripts/daily_metrics.py
      
      - name: 更新したトークンを Secrets に書き戻す
        # token.json が更新された場合（リフレッシュが発生した場合）に対応
        env:
          GH_TOKEN: ${{ secrets.GH_TOKEN }}
          REPO: ${{ github.repository }}
        run: |
          NEW_TOKEN=$(cat token.json)
          gh secret set GOOGLE_TOKEN_JSON --body "$NEW_TOKEN" --repo "$REPO"