メール転送→freee自動仕訳の仕組みを作った
— AWS Lambda + Claude Vision + freee API 実装まとめ

全体アーキテクチャ

一気通貫のフローはこうなっている。

[仕入先] メール送信
    ↓
AWS SES (in.tanomeru.ai)  Receipt Rule
    ↓ S3 保存 + Lambda 非同期トリガー
AWS S3  raw/{message_id}  (MIME フルメール)
    ↓ Lambda process_invoice.py が読み取り
Claude Vision API  (claude-3-5-sonnet)
    ↓ PDF/PNG/JPEG → JSON (仕入先名/金額/税額/発行日)
freee API  POST /api/1/deals
    ↓ 取引 (仕訳) 登録完了
Supabase  invoices_raw + invoices_structured 保存
    ↓
SES  処理完了メール をユーザーに送信

メールが届いてから freee への仕訳登録完了まで、おおよそ 30〜60 秒。 AWS Lambda は非同期起動なのでメール送信者を待たせない設計になっている。 処理が終わると SES でユーザーに完了通知メールが届く。

Step 1 — SES でメールを受け取り、S3 に保存する

@in.tanomeru.ai サブドメイン全体を AWS SES の Receipt Rule で受け取る。受け取ったメールは MIME ごと S3 に保存し、 Lambda を非同期で起動する。

TLS 必須 (TlsPolicy: Require) にすることで、 転送時の平文通信を防いでいる。スパムスキャンも有効にして Lambda 側の処理コストを削減。

# template.yaml (AWS SAM) — SES Receipt Rule
SESReceiptRule:
  Type: AWS::SES::ReceiptRule
  Properties:
    RuleSetName: !Ref SESRuleSetName
    Rule:
      Name: tanomeru-invoice-rule
      Enabled: true
      ScanEnabled: true          # スパムスキャン有効
      TlsPolicy: Require         # TLS 必須
      Recipients:
        - "@in.tanomeru.ai"      # サブドメイン全体を受け取る
      Actions:
        - S3Action:
            BucketName: !Ref RawEmailBucket
            ObjectKeyPrefix: "raw/"
        - LambdaAction:
            FunctionArn: !GetAtt ProcessInvoiceFunction.Arn
            InvocationType: Event  # 非同期起動

Lambda ハンドラは S3 からメール本体を取得し、MIME パースして添付ファイルを抽出する。 宛先アドレス (user-uuid@in.tanomeru.ai) から ユーザーを特定する仕組みだ。

# lambda/process_invoice.py — メール受信ハンドラ
import boto3, email, os
from email import policy

s3 = boto3.client('s3')

def handler(event, context):
    record = event['Records'][0]['ses']
    message_id = record['mail']['messageId']

    # S3 からメール本体を取得
    obj = s3.get_object(
        Bucket=os.environ['RAW_EMAIL_BUCKET'],
        Key=f"raw/{message_id}"
    )
    raw_email = obj['Body'].read()

    # MIME パース → 添付ファイル (PDF/PNG/JPEG) 抽出
    msg = email.message_from_bytes(raw_email, policy=policy.default)
    attachments = extract_attachments(msg)

    # 宛先アドレスからユーザー UUID を特定
    recipient = record['receipt']['recipients'][0]
    user_uuid = recipient.split('@')[0]

    return parse_and_register(user_uuid, attachments, message_id)

Step 2 — Claude Vision で請求書をパースする

添付された PDF / PNG / JPEG を Claude Vision (claude-3-5-sonnet-20241022) に 投げて、構造化 JSON を返させる。プロンプトにスキーマを埋め込む設計を採用した。

β 改善点: α 版では Claude が JSON を コードブロック (```json ... ```) で返すケースがあり、 パース失敗が起きていた。正規表現 fallback を追加して解消した。

# lambda/claude_parser.py
import anthropic, base64, json, re

client = anthropic.Anthropic()

INVOICE_SCHEMA = """
{
  "vendor_name":    "str  — 請求元の法人名・屋号",
  "issue_date":     "str  — YYYY-MM-DD",
  "due_date":       "str | null",
  "total_amount":   "int  — 税込合計 (円)",
  "tax_amount":     "int  — 消費税額 (円)",
  "tax_rate":       "float — 0.10 or 0.08",
  "invoice_number": "str | null — 適格請求書番号 (T始まり13桁)",
  "items": [{"description": "str", "amount": "int"}]
}
"""

def parse_invoice(file_bytes: bytes, media_type: str) -> dict:
    b64 = base64.standard_b64encode(file_bytes).decode()

    response = client.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=1024,
        messages=[{
            "role": "user",
            "content": [
                {
                    "type": "image",
                    "source": {
                        "type": "base64",
                        "media_type": media_type,   # image/png / image/jpeg
                        "data": b64,
                    },
                },
                {
                    "type": "text",
                    "text": (
                        "この請求書から以下のスキーマで情報を抽出してください。"
                        "JSON 形式のみで返答してください.\n"
                        + INVOICE_SCHEMA
                    ),
                },
            ],
        }],
    )

    text = response.content[0].text
    # β 改善: コードブロック形式で返ってくる場合の fallback
    match = re.search(r'\{.*\}', text, re.DOTALL)
    return json.loads(match.group() if match else text)

invoice_number には 適格請求書番号 (T始まり 13 桁のインボイス番号) を取得させている。 freee 側の税区分 (tax_code) は 10%/8% の税率から自動マッピングする。

