プログラムによるメールの読み取りと送信
メッセージの一覧表示、本文の取得、MIMEメールの送信を学びます。
「プログラムによるメールの読み取りと送信」はCoddyKit上の無料AI Agentsレッスンです。 これはレッスン2/4です。 下記で完全なレッスンを無料で読むことができます。その後、ブラウザ内の組み込みコードエディタと24時間対応のAIチューターでハンズオン演習できます。 これはAI Agents学習パスの一部であり、ウェブとCoddyKitアプリ全体で進捗が同期されます。 AI Agentsコースには全4レッスンが含まれています。
Gmail APIでメッセージを一覧表示する
メールを読み取る最初の手順は、条件に一致するメッセージを一覧表示することです。service.users().messages().list()はメッセージIDとスレッドIDを返しますが、完全な内容は返しません。その後、各メッセージを個別に取得します。この2段階のパターンにより、一覧表示の呼び出しを高速に保てます。
def list_messages(service, user_id='me', query='', max_results=10):
results = service.users().messages().list(
userId=user_id,
q=query, # Gmail search query
maxResults=max_results
).execute()
messages = results.get('messages', [])
print(f'Found {len(messages)} messages')
return messages
# Examples of Gmail search queries:
# 'is:unread' — unread messages
# 'from:boss@company.com is:unread' — unread from boss
# 'subject:invoice label:inbox' — invoices in inbox
# 'after:2026/05/01 has:attachment' — recent with attachments
messages = list_messages(gmail_service, query='is:unread label:inbox')完全なメッセージの取得
service.users().messages().get()を使用して、メッセージの完全な内容を取得します。formatパラメータによって返されるデータ量を制御できます。'full'はヘッダーと本文を含み、'metadata'はヘッダーのみを返し、'minimal'はIDとラベルだけを返します。
def get_message(service, message_id, user_id='me'):
message = service.users().messages().get(
userId=user_id,
id=message_id,
format='full' # 'full', 'metadata', or 'minimal'
).execute()
return message
# Fetch the first unread message
messages = list_messages(gmail_service, query='is:unread', max_results=1)
if messages:
msg = get_message(gmail_service, messages[0]['id'])
print('Thread ID:', msg['threadId'])
print('Labels:', msg['labelIds'])
print('Snippet:', msg['snippet'][:100])メールヘッダーの抽出
ヘッダー（From、To、Subject、Date）は、message['payload']['headers']に{'name': ..., 'value': ...}形式の辞書のリストとして保存されています。名前を指定してヘッダーを抽出するヘルパーを作成してください。これは頻繁に使用することになります。
def get_header(message, name):
headers = message.get('payload', {}).get('headers', [])
for h in headers:
if h['name'].lower() == name.lower():
return h['value']
return ''
def extract_email_meta(message):
return {
'id': message['id'],
'from': get_header(message, 'From'),
'to': get_header(message, 'To'),
'subject': get_header(message, 'Subject'),
'date': get_header(message, 'Date'),
'snippet': message.get('snippet', '')
}
meta = extract_email_meta(msg)
print(f'From: {meta["from"]}')
print(f'Subject: {meta["subject"]}')
print(f'Date: {meta["date"]}')メール本文のデコード（base64）
Gmail APIのメール本文はbase64urlエンコードされています。これは、+が-に、/が_になる、URLで安全に使用できるbase64の変種です。デコードにはbase64.urlsafe_b64decode()を使用します。単純な（単一パート）メールとマルチパートメールの両方に対応してください。
import base64
def decode_body(data):
if not data:
return ''
decoded_bytes = base64.urlsafe_b64decode(data + '==')
return decoded_bytes.decode('utf-8', errors='replace')
def get_email_body(message):
payload = message.get('payload', {})
mime_type = payload.get('mimeType', '')
# Simple (non-multipart) email
if 'body' in payload and payload['body'].get('data'):
return decode_body(payload['body']['data'])
# Multipart email: find the text/plain or text/html part
parts = payload.get('parts', [])
for part in parts:
if part.get('mimeType') == 'text/plain':
return decode_body(part['body'].get('data', ''))
# Fallback: try text/html
for part in parts:
if part.get('mimeType') == 'text/html':
return decode_body(part['body'].get('data', ''))
return message.get('snippet', '')
# --- demo ---
encoded = base64.urlsafe_b64encode(b'Hello from the agent!').decode().rstrip('=')
print('Decoded body:', decode_body(encoded))
message = {
'payload': {
'mimeType': 'multipart/alternative',
'parts': [
{'mimeType': 'text/plain', 'body': {'data': encoded}}
]
}
}
print('Email body:', get_email_body(message))
マルチパートメールの再帰的な処理
添付ファイル、インライン画像、混在コンテンツを含む複雑なメールは、入れ子になったマルチパート構造になっています。本文のパートは任意の深さまで入れ子になる可能性があります。パートツリーをたどる再帰関数を使えば、すべてのケースに対応できます。
import base64
def extract_parts(payload, target_mime='text/plain'):
parts_text = []
mime_type = payload.get('mimeType', '')
if mime_type == target_mime:
data = payload.get('body', {}).get('data', '')
if data:
decoded = base64.urlsafe_b64decode(data + '==').decode('utf-8', errors='replace')
parts_text.append(decoded)
# Recurse into sub-parts
for part in payload.get('parts', []):
parts_text.extend(extract_parts(part, target_mime))
return parts_text
def get_plain_text(message):
payload = message.get('payload', {})
texts = extract_parts(payload, 'text/plain')
return '\n\n'.join(texts) if texts else message.get('snippet', '')
body_text = get_plain_text(msg)
print(f'Body ({len(body_text)} chars):', body_text[:200])メッセージを既読にする
メールの処理後、エージェントはUNREADラベルを削除して、そのメールを既読にする必要があります。removeLabelIds=['UNREAD']を指定してservice.users().messages().modify()を使用します。また、PROCESSEDなどのラベルを追加して、エージェントが処理したメールを追跡することもできます。
def mark_as_read(service, message_id, user_id='me'):
service.users().messages().modify(
userId=user_id,
id=message_id,
body={'removeLabelIds': ['UNREAD']}
).execute()
print(f'Marked {message_id} as read')
def add_label(service, message_id, label_id, user_id='me'):
service.users().messages().modify(
userId=user_id,
id=message_id,
body={'addLabelIds': [label_id]}
).execute()
# Get label ID by name
def get_label_id(service, label_name, user_id='me'):
labels = service.users().labels().list(userId=user_id).execute()
for label in labels.get('labels', []):
if label['name'].lower() == label_name.lower():
return label['id']
return None
# --- 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 messages(self):
return self
def modify(self, **kwargs):
print(f'[gmail api] messages.modify({kwargs})')
return _Exec({'id': kwargs.get('id')})
def labels(self):
return self
def list(self, **kwargs):
return _Exec({'labels': [{'id': 'Label_1', 'name': 'Processed'}]})
class _FakeService:
def users(self):
return _FakeUsers()
service = _FakeService()
mark_as_read(service, 'msg_42')
add_label(service, 'msg_42', 'Label_1')
print('Label id for "Processed":', get_label_id(service, 'Processed'))
MIMETextによるメールの作成
メールを送信するには、まずPython標準のemailライブラリを使用してMIMEメッセージとして作成します。次に、生のバイト列をbase64urlエンコードし、Gmail APIにPOSTします。MIMETextがメッセージ本文を適切にエンコードします。
import base64
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
def create_message(sender, to, subject, body_text, body_html=None):
if body_html:
msg = MIMEMultipart('alternative')
msg.attach(MIMEText(body_text, 'plain', 'utf-8'))
msg.attach(MIMEText(body_html, 'html', 'utf-8'))
else:
msg = MIMEText(body_text, 'plain', 'utf-8')
msg['From'] = sender
msg['To'] = to
msg['Subject'] = subject
# Encode as base64url
raw = base64.urlsafe_b64encode(msg.as_bytes()).decode('utf-8')
return {'raw': raw}
message = create_message(
sender='agent@yourcompany.com',
to='recipient@example.com',
subject='Weekly Summary',
body_text='Hello,\n\nHere is your summary.\n\nBest,\nAgent'
)
# --- demo ---
print('Message keys:', list(message.keys()))
print('Base64 length:', len(message['raw']))
Gmail APIによるメールの送信
service.users().messages().send()を使用して、作成したメッセージを送信します。userId='me'パラメータは認証済みユーザーを指します。APIは、IDとスレッドIDを含む送信済みメッセージを返します。
from googleapiclient.errors import HttpError
def send_message(service, message, user_id='me'):
try:
sent = service.users().messages().send(
userId=user_id,
body=message
).execute()
print(f'Message sent! ID: {sent["id"]}')
return sent
except HttpError as e:
import json
body = json.loads(e.content.decode())
print(f'Send failed ({e.resp.status}): {body.get("error", {}).get("message")}')
return None
# Send the message
message = create_message(
sender='me',
to='team@company.com',
subject='Agent Report',
body_text='Processing complete. 42 tasks handled.'
)
send_message(gmail_service, message)メールの下書き作成と送信
エージェントは、すぐに送信する代わりに、人間による確認用の下書きを作成できます。service.users().drafts().create()を使用します。その後、人間がGmailのUIから下書きを確認して送信できます。人間の承認が必要なメールには、この方法が推奨されます。
def create_draft(service, message, user_id='me'):
draft = service.users().drafts().create(
userId=user_id,
body={'message': message}
).execute()
print(f'Draft created: {draft["id"]}')
return draft
def send_draft(service, draft_id, user_id='me'):
sent = service.users().drafts().send(
userId=user_id,
body={'id': draft_id}
).execute()
print(f'Draft sent as message: {sent["id"]}')
return sent
# Create a draft for review
message = create_message(
sender='me',
to='client@example.com',
subject='Proposal Follow-up',
body_text='Dear Client,\n\nFollowing up on our proposal...'
)
draft = create_draft(gmail_service, message)
# Human reviews in Gmail, then agent sends:
# send_draft(gmail_service, draft['id'])複数のメールのバッチ処理
大量のメールを処理する場合、タイトなループで1通ずつ取得しないでください。クォータ制限に達してしまいます。短い遅延を入れた制御されたループを使用するか、Gmail APIのバッチリクエスト機能を使って複数の操作を1回のHTTP呼び出しにまとめてください。
import time
def process_unread_emails(service, max_emails=20):
messages = list_messages(
service,
query='is:unread label:inbox',
max_results=max_emails
)
processed = []
for i, msg_ref in enumerate(messages):
# Rate-limit: process max 5 per second
if i > 0 and i % 5 == 0:
time.sleep(1)
msg = get_message(service, msg_ref['id'])
meta = extract_email_meta(msg)
body = get_plain_text(msg)
result = {
'id': msg['id'],
'from': meta['from'],
'subject': meta['subject'],
'body_preview': body[:200]
}
processed.append(result)
mark_as_read(service, msg['id'])
return processedメールへの返信（スレッド内）
スレッド内で返信を送信するには、元のメッセージのMessage-IDヘッダーをIn-Reply-ToヘッダーとReferencesヘッダーに設定し、送信呼び出しにthreadIdを渡します。これにより、返信が同じGmailの会話スレッドに保持されます。
import base64
from email.mime.text import MIMEText
def create_reply(original_message, reply_text, sender='me'):
original_msg_id = get_header(original_message, 'Message-ID')
to = get_header(original_message, 'From')
subject = get_header(original_message, 'Subject')
if not subject.startswith('Re:'):
subject = 'Re: ' + subject
msg = MIMEText(reply_text, 'plain', 'utf-8')
msg['From'] = sender
msg['To'] = to
msg['Subject'] = subject
msg['In-Reply-To'] = original_msg_id
msg['References'] = original_msg_id
raw = base64.urlsafe_b64encode(msg.as_bytes()).decode('utf-8')
return {
'raw': raw,
'threadId': original_message['threadId'] # keeps it in thread
}クイックチェック：base64メール本文
Gmail APIのメッセージ処理についての理解度を確認しましょう。
メールの読み取りと送信のまとめ
これで、エージェントはプログラムからメールを読み取り、送信できるようになりました。
- 一覧表示：
messages().list(q='is:unread')はIDを返します。Gmailの検索構文を使用してください - 取得：
messages().get(id=..., format='full')は完全なメッセージを返します - ヘッダーの解析：
payload.headersからFrom/Subject/Dateを抽出します - 本文のデコード：テキストには
base64.urlsafe_b64decode(data)を使用し、マルチパートの各パートを再帰的に処理します - 送信：
MIMETextで作成し、base64urlエンコードしてmessages().send()経由でPOSTします - スレッド内で返信：
In-Reply-ToヘッダーとthreadIdを送信本文に設定します
よくある質問
「プログラムによるメールの読み取りと送信」レッスンは無料ですか？
はい。「プログラムによるメールの読み取りと送信」の完全なテキストはこのウェブで無料で読めます。インタラクティブに演習し（組み込みコードエディタと24時間対応のAIチューター）、AI Agentsコースの残りをアンロックするには、CoddyKit PROにアップグレードしてください。 AI Agentsコースには全4レッスンが含まれています。
「プログラムによるメールの読み取りと送信」で何を学びますか？
メッセージの一覧表示、本文の取得、MIMEメールの送信を学びます。 ブラウザで直接実行するハンズオンコードでAI Agentsを演習し、24時間対応のAIチューターがレッスンを進める中での質問に答えます。
AI Agentsを始めるのに経験は必要ですか？
事前経験は必要ありません。CoddyKitのAI Agentsは初級者から上級者向けに構成されているため、ここから始めるか最初から始めて、自分のペースで進むことができます。 これはレッスン2/4です。
「プログラムによるメールの読み取りと送信」レッスンにはどのくらい時間がかかりますか？
ほとんどのCoddyKitレッスンは約5～10分かかります。各レッスンはコンパクトでインタラクティブなので、着実に進歩し、ウェブとアプリ全体で正確に前回の場所から再開できます。
このAI Agentsレッスンでコードを書いて実行できますか？
はい。すべてのAI Agentsレッスンに組み込みコードエディタが含まれているため、ブラウザでリアルコードを書いて実行し、即座のAIフィードバックを取得できます。ローカル設定は不要です。
このコースのすべてのレッスン
- API経由でGmailに接続する
- プログラムによるメールの読み取りと送信
- カレンダーイベントの作成と検索
- シンプルなメールアシスタントエージェントの構築