業務の中で「また同じメールに同じような返信を書いている」「スプレッドシートへのデータ転記を毎週手動でやっている」と感じたことはないでしょうか。
私がアプリ開発の傍らで直面したのもまさにそれでした。問い合わせメールの対応、リリースノートの作成、アプリの収益データのシート整理——どれも重要な作業ですが、毎回手を動かすのは非効率です。これらを Antigravity と Google Workspace API、そして Gemini API を組み合わせて自動化したとき、週に数時間分の作業が完全になくなりました。
ここではその仕組みを完全に再現できるように、認証設定から本番運用まで順を追って解説します。単なるチュートリアルではなく、実際に私が運用しているシステムの設計思想と、途中でぶつかった落とし穴も含めてお伝えします。
全体アーキテクチャの設計思想
実装に入る前に、システム全体の構造を整理しておきます。
今回構築するのは、4つのGoogle Workspace APIをAgentsとして連携させる自律型パイプラインです。
- Gmail エージェント: 受信メールを分類し、Gemini API で意図を解析。ルールに基づいて自動返信を生成・送信
- Drive エージェント: 指定フォルダを監視し、新しいファイルをトリガーに処理を開始。テンプレートから文書を自動生成
- Sheets エージェント: 外部APIやDBからデータを収集してスプレッドシートに書き込み、サマリーレポートを生成
- Docs エージェント: Sheetsのデータを元にGoogleドキュメントを自動生成し、指定のユーザーと共有
Antigravity の agents.md でこれら4つのエージェントの役割を定義し、Gemini API がコンテンツ判断のコアを担います。
重要な設計判断として、すべての自動返信に「人間レビューフラグ」を実装しています。信頼スコアが一定以下のメールは自動送信せず、Draft として保存する設計にしました。これは誤送信のリスクをゼロにするための必須安全弁です。
Google Workspace API 認証の正しい設定
Workspace APIへの接続で最初に迷うのが認証方式の選択です。2種類あります。
サービスアカウント vs OAuth 2.0 の使い分け
サービスアカウント（推奨：サーバーサイド自動化）
ユーザーの操作なしにバックグラウンドで動作させるには、サービスアカウントが最適です。ただし、Gmailへのアクセスにはドメイン管理者によるドメイン全体の委任（Domain-Wide Delegation）が必要です。Google Workspace（旧G Suite）の有料プランを使っている場合はこちらを使いましょう。
OAuth 2.0（推奨：個人アカウントや小規模利用）
個人のGmailアカウントや、管理者権限のない環境では OAuth 2.0 を使います。初回のみブラウザ認証が必要ですが、リフレッシュトークンをファイルに保存することで以降は無人実行が可能です。
今回は OAuth 2.0 + リフレッシュトークン保存方式 で実装します。個人開発者に最も現実的な方法です。
Python クライアントの初期設定
# 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),
}重要なポイント: SCOPEは後から追加するとtoken.jsonを削除して再認証が必要になります。設計段階で必要なスコープを全て確定させてから実装を始めてください。私はこれを見落として2回再認証する羽目になりました。
Gmail エージェント — 受信メール自動処理の実装
メール取得フィルターの設計
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この設計で特に気を付けているのは「エラー時は必ずdraftに回す」という原則です。Gemini APIが失敗した場合に空のメールが自動送信されるのは最悪のシナリオです。エラーハンドリングを書くときは「失敗したらどうなるか」を常に最初に考えるようにしています。
Google Drive + Docs — 文書生成パイプラインの実装
テンプレート駆動の文書自動生成
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 urlGoogle Sheets エージェント — データ集計とレポート自動生成
アプリの収益データやユーザー数を毎週手動でスプレッドシートに貼り付けていた作業が、このエージェントで完全になくなりました。
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 insightAntigravity でマルチエージェントオーケストレーションを設定する
上記の3つのエージェントを Antigravity で連携させるための agents.md 設定を示します。
# 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")Antigravity の Agent パネルで agents.md を認識させると、各エージェントのトリガーと依存関係が視覚的に確認できます。エージェントファーストのワークフロー設計を参考にしながら、このシステムを構築しました。
よくある落とし穴と回避策
実際に運用する中で遭遇した問題を3つ挙げます。
1. Gmail API のクォータは「1日の総量」ではなく「1分あたり」で効く
ポーリング間隔を決めるとき、私は1日に何ユニット使うかを数えていました。ところが実際に 429 を踏んだのは、未読が溜まった朝にバッチが一気に処理へ入った瞬間でした。数え方を間違えていたわけです。
Gmail API のクォータは公式の Usage limits に3種類が並んでいます。個人開発の規模で効いてくる順に並べ替えると、こうなります。
| 種別 | 上限 | 個人規模で当たるか |
|---|---|---|
| 1分あたり / ユーザーあたり / プロジェクトあたり | 6,000 クォータユニット | ここに当たる |
| 1分あたり / プロジェクトあたり | 1,200,000 クォータユニット | まず当たらない |
| 1日あたり / プロジェクトあたり（課金しきい値） | 80,000,000 クォータユニット | まず当たらない |
3番目は「これを超えると課金が発生する」という billing threshold であって、超えた瞬間に止まる壁ではありません。壁として立ちはだかるのは1行目だけです。
メソッド単価も直感とずれます。本記事の fetch_unread_emails が呼んでいるものを中心に、Gmail API の Usage limits から抜き出します。
| メソッド | クォータユニット |
|---|---|
history.list | 2 |
messages.list | 5 |
messages.modify | 5 |
drafts.create | 10 |
messages.get | 20 |
messages.send | 100 |
users.watch | 100 |
一覧が 5 で、本文取得が 20。max_results=10 の1回のポーリングは 5 + 20 × 10 = 205 ユニットです。ここまでは何の問題もありません。
危ないのは max_results を大きくしたときです。未読 300 件を一度に取りに行くと messages.get だけで 6,000 ユニット。1分あたり / ユーザーあたりの上限にちょうど届き、途中から 429 rateLimitExceeded が返り始めます。上限に当たるかどうかを決めるのは、1日の総量ではなく1分間の詰め込み方です。
解決策(1): 1分あたりの予算をクライアント側で持つ。 残ユニットを教えてくれる API はないので、自分で数えます。
import time
from collections import deque
class QuotaBudget:
"""1分あたり/ユーザーあたり 6,000 ユニットを自前で見張る。"""
COST = {
"history.list": 2,
"messages.list": 5,
"messages.modify": 5,
"drafts.create": 10,
"messages.get": 20,
"messages.send": 100,
"users.watch": 100,
}
def __init__(self, limit_per_minute: int = 6000, safety: float = 0.8):
# 他プロセスと同じユーザー枠を共有するため安全率をかけて 4,800 で運用する
self.budget = int(limit_per_minute * safety)
self.window: deque[tuple[float, int]] = deque()
def _spent(self) -> int:
now = time.monotonic()
while self.window and now - self.window[0][0] > 60:
self.window.popleft()
return sum(cost for _, cost in self.window)
def spend(self, method: str) -> None:
"""API 呼び出しの直前に通す。予算を超えるならウィンドウが空くまで待つ。"""
cost = self.COST[method]
while self._spent() + cost > self.budget:
oldest = self.window[0][0]
sleep_for = max(60 - (time.monotonic() - oldest) + 0.5, 1.0)
print(f"⏳ クォータ予算に到達。{sleep_for:.1f} 秒待機します")
time.sleep(sleep_for)
self.window.append((time.monotonic(), cost))fetch_unread_emails のループ内で budget.spend("messages.get") を挟むだけで、429 を踏む前に自分から減速してくれます。エラーを受けてから謝るのではなく、踏まないように歩幅を変える。運用に載せてから、この違いは大きいと感じています。
解決策(2): 差分同期に切り替えて単価を下げる。 毎回 messages.list で全体を舐め直す必要はありません。history.list（2ユニット）で前回からの差分だけを取れば、受信のない時間帯の消費はほぼ消えます。
def fetch_new_since(self, start_history_id: str) -> tuple[list[str], str]:
"""前回の historyId 以降に増えたメッセージ ID だけを返す。"""
ids: list[str] = []
page_token = None
latest = start_history_id
while True:
res = self.gmail.users().history().list(
userId="me",
startHistoryId=start_history_id,
historyTypes=["messageAdded"],
pageToken=page_token,
).execute()
for entry in res.get("history", []):
for added in entry.get("messagesAdded", []):
ids.append(added["message"]["id"])
latest = res.get("historyId", latest)
page_token = res.get("nextPageToken")
if not page_token:
break
return ids, lateststartHistoryId は前回処理した時点の値を token.json の隣に保存しておきます。注意点がひとつ。Gmail が保持する history は無期限ではないため、しばらく止めていたバッチを再開すると 404 が返ります。そのときは messages.list による全件同期へ落とす分岐を必ず用意してください。ここを省くと、復旧したつもりのバッチが静かに空振りし続けます。
未読が平均3件、15分間隔で1日96回まわす前提で、上の単価表から計算するとこうなります。
| 方式 | 1回あたり | 1日あたり |
|---|---|---|
messages.list + messages.get ×10（当初） | 205 | 19,680 |
history.list + 差分のみ messages.get ×3 | 62 | 5,952 |
Pub/Sub 通知 + 差分のみ messages.get | 受信時のみ | 実受信件数 × 20 ＋ 購読更新分 |
いずれも1分あたりの上限に対しては十分に小さい数字です。それでも差分同期へ寄せる価値があるのは、朝のスパイクが平らになるからです。1日の合計を減らすためではなく、瞬間最大風速を下げるために切り替えます。
解決策(3): Pub/Sub 通知でポーリングそのものを消す。 users.watch は 100 ユニットを消費しますが、購読を張ってしまえば受信のない時間帯の消費はゼロになります。
# 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}")購読の有効期限は7日です。GitHub Actions の日次ジョブに users.watch の呼び直しを混ぜておくのが確実で、更新を忘れると通知が来なくなったことにも気づけません。
なお予算管理を入れても 429 が完全になくなるわけではありません。同じ Google アカウントを別のツールが叩いていれば、こちらの見積もりは外れます。返ってきたときは指数バックオフで待ってから同じリクエストを投げ直します。即座に再試行すると、ウィンドウが空かないまま二重に叩くことになります。
import random
import time
from googleapiclient.errors import HttpError
RETRYABLE = (403, 429, 500, 502, 503, 504)
def call_with_backoff(request, max_attempts: int = 5):
"""429 / 5xx を指数バックオフで再試行する。request は execute() 前のオブジェクト。"""
for attempt in range(max_attempts):
try:
return request.execute()
except HttpError as e:
if e.resp.status not in RETRYABLE:
raise
wait = (2 ** attempt) + random.uniform(0, 1)
print(f"⚠️ {e.resp.status} を受信。{wait:.1f} 秒後に再試行します（{attempt + 1}/{max_attempts}）")
time.sleep(wait)
raise RuntimeError("バックオフ上限に達しました。処理を中断します")403 を再試行対象に含めているのは、Gmail API がレート超過を 403 userRateLimitExceeded として返すことがあるためです。スコープ不足の 403 は再試行しても直りませんが、そちらは次項のとおり最初の1回で必ず落ちるので、実運用では取り違えずに済みます。
2. OAuth トークンの有効期限（1時間）
access_tokenは1時間で失効します。長時間バッチ処理をしていると途中で認証エラーが発生します。
解決策: get_credentials() 関数でリフレッシュ処理を実装していますが、長時間処理では明示的に再認証するようにしました。
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 creds3. スコープ設定のミスによる 403 エラー
https://www.googleapis.com/auth/gmail.modify ではなく gmail.readonly を設定して返信送信が 403 になったことがあります。drive スコープも drive.readonly にすると書き込みができません。
一度認証したトークンにスコープを追加するには token.json を削除して再認証が必要です。開発初期に正しいスコープを確定させる点が肝心です。
本番運用とスケジューリング
GitHub Actions を使ってサーバーなしで定期実行できます。
# .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"重要: token.json はシークレットに保存してください。GitHubリポジトリにコミットしてはいけません（.gitignore に追加必須）。また、GitHub Actions でリフレッシュが発生したとき、更新後のトークンをシークレットに書き戻す処理も忘れずに実装してください。
バックグラウンドエージェントの実践活用と組み合わせると、Antigravity上でのデバッグと GitHub Actions での本番運用を効率的に切り替えられます。
実際の運用結果とこれからの展望
このシステムを3週間運用した結果、問い合わせメール対応に使っていた週2〜3時間が、週20〜30分程度まで縮みました。ゼロにはなりません。返信の自動送信は全メールの約60%（confidence が 0.85 以上）で、残り40%は Draft として保存され、私が読んで送る形が残っているためです。
この40%を無理に0へ寄せなかったのは、しきい値を下げた途端に「送ってはいけない返信」が混ざる感触があったからです。判断の一部を手元に残すことを、コストではなく設計上の選択として受け入れています。
特に効果的だったのは SheetsエージェントによるAdMob収益の自動記録です。Firebase コンソールからの手動コピー&ペースト作業が完全になくなり、毎週月曜に週次インサイトが自動生成されるようになりました。
次のステップとして考えているのは、MCPサーバーの構築を使って、このシステムをAntigravityの内部ツールとして直接呼び出せるようにすることです。エージェントの実行状況をIDE上からリアルタイムで確認できるようになれば、さらに使いやすくなるはずです。
今日できる最初の一歩として、まず get_credentials() を実行してOAuth認証を完了させることをおすすめします。認証さえ通れば、あとは各エージェントを順番に試していくだけです。