シンプルなメールアシスタントエージェントの構築
一連の流れを学びます：受信トレイを読む → 要約する → 返信案を作成する → 承認を待つ。
「シンプルなメールアシスタントエージェントの構築」はCoddyKit上の無料AI Agentsレッスンです。 これはレッスン4/4です。 下記で完全なレッスンを無料で読むことができます。その後、ブラウザ内の組み込みコードエディタと24時間対応のAIチューターでハンズオン演習できます。 これはAI Agents学習パスの一部であり、ウェブとCoddyKitアプリ全体で進捗が同期されます。 AI Agentsコースには全4レッスンが含まれています。
メールアシスタントエージェントのアーキテクチャ
メールアシスタントエージェントは、標準的なパイプライン取得 → 分類 → 判断 → 下書き作成 → 承認 → 送信に従います。エージェントは未読メールを読み取り、LLMを使って分類と返信の下書きを作成し、人による承認を待ってから送信します。このHuman-in-the-loop設計により、メールを完全に自律送信することで発生する重大なミスを防げます。
# Email Assistant Pipeline:
#
# 1. FETCH: Pull unread emails from Gmail API
# 2. CLASSIFY: LLM labels each email
# - 'action_needed': requires a reply
# - 'fyi': informational, no reply needed
# - 'spam': should be archived
# 3. DRAFT: LLM generates reply for 'action_needed' emails
# 4. APPROVE: Human reviews drafts in Gmail UI
# 5. SEND: Agent sends approved drafts
#
# Tools: Gmail API, Anthropic/OpenAI API, json, base64
print('Email assistant pipeline: fetch -> classify -> draft -> approve -> send')エージェントツールの定義
明確なツール定義でエージェントを構成します。各ツールは、特定の責務を持つPython関数です。このモジュール性により、エージェントのテストやデバッグが容易になり、新しい機能も簡単に追加できます。
class EmailAssistantTools:
def __init__(self, gmail_service, llm_client):
self.gmail = gmail_service
self.llm = llm_client
def fetch_unread(self, max_emails=10):
'''Fetch unread emails from inbox.'''
pass
def classify_email(self, subject, body, sender):
'''Ask LLM to classify: action_needed / fyi / spam.'''
pass
def draft_reply(self, subject, body, sender, context):
'''Ask LLM to draft a reply to an email.'''
pass
def create_draft(self, message_id, reply_text):
'''Save draft reply in Gmail for human review.'''
pass
def send_approved_drafts(self):
'''Send all drafts marked as approved.'''
pass
print('Tool-based architecture enables testing each step independently')ステップ1: メールの取得と解析
最初のツールで未読メールを取得し、LLMに必要な主要フィールドである件名、送信者、プレーンテキストの本文を抽出します。LLMのコンテキスト制限内に収め、APIコストを抑えるため、本文は短く保ち、先頭の2000文字までに切り詰めてください。
import base64
def fetch_emails_for_classification(gmail_service, max_emails=10):
messages_list = gmail_service.users().messages().list(
userId='me',
q='is:unread label:inbox',
maxResults=max_emails
).execute().get('messages', [])
emails = []
for ref in messages_list:
msg = gmail_service.users().messages().get(
userId='me', id=ref['id'], format='full'
).execute()
headers = {h['name'].lower(): h['value']
for h in msg['payload'].get('headers', [])}
body = extract_plain_text_body(msg)[:2000] # trim for LLM
emails.append({
'id': msg['id'],
'thread_id': msg['threadId'],
'from': headers.get('from', ''),
'subject': headers.get('subject', '(no subject)'),
'body': body
})
print(f'Fetched {len(emails)} unread emails')
return emailsステップ2: LLMによる分類
各メールをLLMに送信して分類します。モデルにカテゴリと理由を含むJSONオブジェクトを返すよう求める、構造化されたプロンプトを使用してください。自由形式のテキストを解析するよりも信頼性が高いため、JSON出力を明示的に要求します。
import json
import anthropic
import os
client = anthropic.Anthropic(api_key=os.environ['ANTHROPIC_API_KEY'])
def classify_email(subject, sender, body):
prompt = (
'Classify this email as exactly one of: action_needed, fyi, spam.\n'
'Return JSON only: {"category": "...", "reason": "..."}\n\n'
f'From: {sender}\n'
f'Subject: {subject}\n\n'
f'Body:\n{body[:1500]}'
)
response = client.messages.create(
model='claude-opus-4-5',
max_tokens=200,
messages=[{'role': 'user', 'content': prompt}]
)
try:
result = json.loads(response.content[0].text)
return result.get('category', 'fyi'), result.get('reason', '')
except json.JSONDecodeError:
return 'fyi', 'Could not parse LLM response'ステップ3: 返信下書きの生成
action_neededに分類されたメールについては、LLMに返信の下書きを生成させます。エージェントの役割と文体に関するコンテキストを与えてください。プロフェッショナルで簡潔な返信を求め、人が記入する必要がある箇所にはプレースホルダーを含めます。
def draft_reply(subject, sender, body, agent_context):
prompt = (
'You are an email assistant drafting a professional reply.\n'
'Guidelines:\n'
'- Be concise and professional\n'
'- Use [FILL IN] for any info you don\'t know\n'
'- Start with a greeting, end with a sign-off\n\n'
f'Context about the recipient\'s work: {agent_context}\n\n'
f'Original email from {sender}:\n'
f'Subject: {subject}\n\n'
f'{body[:1500]}\n\n'
'Draft a reply:'
)
response = client.messages.create(
model='claude-opus-4-5',
max_tokens=500,
messages=[{'role': 'user', 'content': prompt}]
)
return response.content[0].text.strip()ステップ4: 人による確認のための下書き保存
AIが生成したメールを、人による確認なしに送信してはいけません。Gmailの下書きとして保存すれば、人がGmailを開いて確認し、必要に応じて編集してから送信できます。下書き作成APIは送信APIと同じで、drafts().create()を使用する点だけが異なります。
import base64
from email.mime.text import MIMEText
def save_draft_reply(gmail_service, original_message, reply_text):
sender_header = next(
(h['value'] for h in original_message['payload'].get('headers', [])
if h['name'].lower() == 'from'), ''
)
subject = next(
(h['value'] for h in original_message['payload'].get('headers', [])
if h['name'].lower() == 'subject'), ''
)
msg_id_header = next(
(h['value'] for h in original_message['payload'].get('headers', [])
if h['name'].lower() == 'message-id'), ''
)
mime_msg = MIMEText(reply_text, 'plain', 'utf-8')
mime_msg['To'] = sender_header
mime_msg['Subject'] = 'Re: ' + subject
mime_msg['In-Reply-To'] = msg_id_header
mime_msg['References'] = msg_id_header
raw = base64.urlsafe_b64encode(mime_msg.as_bytes()).decode()
draft = gmail_service.users().drafts().create(
userId='me',
body={'message': {'raw': raw, 'threadId': original_message['threadId']}}
).execute()
print(f'Draft saved: {draft["id"]}')
return draft['id']
# --- demo: minimal stand-in for the Gmail API's service object ---
class _Exec:
def __init__(self, result):
self._result = result
def execute(self):
return self._result
class _FakeDrafts:
def create(self, userId, body):
return _Exec({'id': 'r9000abc'})
class _FakeUsers:
def drafts(self):
return _FakeDrafts()
class _FakeGmailService:
def users(self):
return _FakeUsers()
original_message = {
'threadId': 'thread_1',
'payload': {'headers': [
{'name': 'From', 'value': 'customer@example.com'},
{'name': 'Subject', 'value': 'Question about my order'},
{'name': 'Message-ID', 'value': '<abc123@mail.example.com>'}
]}
}
save_draft_reply(_FakeGmailService(), original_message, 'Thanks for reaching out, we will look into it.')
ステップ5: 処理済みメールのマーク
メールの処理（分類と下書き作成、またはアーカイブ）が完了したら、再処理を防ぐためにマークを付けます。AgentProcessedのようなカスタムラベルを追加し、UNREADラベルを削除します。ラベルが存在しない場合は、最初に一度だけ作成してください。
def get_or_create_label(gmail_service, label_name):
labels = gmail_service.users().labels().list(userId='me').execute()
for label in labels.get('labels', []):
if label['name'] == label_name:
return label['id']
# Create the label
new_label = gmail_service.users().labels().create(
userId='me',
body={
'name': label_name,
'labelListVisibility': 'labelShow',
'messageListVisibility': 'show'
}
).execute()
print(f'Created label: {label_name}')
return new_label['id']
def mark_processed(gmail_service, message_id, agent_label_id):
gmail_service.users().messages().modify(
userId='me',
id=message_id,
body={
'addLabelIds': [agent_label_id],
'removeLabelIds': ['UNREAD']
}
).execute()
# --- demo: minimal stand-in for the Gmail API's service object ---
class _Exec:
def __init__(self, result):
self._result = result
def execute(self):
return self._result
class _FakeUsers:
def __init__(self):
self._labels = [{'id': 'Label_1', 'name': 'Processed'}]
def labels(self):
return self
def list(self, userId):
return _Exec({'labels': self._labels})
def create(self, userId, body):
print(f'Created label: {body["name"]}')
return _Exec({'id': 'Label_2', 'name': body['name']})
def messages(self):
return self
def modify(self, userId, id, body):
print(f'Marked {id} processed with {body}')
return _Exec({'id': id})
class _FakeGmailService:
def users(self):
return _FakeUsers()
gmail_service = _FakeGmailService()
label_id = get_or_create_label(gmail_service, 'Processed')
print(f'Label id: {label_id}')
new_label_id = get_or_create_label(gmail_service, 'AgentHandled')
mark_processed(gmail_service, 'msg_99', new_label_id)
エージェントの実行ループ全体
すべてのステップを1つのrun()メソッドにまとめます。エージェントはメールを順に処理し、各メールを分類し、対応が必要なメールには下書きを作成し、スパムをアーカイブして、すべてのメールを処理済みとしてマークします。最後に概要をログに記録します。
def run_email_agent(gmail_service, agent_context, max_emails=10):
processed_label = get_or_create_label(gmail_service, 'AgentProcessed')
emails = fetch_emails_for_classification(gmail_service, max_emails)
summary = {'action_needed': 0, 'fyi': 0, 'spam': 0, 'drafts_created': 0}
for email in emails:
category, reason = classify_email(
email['subject'], email['from'], email['body']
)
summary[category] += 1
print(f'[{category}] {email["subject"][:60]} - {reason[:50]}')
if category == 'action_needed':
# Fetch full message for reply context
full_msg = gmail_service.users().messages().get(
userId='me', id=email['id'], format='full'
).execute()
reply = draft_reply(
email['subject'], email['from'],
email['body'], agent_context
)
save_draft_reply(gmail_service, full_msg, reply)
summary['drafts_created'] += 1
mark_processed(gmail_service, email['id'], processed_label)
print('\nAgent run complete:')
for k, v in summary.items():
print(f' {k}: {v}')
return summaryエラー処理と耐障害性
各メールの処理ステップをtry/exceptで囲みます。1通のメールで発生したエラーによって、残りのメールの処理が止まってはいけません。後で確認できるよう、メールIDと件名を含めてエラーをログに記録します。
from googleapiclient.errors import HttpError
def process_email_safely(gmail_service, email, agent_context, label_id):
try:
category, reason = classify_email(
email['subject'], email['from'], email['body']
)
if category == 'action_needed':
full_msg = gmail_service.users().messages().get(
userId='me', id=email['id'], format='full'
).execute()
reply = draft_reply(
email['subject'], email['from'],
email['body'], agent_context
)
save_draft_reply(gmail_service, full_msg, reply)
mark_processed(gmail_service, email['id'], label_id)
return category
except HttpError as e:
print(f'Gmail API error on {email["id"]}: {e.resp.status}')
return 'error'
except Exception as e:
print(f'Unexpected error on {email["subject"][:50]}: {e}')
return 'error'エージェントの定期実行のスケジュール設定
time.sleep()で実行間隔を空ける単純なループを使って、メールエージェントをスケジュール実行します。本番環境では、cron、APScheduler、クラウド関数などのタスクスケジューラを使用してください。受信者のタイムゾーンに配慮し、営業時間外に実行されないよう必ずチェックを追加します。
import time
import datetime
def is_business_hours():
now = datetime.datetime.now()
# Mon-Fri, 9am-6pm local time
return (now.weekday() < 5 and 9 <= now.hour < 18)
def run_scheduler(gmail_service, agent_context,
interval_minutes=30, max_emails=20):
print(f'Email agent started. Checking every {interval_minutes} min.')
while True:
if is_business_hours():
print(f'\n[{datetime.datetime.now().strftime("%H:%M")}] Running agent...')
try:
run_email_agent(gmail_service, agent_context, max_emails)
except Exception as e:
print(f'Agent run failed: {e}')
else:
print('Outside business hours, skipping run')
time.sleep(interval_minutes * 60)ロギングと監査証跡
エージェントが処理したすべてのメールをJSONログに記録します。これにより、エージェントの動作を確認したり、分類ミスを診断したり、時間の経過とともにプロンプトを改善したりできる監査証跡が作成されます。
import json
import datetime
from pathlib import Path
LOG_FILE = Path('agent_audit.jsonl')
def log_action(email_id, subject, sender, category, action, draft_id=None):
entry = {
'timestamp': datetime.datetime.now().isoformat(),
'email_id': email_id,
'subject': subject[:100],
'from': sender,
'category': category,
'action': action,
'draft_id': draft_id
}
with open(LOG_FILE, 'a', encoding='utf-8') as f:
f.write(json.dumps(entry) + '\n')
# Usage in the main loop:
log_action(
email_id='18abc123',
subject='Invoice #1234',
sender='billing@vendor.com',
category='action_needed',
action='draft_created',
draft_id='r9000abc'
)
# --- demo ---
print('Audit log contents:')
print(LOG_FILE.read_text(encoding='utf-8'))
クイックチェック: Human-in-the-Loop
メールアシスタントの設計についての理解度を確認しましょう。
メールアシスタントエージェントのまとめ
これで、メールアシスタントエージェントのパイプライン全体を構築できました。
- 取得: Gmail APIの検索クエリで未読メールを取得
- 分類: LLMが各メールをaction_needed/fyi/spamに分類し、JSONを出力
- 下書き作成: LLMがプロフェッショナルな文体で返信を生成し、[FILL IN]プレースホルダーを含める
- 下書き保存: 人による確認のためGmailに保存し、自動送信は行わない
- 処理済みとしてマーク: カスタムラベルを追加し、再処理を防ぐためUNREADを削除
- エラーの分離: 各メールをtry/exceptで囲み、1件の失敗でバッチ全体が停止しないようにする
- 監査ログ: すべての操作を記録し、確認と改善に使用するJSONLファイル
よくある質問
「シンプルなメールアシスタントエージェントの構築」レッスンは無料ですか？
はい。「シンプルなメールアシスタントエージェントの構築」の完全なテキストはこのウェブで無料で読めます。インタラクティブに演習し（組み込みコードエディタと24時間対応のAIチューター）、AI Agentsコースの残りをアンロックするには、CoddyKit PROにアップグレードしてください。 AI Agentsコースには全4レッスンが含まれています。
「シンプルなメールアシスタントエージェントの構築」で何を学びますか？
一連の流れを学びます：受信トレイを読む → 要約する → 返信案を作成する → 承認を待つ。 ブラウザで直接実行するハンズオンコードでAI Agentsを演習し、24時間対応のAIチューターがレッスンを進める中での質問に答えます。
AI Agentsを始めるのに経験は必要ですか？
事前経験は必要ありません。CoddyKitのAI Agentsは初級者から上級者向けに構成されているため、ここから始めるか最初から始めて、自分のペースで進むことができます。 これはレッスン4/4です。
「シンプルなメールアシスタントエージェントの構築」レッスンにはどのくらい時間がかかりますか？
ほとんどのCoddyKitレッスンは約5～10分かかります。各レッスンはコンパクトでインタラクティブなので、着実に進歩し、ウェブとアプリ全体で正確に前回の場所から再開できます。
このAI Agentsレッスンでコードを書いて実行できますか？
はい。すべてのAI Agentsレッスンに組み込みコードエディタが含まれているため、ブラウザでリアルコードを書いて実行し、即座のAIフィードバックを取得できます。ローカル設定は不要です。
このコースのすべてのレッスン
- API経由でGmailに接続する
- プログラムによるメールの読み取りと送信
- カレンダーイベントの作成と検索
- シンプルなメールアシスタントエージェントの構築