Step 3 — freee API に仕訳を登録する

パース結果を freee API の POST /api/1/deals で 支出取引として登録する。勘定科目は「仕入高 (ID: 59)」をデフォルトにしているが、 ユーザーが freee 上で後から変更できる設計だ。

β 改善点 (トークン URL 修正): α 版では accounts.freee.co.jp (legacy) を使っていた。 freee 公式推奨の accounts.secure.freee.co.jp に統一することで、 token refresh が確実に動くようになった。

# lambda/freee_client.py
import requests

FREEE_API_BASE = "https://api.freee.co.jp"
# β 改善: accounts.secure.freee.co.jp に統一 (旧 accounts.freee.co.jp は legacy)
FREEE_TOKEN_URL = "https://accounts.secure.freee.co.jp/public_api/token"

def register_deal(access_token: str, company_id: int, invoice: dict) -> dict:
    """freee に支出取引 (仕訳) を登録する"""
    payload = {
        "company_id": company_id,
        "issue_date":  invoice["issue_date"],
        "due_date":    invoice.get("due_date"),
        "type":        "expense",
        "partner_name": invoice["vendor_name"],
        "ref_number":  invoice.get("invoice_number"),
        "details": [
            {
                "account_item_id": 59,  # 仕入高 (ユーザーが後から変更可)
                "tax_code": 1 if invoice["tax_rate"] == 0.10 else 5,
                "amount":   invoice["total_amount"],
                "description": item.get("description",
                                f"{invoice['vendor_name']}からの請求"),
            }
            for item in invoice.get("items", [{}])
        ],
    }
    resp = requests.post(
        f"{FREEE_API_BASE}/api/1/deals",
        json=payload,
        headers={
            "Authorization":  f"Bearer {access_token}",
            "Content-Type":   "application/json",
        },
    )
    resp.raise_for_status()
    return resp.json()

access_token は有効期限があるため、expires_at を チェックして 5 分バッファでプロアクティブにリフレッシュし、401 時は 1 回リトライする二段構えにした。

処理結果を Supabase に保存する

処理結果は Supabase (PostgreSQL) の 2 テーブルに書き込む。invoices_raw は 処理ステータス管理、invoices_structured は freee に登録した仕訳データの保存用だ。ユーザーが /receipts ページで 処理履歴を確認できるようにしている。

# 処理結果を Supabase に保存
from datetime import datetime

def save_result(supabase, user_id, message_id, invoice, deal_id, status):
    # invoices_raw: 処理ステータス更新
    supabase.table("invoices_raw").update({
        "processing_status": status,          # completed / failed / pending
        "processed_at": datetime.utcnow().isoformat(),
    }).eq("message_id", message_id).execute()

    if deal_id:
        # invoices_structured: freee 登録済みデータを保存
        supabase.table("invoices_structured").insert({
            "user_id":       user_id,
            "message_id":    message_id,
            "freee_deal_id": deal_id,
            "vendor_name":   invoice["vendor_name"],
            "total_amount":  invoice["total_amount"],
            "issue_date":    invoice["issue_date"],
        }).execute()

β で追加した「処理完了 / 失敗通知」

α 版では処理が成功しても失敗しても、ユーザーには何も通知されなかった。 freee を開いて仕訳が入っていることを確認するしかなかった — これは UX として致命的だ。

β では SES の send_email で 処理完了メールと失敗時のエラー通知メールを送るようにした。 エラーは 4 カテゴリに分類して、ユーザーに「何が起きたか」を伝える:

• A: 添付ファイルなし
• B: AI による請求書認識失敗
• C: freee API エラー (認証切れ・権限不足)
• D: その他の内部エラー

# lambda/notifier.py — SES で処理完了 / 失敗メールを送信
import boto3, os

ses = boto3.client('ses', region_name='ap-northeast-1')

def notify(sender_email: str, success: bool, vendor_name: str = '', error_category: str = ''):
    subject = (
        f"[tanomeru] {vendor_name} の請求書を仕訳しました ✅"
        if success else
        "[tanomeru] 請求書の処理に失敗しました ⚠️"
    )
    body = (
        f"freeeに仕訳を登録しました。\n取引先: {vendor_name}"
        if success else
        f"処理中にエラーが発生しました (カテゴリ: {error_category})。\nfreeeから手動で登録してください。"
    )
    ses.send_email(
        Source=os.environ['NOTIFICATION_FROM_EMAIL'],  # noreply@in.tanomeru.ai
        Destination={'ToAddresses': [sender_email]},
        Message={
            'Subject': {'Data': subject, 'Charset': 'UTF-8'},
            'Body':    {'Text': {'Data': body,    'Charset': 'UTF-8'}},
        },
    )

α → β で改善したこと (まとめ)

まとめ

AWS SES → S3 → Lambda → Claude Vision → freee API という構成で、 「メール転送 1 回で freee に自動仕訳」を個人開発で実現した。 スタック全体で新規サービスはゼロ — すべて既存の managed service を組み合わせるだけで作れた。

α から β への改善は「動く」から「信頼できる」へのシフトだった。 エラー監視・完了通知・token 自動リフレッシュ。ユーザーに「何も起きていない」と感じさせない 設計が大事だと改めて思った。

コードの詳細や質問は X (@tanomeru_ai) で気軽にどうぞ。 Zenn / Qiita への cross-post も予定している。