取り組みの背景:なぜ「AI秘書」が個人事業主に必要か
一人で事業を回すということは、営業・開発・経理・スケジュール管理・顧客対応——すべてを自分でこなすということです。時間は有限であり、これらの反復的なタスクに多くの時間を奪われると、本当に集中すべき創造的な仕事ができなくなります。
Gemini API が2025年後半から提供している高度な Function Calling と長文脈理解(最大100万トークン)は、こうした課題を根本から変える力を持っています。ここではGemini 2.5 Pro を中核に据えた個人事業主向けAI秘書システムを一から構築し、本番運用可能なレベルまで仕上げる全工程を詳しく解説します。
この記事で構築するシステムの全体像
本記事のAI秘書システムは次の4つの機能から成ります。
- タスク自動優先付け: メール・Slack・カレンダーから得たコンテキストを元に、今日のタスクを自動でスコアリング・並べ替える
- メール要約と返信草案生成: Gmailの未読メールを重要度順に整理し、返信草案を自動生成する
- スケジュール最適化: 会議の合間の「深作業ブロック」を自動で確保し、Google Calendarに書き込む
- 日次レポート自動生成: 毎晩20時に1日の作業サマリーをSlackに送信する
対象読者は Python の基礎知識があり、Google Cloud の基本操作に慣れている個人開発者・フリーランス・インディーハッカー の方々です。
アーキテクチャ概要
システム全体は以下のコンポーネントで構成されます。
データソース層
- Gmail API(メール取得)
- Google Calendar API(スケジュール取得・書き込み)
- ローカルタスクファイル(JSON/Markdown)
AIエンジン層
- Gemini 2.5 Pro(主要推論・Function Calling)
- Gemini 2.5 Flash(軽量タスク・コスト削減)
実行環境
- Cloud Run(サーバーレスコンテナ)
- Cloud Scheduler(定期実行)
- Secret Manager(APIキー管理)
出力先
- Slack Webhook(通知・レポート)
- Google Calendar(スケジュール書き込み)
- ローカルファイル / Notion API(タスク更新)
このアーキテクチャの最大の特徴は、すべての外部サービス連携を Gemini の Function Calling で統一制御している点です。Gemini が「Gmail から未読メールを取得して」「優先度の高い3件を返信草案付きで教えて」という自然言語指示を受け取り、必要なツールを自律的に選択・実行します。
環境準備
必要なライブラリのインストール
# Python 3.11以上を推奨
pip install google-generativeai google-auth google-auth-oauthlib \
google-api-python-client slack-sdk python-dotenv \
pydantic rich schedule認証情報の準備
Google Cloud プロジェクトで以下のAPIを有効化してください。
- Gmail API
- Google Calendar API
- Generative Language API
# config.py
import os
from dotenv import load_dotenv
load_dotenv()
# Gemini API
GEMINI_API_KEY = os.getenv("GEMINI_API_KEY") # AIzaSy... ではなく YOUR_GEMINI_API_KEY 形式で .env に記載
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",
]コアモジュール①:Function Calling の設計
AI秘書システムの心臓部は Gemini への Function Calling 定義です。Gemini は、ツール定義を見て「何ができるか」を理解し、ユーザーの要求に応じて必要なツールを自律的に呼び出します。
# tools/definitions.py
from google.generativeai.types import FunctionDeclaration, Tool
# --- Gmail ツール ---
get_unread_emails = FunctionDeclaration(
name="get_unread_emails",
description="Gmailから未読メールを取得する。重要度フィルタリングや送信者絞り込みが可能",
parameters={
"type": "object",
"properties": {
"max_results": {
"type": "integer",
"description": "取得する最大件数(デフォルト: 10)",
},
"sender_filter": {
"type": "string",
"description": "特定の送信者でフィルタする(例: '@client.com')",
},
"importance": {
"type": "string",
"enum": ["high", "normal", "low", "all"],
"description": "重要度フィルタ",
},
},
"required": [],
},
)
create_draft_reply = FunctionDeclaration(
name="create_draft_reply",
description="指定メールへの返信草案をGmailの下書きとして保存する",
parameters={
"type": "object",
"properties": {
"message_id": {
"type": "string",
"description": "返信対象のGmailメッセージID",
},
"draft_body": {
"type": "string",
"description": "返信本文(Markdown形式可)",
},
"tone": {
"type": "string",
"enum": ["formal", "casual", "technical"],
"description": "文体のトーン",
},
},
"required": ["message_id", "draft_body"],
},
)
# --- Calendar ツール ---
get_today_schedule = FunctionDeclaration(
name="get_today_schedule",
description="今日のGoogleカレンダーの予定を取得する",
parameters={
"type": "object",
"properties": {
"include_tomorrow": {
"type": "boolean",
"description": "明日の予定も含めるか",
},
},
"required": [],
},
)
add_focus_block = FunctionDeclaration(
name="add_focus_block",
description="Googleカレンダーに集中作業ブロック(深作業時間)を追加する",
parameters={
"type": "object",
"properties": {
"title": {
"type": "string",
"description": "ブロックのタイトル(例: '深作業: 記事執筆')",
},
"start_time": {
"type": "string",
"description": "開始時刻 ISO 8601形式(例: '2026-03-31T09:00:00+09:00')",
},
"duration_minutes": {
"type": "integer",
"description": "作業ブロックの長さ(分)",
},
"description": {
"type": "string",
"description": "ブロックの詳細説明",
},
},
"required": ["title", "start_time", "duration_minutes"],
},
)
# --- タスク管理ツール ---
get_task_list = FunctionDeclaration(
name="get_task_list",
description="現在のタスクリストを取得し、優先度と締め切りを含めて返す",
parameters={
"type": "object",
"properties": {
"status_filter": {
"type": "string",
"enum": ["todo", "in_progress", "done", "all"],
"description": "ステータスでフィルタ",
},
},
"required": [],
},
)
update_task_priority = FunctionDeclaration(
name="update_task_priority",
description="タスクの優先度を更新する。AIが文脈を判断して自動スコアリングする際に使用",
parameters={
"type": "object",
"properties": {
"task_id": {"type": "string", "description": "タスクID"},
"priority": {
"type": "string",
"enum": ["critical", "high", "medium", "low"],
},
"reason": {
"type": "string",
"description": "優先度変更の理由(AIが生成する説明)",
},
},
"required": ["task_id", "priority"],
},
)
# ツールセット定義
SECRETARY_TOOLS = Tool(
function_declarations=[
get_unread_emails,
create_draft_reply,
get_today_schedule,
add_focus_block,
get_task_list,
update_task_priority,
]
)コアモジュール②:ツール実行エンジン
Function Calling で Gemini がツール呼び出しを要求したとき、実際に処理を行うエグゼキューターを実装します。
# 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:
"""Gemini が呼び出したツールを実行し、結果を返す"""
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"未定義のツール: {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"):
"""Gmail未読メール取得の実装"""
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", "(件名なし)"),
"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 = ""):
"""Googleカレンダーに集中作業ブロックを追加"""
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 "AI秘書が自動設定した集中作業ブロック",
"start": {"dateTime": start_dt.isoformat(), "timeZone": "Asia/Tokyo"},
"end": {"dateTime": end_dt.isoformat(), "timeZone": "Asia/Tokyo"},
"colorId": "9", # ブルーベリー色で視認性UP
"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", ""),
}コアモジュール③:AI秘書エージェントループ
Gemini の Function Calling はマルチターン形式で動作します。Gemini が「ツールを呼び出したい」と応答したら、こちらで実行して結果を返し、Gemini が最終回答を生成するまでこのループを繰り返します。
# 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 = """あなたは個人事業主をサポートする優秀なAI秘書です。
## あなたの役割
- ユーザーのタスク・メール・スケジュールを総合的に把握し、最適な行動を提案する
- 必要に応じてツールを呼び出し、実際にカレンダーやGmailを操作する
- 判断の根拠を簡潔に説明し、ユーザーが内容を確認・修正しやすくする
## 優先度の判断基準
- 締め切りが24時間以内:critical
- クライアント・パートナーからの連絡:high
- 売上に直結するタスク:high
- その他の作業:medium または low
## 行動原則
- 一度に多くを詰め込まない。集中できる作業ブロックは最大90分
- 返信草案は簡潔に。3文以内で用件を伝えることを意識する
- ユーザーの作業スタイルに合わせて柔軟に対応する
"""
def run_secretary_session(user_request: str, verbose: bool = False) -> str:
"""
AI秘書セッションを実行する。Function Callingループを自動処理。
Args:
user_request: ユーザーからの自然言語リクエスト
verbose: ツール呼び出しのログを出力するか
Returns:
Geminiの最終回答テキスト
"""
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 ループ(最大10回まで)
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"🔧 ツール呼び出し: {fc.name}({dict(fc.args)})")
# ツールを実行
result = execute_tool(fc.name, dict(fc.args))
if verbose:
print(f"✅ 結果: {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
# ツール実行結果を Gemini に返す
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 "(応答の生成に失敗しました)"実際の使用例:朝のモーニングブリーフィング
以下のコードを毎朝8時に Cloud Scheduler で実行するだけで、AI秘書が今日のスケジュールを最適化し、重要メールの要約をSlackに送ります。
# jobs/morning_briefing.py
import json
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():
"""毎朝8時に実行するモーニングブリーフィング"""
today = datetime.now(JST).strftime("%Y年%m月%d日")
request = f"""
今日は{today}です。以下のタスクを順番に実行してください:
1. 今日のカレンダーを取得して、会議の合間に90分の深作業ブロックを最適な時間帯に1つ追加する
2. 重要度「high」の未読メール(最大5件)を取得し、各メールを1〜2文で要約する
3. タスクリストを取得し、今日やるべき「critical」と「high」のタスクをトップ3で教える
4. これらをまとめた「今日の作戦」を日本語で簡潔に教える
カレンダーの深作業ブロックには「今日の最重要タスク名」をタイトルに入れてください。
"""
result = run_secretary_session(request, verbose=False)
# Slack に送信
if SLACK_WEBHOOK_URL:
client = WebhookClient(SLACK_WEBHOOK_URL)
client.send(
text=f"🌅 *{today} モーニングブリーフィング*\n\n{result}"
)
print("✅ Slack に送信完了")
return result
if __name__ == "__main__":
result = run_morning_briefing()
print(result)実行例(出力イメージ):
🌅 2026年3月31日 モーニングブリーフィング
📅 **カレンダー最適化**
・10:00〜12:00 に「🎯 深作業: 新機能実装」ブロックを追加しました
・午後は14:00〜16:00 にミーティングが2件あります
📧 **重要メール TOP 3**
1. [ABCクライアント] 契約更新の件 → 今週中に返答が必要
2. [パートナー山田様] 共同プロジェクト提案 → 概算見積もりを求めている
3. [AWS] 費用アラート: 先月比+30% → 確認が必要
✅ **今日のタスク TOP 3**
1. 🔴 新機能実装(締め切り: 今日 17:00)
2. 🔴 ABCクライアントへ返信(締め切り: 今週金曜)
3. 🟠 AWS費用確認と予算調整
🎯 **今日の作戦**
10時〜12時の集中ブロックで新機能実装を完成させ、午後のミーティング前にABCクライアントへの返信草案を作成するのが最善策です。
コアモジュール④:夜間レポート自動生成
毎晩20時に1日の成果をまとめ、翌日の準備状況を確認するレポートを生成します。
# 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():
"""毎晩20時に実行する振り返りレポート"""
today = datetime.now(JST).strftime("%Y年%m月%d日")
request = f"""
{today}の終業振り返りレポートを作成してください。
1. タスクリストを取得し、「done」ステータスのタスク数と「todo」「in_progress」の残タスクを教える
2. 今日のカレンダーを取得し、完了した予定と明日の予定を整理する
3. 未処理の重要メールがあれば3件まで教える
最後に:
- 今日の達成率(完了タスク÷全タスク)
- 明日最初にやるべきことをひとつ
を教えてください。全体を150文字以内の「一言まとめ」で締めてください。
"""
result = run_secretary_session(request)
if SLACK_WEBHOOK_URL:
client = WebhookClient(SLACK_WEBHOOK_URL)
client.send(text=f"🌙 *{today} 終業レポート*\n\n{result}")
return result本番デプロイ:Cloud Run + Cloud Scheduler
ローカルで動作確認が取れたら、Cloud Run にデプロイして完全自動化を実現します。
# Dockerfile
FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
# Google OAuth トークンは Secret Manager から注入
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 のエントリポイント
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]}デプロイコマンド:
# Cloud Build & 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
# Cloud Scheduler でモーニングブリーフィングを毎朝8時に設定
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"コスト管理とモデル使い分け
このシステムをフル稼働させると月にどれくらいコストがかかるか試算してみます。
月間推定コスト(個人利用レベル)
モーニングブリーフィング(平日20日)の場合:1回あたり Gemini 2.5 Pro を約3,000トークン使用(入力)、約1,000トークン生成(出力)と想定します。
# コスト最適化:単純なタスクはFlashを使う
def run_simple_task(request: str) -> str:
"""軽量タスク(要約・分類など)はFlashで処理してコスト削減"""
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:
"""複雑な推論・判断が必要なタスクはProを使用"""
return run_secretary_session(request)重要度の低いメール要約には Flash を使い、タスクの優先付けや集中ブロックの計画には Pro を使う、という使い分けが費用対効果の高い運用方法です。
よくあるエラーと対処法
実装時によく遭遇するエラーとその解決策を整理します。
エラー1: google.auth.exceptions.RefreshError: Token has been expired or revoked.
OAuth トークンの有効期限切れです。以下のコードで自動リフレッシュを実装してください。
from google.oauth2.credentials import Credentials
from google.auth.transport.requests import Request
from google_auth_oauthlib.flow import InstalledAppFlow
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:
flow = InstalledAppFlow.from_client_secrets_file(
GOOGLE_CREDENTIALS_FILE, SCOPES
)
creds = flow.run_local_server(port=0)
with open(GOOGLE_TOKEN_FILE, "w") as token:
token.write(creds.to_json())
return credsエラー2: Function Calling ループが終わらない
Gemini が同じツールを何度も呼び出し続ける場合があります。max_iterations を設定し、かつシステムプロンプトで「同じツールを2回以上呼び出さないこと」と明示することで防げます。
エラー3: ResourceExhausted: 429 RESOURCE_EXHAUSTED
レート制限に引っかかっています。Gemini 2.5 Pro の RPM(分あたりリクエスト数)制限に注意し、指数バックオフを実装してください。
import time
import 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"⏳ レート制限。{delay:.1f}秒後にリトライ ({attempt+1}/{max_retries})")
time.sleep(delay)
else:
raiseまとめ
- Function Calling の設計が品質を決める: ツール定義の
descriptionとparametersを丁寧に書くほど、Gemini の判断精度が上がる - マルチターンループで複雑な自律処理が実現: ツール実行結果を Gemini に返し、完結するまでループを回す設計が重要
- コスト管理は用途別モデル使い分けで: Flash(軽量タスク)とPro(複雑推論)を明確に使い分ける
- Cloud Run + Scheduler で完全自動化: ローカルで動けば、そのまま本番化まで1時間以内で完了できる
今後の拡張として、Gemini の長文脈(100万トークン)を活かした過去1ヶ月のメール全件分析や、音声入力で秘書に話しかける Gemini Live API との連携なども面白い方向性です。
AI秘書が繰り返しの作業を引き受けてくれることで、あなたは本当に集中すべき創造的な仕事に時間とエネルギーを注ぐことができます。ぜひ週末の数時間で、最初のモーニングブリーフィングを動かしてみてください。
Function Calling の基礎から実装パターンを体系的に理解したい方には、Gemini API Function Calling 本番実装ガイドも合わせてご覧ください。また、CI/CDパイプラインへのAI組み込みをさらに深掘りしたい場合はGemini APIで業務自動化を本格運用するが参考になります。AI秘書システムを自律エージェントへと発展させる際には、Gemini API エージェント本番構築ガイドも参照してみてください。
実装パターンが豊富に解説されています。