Claude Code + Gmail API でメール自動化を実現する完全ガイド【2026年版】

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
from googleapiclient.errors import HttpError

# ================================
# 設定
# ================================
SCOPES = [“https://www.googleapis.com/auth/gmail.modify”]

logging.basicConfig(
level=logging.INFO,
format=“%(asctime)s [%(levelname)s] %(message)s”,
handlers=[
logging.FileHandler(“gmail_auto.log”, encoding=“utf-8”),
logging.StreamHandler()
]
)
log = logging.getLogger(__name__)

# ================================
# Gmail APIクライアントの初期化
# ================================
def get_gmail_service():
“””OAuth 2.0認証を行いGmail APIクライアントを返す”””
creds = None
if Path(“token.json”).exists():
creds = Credentials.from_authorized_user_file(“token.json”, 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(“credentials.json”, SCOPES)
creds = flow.run_local_server(port=0)
with open(“token.json”, “w”) as f:
f.write(creds.to_json())

service = build(“gmail”, “v1”, credentials=creds)
log.info(“Gmail API接続成功”)
return service

# ================================
# ヘルパー: ラベルIDの取得（なければ作成）
# ================================
def get_or_create_label(service, label_name: str) -> str:
“””ラベルIDを取得する。存在しない場合は新規作成する”””
labels_result = service.users().labels().list(userId=“me”).execute()
for label in labels_result.get(“labels”, []):
if label[“name”] == label_name:
return label[“id”]

# ラベルが存在しない場合は新規作成
new_label = service.users().labels().create(
userId=“me”,
body={“name”: label_name,
“labelListVisibility”: “labelShow”,
“messageListVisibility”: “show”}
).execute()
log.info(f”ラベル新規作成: {label_name}”)
return new_label[“id”]

# ================================
# 機能1: 自動返信
# ================================
def auto_reply_inquiries(service, config: dict) -> int:
“””お問い合わせメールを検出して自動返信する”””
# 検索クエリ: 未読 + 件名キーワード + 未返信ラベルなし
query = (
“is:unread”
‘ subject:(お問い合わせ OR 問い合わせ OR “contact us”)’
‘ -label:自動返信済み’
)
results = (service.users().messages()
.list(userId=“me”, q=query, maxResults=50)
.execute())
messages = results.get(“messages”, [])
log.info(f”自動返信対象: {len(messages)}件”)

replied_label_id = get_or_create_label(service, “自動返信済み”)
reply_count = 0
for msg_ref in messages:
try:
msg = (service.users().messages()
.get(userId=“me”, id=msg_ref[“id”], format=“full”)
.execute())
headers = {h[“name”]: h[“value”]
for h in msg[“payload”][“headers”]}
sender = headers.get(“From”, “”)
subject = headers.get(“Subject”, “”)
message_id = headers.get(“Message-ID”, “”)
thread_id = msg.get(“threadId”)

# 返信メールの作成
reply_subject = (f”Re: {subject}” if not subject.startswith(“Re:”)
else subject)
reply_body = config[“auto_reply”][“body”]

mime_msg = MIMEText(reply_body, “plain”, “utf-8”)
mime_msg[“From”] = “me”
mime_msg[“To”] = sender
mime_msg[“Subject”] = reply_subject
mime_msg[“In-Reply-To”] = message_id
mime_msg[“References”] = message_id

raw = base64.urlsafe_b64encode(
mime_msg.as_bytes()).decode()
(service.users().messages()
.send(userId=“me”,
body={“raw”: raw, “threadId”: thread_id})
.execute())

# 「自動返信済み」ラベルを付けて既読にする
service.users().messages().modify(
userId=“me”, id=msg_ref[“id”],
body={“addLabelIds”: [replied_label_id],
“removeLabelIds”: [“UNREAD”]}
).execute()

log.info(f”自動返信送信: {sender} / 件名: {subject}”)
reply_count += 1
except HttpError as e:
log.error(f”返信エラー: {e}”)

return reply_count

# ================================
# 機能2: ラベル自動分類
# ================================
LABEL_RULES = [
{“query”: “from:@example-company.com”, “label”: “取引先”, “star”: False},
{“query”: ‘subject:(請求書 OR Invoice)’, “label”: “請求書”, “star”: False},
{“query”: ‘subject:(緊急 OR URGENT OR “最優先”)’, “label”: “緊急”, “star”: True},
{“query”: ‘subject:(週次レポート OR 月次レポート)’, “label”: “レポート”,“star”: False},
]

def auto_label_emails(service) -> int:
“””ルールに基づいてメールを自動ラベル分類する”””
total_labeled = 0
for rule in LABEL_RULES:
label_id = get_or_create_label(service, rule[“label”])
query = f'{rule[“query”]} -label:{rule[“label”]}’

results = (service.users().messages()
.list(userId=“me”, q=query, maxResults=100)
.execute())
messages = results.get(“messages”, [])

if not messages:
continue
# バッチModifyでまとめてラベルを追加
ids = [m[“id”] for m in messages]
add_labels = [label_id]
if rule[“star”]:
add_labels.append(“STARRED”)

service.users().messages().batchModify(
userId=“me”,
body={“ids”: ids, “addLabelIds”: add_labels}
).execute()

log.info(f”ラベル「{rule[‘label’]}」を {len(ids)} 件に付与”)
total_labeled += len(ids)

return total_labeled

# ================================
# 機能3: 添付ファイル自動保存
# ================================
def save_pdf_attachments(service, save_dir: str = “attachments”) -> int:
“””PDFの添付ファイルを自動でフォルダに保存する”””
Path(save_dir).mkdir(exist_ok=True)
query = “has:attachment filename:pdf -label:添付保存済み”
results = (service.users().messages()
.list(userId=“me”, q=query, maxResults=50)
.execute())
messages = results.get(“messages”, [])
log.info(f”PDF添付あり: {len(messages)}件”)

saved_label_id = get_or_create_label(service, “添付保存済み”)
save_count = 0
for msg_ref in messages:
try:
msg = (service.users().messages()
.get(userId=“me”, id=msg_ref[“id”], format=“full”)
.execute())
headers = {h[“name”]: h[“value”]
for h in msg[“payload”][“headers”]}
sender = headers.get(“From”, “unknown”).split(“<“)[0].strip()
date_str = datetime.today().strftime(“%Y-%m-%d”)
safe_sender = “”.join(c for c in sender if c.isalnum() or c in ” _-“)

# MIMEパートを再帰的に探索してPDF添付を取得
def extract_attachments(payload):
if payload.get(“filename”) and payload[“filename”].lower().endswith(“.pdf”):
att_id = payload.get(“body”, {}).get(“attachmentId”)
if att_id:
att = (service.users().messages().attachments()
.get(userId=“me”, messageId=msg_ref[“id”], id=att_id)
.execute())
data = base64.urlsafe_b64decode(att[“data”])
safe_name = “”.join(c for c in payload[“filename”]
if c.isalnum() or c in ” ._-“)
file_path = P

STEP 2：Gmailのラベルを自動作成・管理する

from google.oauth2.credentials import Credentials
from googleapiclient.discovery import build

def setup_gmail_labels(service, label_configs):
    """Gmailラベルを一括作成（Claude Code生成）"""
    existing = {l['name']: l['id'] 
                for l in service.users().labels().list(userId='me').execute().get('labels', [])}
    
    created = []
    for config in label_configs:
        if config['name'] in existing:
            print(f"  既存: {config['name']}")
            continue
        label = service.users().labels().create(userId='me', body={
            'name': config['name'],
            'labelListVisibility': 'labelShow',
            'messageListVisibility': 'show',
            'color': config.get('color', {'backgroundColor': '#16a765', 'textColor': '#ffffff'})
        }).execute()
        created.append(label['name'])
        print(f"  作成: {label['name']}")
    
    return created

LABELS = [
    {'name': '★取引先', 'color': {'backgroundColor': '#4986e7', 'textColor': '#ffffff'}},
    {'name': '★社内', 'color': {'backgroundColor': '#16a765', 'textColor': '#ffffff'}},
    {'name': '★請求書', 'color': {'backgroundColor': '#e07798', 'textColor': '#ffffff'}},
    {'name': '★緊急対応', 'color': {'backgroundColor': '#cc3a21', 'textColor': '#ffffff'}},
    {'name': '自動返信済み'},
]
creds = Credentials.from_authorized_user_file('token.json')
service = build('gmail', 'v1', credentials=creds)
created = setup_gmail_labels(service, LABELS)
print(f"✅ {len(created)}個のラベルを作成しました")

STEP 3：添付ファイルを自動保存・整理する

import base64, os, re
from pathlib import Path

def save_attachments(service, max_results: int = 100):
    """添付ファイルを自動保存（Claude Code生成）"""
    results = service.users().messages().list(
        userId='me', q='has:attachment', maxResults=max_results
    ).execute()
    
    saved = 0
    for msg_ref in results.get('messages', []):
        msg = service.users().messages().get(
            userId='me', id=msg_ref['id'], format='full'
        ).execute()
        
        headers = {h['name']: h['value'] for h in msg['payload']['headers']}
        sender = headers.get('From', '')
        domain = re.search(r'@([\w.]+)', sender)
        domain_name = domain.group(1) if domain else 'unknown'
        
        for part in msg['payload'].get('parts', []):
            if part.get('filename') and part.get('body', {}).get('attachmentId'):
                filename = part['filename']
                ext = Path(filename).suffix.lower()
                if ext not in ['.pdf', '.xlsx', '.xls', '.docx', '.doc']:
                    continue
                
                save_dir = Path('attachments') / domain_name
                save_dir.mkdir(parents=True, exist_ok=True)
                
                attachment = service.users().messages().attachments().get(
                    userId='me', messageId=msg_ref['id'],
                    id=part['body']['attachmentId']
                ).execute()
                
                data = base64.urlsafe_b64decode(attachment['data'])
                save_path = save_dir / filename
                save_path.write_bytes(data)
                saved += 1
    
    print(f"✅ {saved}件の添付ファイルを保存しました")

save_attachments(service, max_results=50)

STEP 4：未返信メールを自動検出してSlackに通知する

from datetime import datetime, timedelta
import requests, os

def alert_unreplied(service, webhook_url: str, hours: int = 24):
    """未返信メールをSlackに通知（Claude Code生成）"""
    cutoff = int((datetime.now() - timedelta(hours=hours)).timestamp() * 1000)
    
    results = service.users().messages().list(
        userId='me',
        q=f'in:inbox is:unread -is:sent after:{int((datetime.now()-timedelta(hours=hours)).timestamp())}',
        maxResults=20
    ).execute()
    
    unreplied = []
    for msg_ref in results.get('messages', []):
        msg = service.users().messages().get(
            userId='me', id=msg_ref['id'], format='metadata',
            metadataHeaders=['From', 'Subject', 'Date']
        ).execute()
        
        headers = {h['name']: h['value'] for h in msg['payload']['headers']}
        unreplied.append({
            'from': headers.get('From', ''),
            'subject': headers.get('Subject', ''),
            'date': headers.get('Date', '')
        })
    
    if not unreplied:
        print("未返信メールなし"); return
    
    text = f"⚠️ *{hours}時間以上未返信のメール: {len(unreplied)}件*\n"
    for m in unreplied:
        text += f"• `{m['subject']}` - {m['from']}\n"
    
    requests.post(webhook_url, json={"text": text})
    print(f"✅ Slack通知送信: {len(unreplied)}件の未返信メール")

SLACK_WEBHOOK = os.environ.get('SLACK_WEBHOOK_URL', '')
alert_unreplied(service, SLACK_WEBHOOK, hours=24)

STEP 5：Gmailから日次レポートを自動生成する

from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
import base64, json
from collections import Counter

def send_daily_report(service):
    """Gmailの日次レポートを自動生成・送信"""
    today_query = f'after:{datetime.now().strftime("%Y/%m/%d")}'
    
    messages = service.users().messages().list(
        userId='me', q=today_query, maxResults=200
    ).execute()
    
    senders = []
    for msg_ref in messages.get('messages', [])[:50]:
        msg = service.users().messages().get(
            userId='me', id=msg_ref['id'], format='metadata',
            metadataHeaders=['From', 'Subject']
        ).execute()
        headers = {h['name']: h['value'] for h in msg['payload']['headers']}
        senders.append(headers.get('From', ''))
    
    sender_counts = Counter(senders).most_common(5)
    total = messages.get('resultSizeEstimate', 0)
    
    rows = ''.join(f"{s}{c}件" for s, c in sender_counts)
    html_body = f"""
    
📊 Gmail 日次レポート - {datetime.now().strftime('%Y/%m/%d')}
    
本日の受信件数: {total}件
    
送信者TOP5
    {rows}
    

    
送信者
件数
    
Claude Codeが自動生成したレポートです
    """
    
    msg = MIMEMultipart('alternative')
    msg['Subject'] = f"📧 Gmail日次レポート {datetime.now().strftime('%Y/%m/%d')}"
    msg['From'] = msg['To'] = 'me'
    msg.attach(MIMEText(html_body, 'html'))
    
    raw = base64.urlsafe_b64encode(msg.as_bytes()).decode()
    service.users().messages().send(userId='me', body={'raw': raw}).execute()
    print(f"✅ 日次レポート送信完了 (本日受信: {total}件)")

send_daily_report(service)

STEP 6：大量メールの一括アーカイブ・削除を自動化する

from datetime import datetime, timedelta

def bulk_archive_old_emails(service, days: int = 90, dry_run: bool = True):
    """古いメールを一括アーカイブ（Claude Code生成）"""
    cutoff = (datetime.now() - timedelta(days=days)).strftime("%Y/%m/%d")
    query = f'in:inbox before:{cutoff} -has:userlabels -is:starred'
    
    archived = 0
    page_token = None
    
    while True:
        params = {'userId': 'me', 'q': query, 'maxResults': 100}
        if page_token:
            params['pageToken'] = page_token
        
        results = service.users().messages().list(**params).execute()
        messages = results.get('messages', [])
        
        if not messages:
            break
        
        if not dry_run:
            msg_ids = [m['id'] for m in messages]
            service.users().messages().batchModify(
                userId='me',
                body={
                    'ids': msg_ids,
                    'removeLabelIds': ['INBOX'],
                    'addLabelIds': []
                }
            ).execute()
        
        archived += len(messages)
        page_token = results.get('nextPageToken')
        if not page_token:
            break
        print(f"  処理済み: {archived}件...")
    
    action = "アーカイブ予定" if dry_run else "アーカイブ完了"
    print(f"✅ {action}: {archived}件 ({days}日以上前の未タグメール)")
    return archived

# まずドライランで確認
count = bulk_archive_old_emails(service, days=90, dry_run=True)
print(f"確認: {count}件がアーカイブ対象")
# 実際に実行する場合: bulk_archive_old_emails(service, days=90, dry_run=False)