メール配信で避けて通れないのがバウンス(配信エラー)の管理です。存在しないアドレスに送り続けると送信元の評価が下がり、正常なアドレスへの到達率まで悪化します。
blastengineにはエラー停止の仕組みが組み込まれており、APIでエラー停止リストを取得・管理できます。本記事では、このAPIを活用したバウンス管理の実装方法を解説します。
blastengineのエラー停止の仕組み
blastengineでは以下のルールでアドレスが自動的にエラー停止されます。
- 条件: 1つの宛先に対し、2週間以内に2回以上のハードエラーが発生
- 停止期間: 最大2週間
-
停止中の挙動: 該当アドレスへの配信はドロップされる(レスポンスコード
554)
つまり、blastengine側で自動的にバウンスアドレスへの配信を止めてくれます。ただし、停止期間が過ぎると再び配信対象になるため、自前の配信リストからも除外する運用が重要です。
エラー停止リストの取得(3ステップ)
エラー停止リストの取得はジョブ形式で非同期に行います。
Step 1: ジョブの開始
POST /errors/list でCSV生成ジョブを開始します。
curl -X POST https://app.engn.jp/api/v1/errors/list \
-H "Authorization: Bearer ${TOKEN}" \
-H "Content-Type: application/json" \
-d '{
"error_start": "2026-02-01T00:00:00+09:00",
"error_end": "2026-02-28T23:59:59+09:00"
}'
{
"job_id": 789
}
フィルタ条件として以下が使えます。
| パラメータ | 説明 |
|---|---|
error_start |
エラー停止日時の開始日 |
error_end |
エラー停止日時の終了日 |
email |
特定アドレスで絞り込み(完全一致) |
response_code |
レスポンスコードで絞り込み |
Step 2: ジョブの進捗確認
GET /errors/list/{job_id} でジョブの完了を待ちます。
curl https://app.engn.jp/api/v1/errors/list/789 \
-H "Authorization: Bearer ${TOKEN}"
{
"percentage": 100,
"status": "FINISHED",
"total_count": 45,
"error_file_url": "https://app.engn.jp/api/v1/errors/xxx/xxx/xxx"
}
status が FINISHED になったら次のステップへ進みます。
Step 3: CSVのダウンロード
GET /errors/list/{job_id}/download でエラー停止リストのCSV(zip形式)をダウンロードします。
curl -o error_list.zip \
https://app.engn.jp/api/v1/errors/list/789/download \
-H "Authorization: Bearer ${TOKEN}"
unzip error_list.zip
CSVの内容
| 配信停止番号 | 登録日時 | メールアドレス | 応答コード | エラーメッセージ |
|---|---|---|---|---|
| 1 | 2026-02-15 10:30:00 | invalid@example.jp | 550 | 宛先のメールアドレスがありません |
| 2 | 2026-02-16 14:20:00 | full@example.jp | 552 | メールボックスがいっぱいです |
Pythonで自動化する
エラー停止リストの取得からリストクリーニングまでを自動化する例です。
import requests
import time
import zipfile
import csv
import io
BASE = "https://app.engn.jp/api/v1"
TOKEN = "YOUR_BEARER_TOKEN"
HEADERS = {"Authorization": f"Bearer {TOKEN}"}
def get_error_list(error_start, error_end):
"""エラー停止リストを取得してアドレス一覧を返す"""
# 1. ジョブ開始
res = requests.post(f"{BASE}/errors/list", headers={**HEADERS, "Content-Type": "application/json"}, json={
"error_start": error_start,
"error_end": error_end,
})
# 該当期間にエラー停止アドレスがない場合は404が返る
if res.status_code == 404:
print("エラー停止アドレスはありません")
return []
job_id = res.json()["job_id"]
print(f"ジョブ開始: job_id={job_id}")
# 2. 完了を待つ
while True:
res = requests.get(f"{BASE}/errors/list/{job_id}", headers=HEADERS)
job = res.json()
print(f" 進捗: {job['percentage']}% ステータス: {job['status']}")
if job["status"] == "FINISHED":
break
if job["status"] == "FAILED":
raise Exception("ジョブが失敗しました")
time.sleep(5)
if job["total_count"] == 0:
print("エラー停止アドレスはありません")
return []
# 3. CSVダウンロード
res = requests.get(f"{BASE}/errors/list/{job_id}/download", headers=HEADERS)
z = zipfile.ZipFile(io.BytesIO(res.content))
csv_filename = z.namelist()[0]
with z.open(csv_filename) as f:
# BOM付きUTF-8
content = f.read().decode("utf-8-sig")
reader = csv.DictReader(io.StringIO(content))
errors = list(reader)
print(f"エラー停止アドレス: {len(errors)}件")
return errors
# 今月のエラー停止リストを取得
errors = get_error_list(
error_start="2026-02-01T00:00:00+09:00",
error_end="2026-02-28T23:59:59+09:00"
)
# エラー停止アドレスの一覧を出力
bounce_addresses = set()
for e in errors:
print(f" {e['メールアドレス']} (コード: {e['応答コード']})")
bounce_addresses.add(e["メールアドレス"])
配信リストのクリーニングと組み合わせる
取得したエラー停止アドレスを使って、次回の一斉配信から除外する例です。
def clean_and_send(all_recipients, bounce_addresses, delivery_id):
"""バウンスアドレスを除外して宛先を登録"""
cleaned = [r for r in all_recipients if r["email"] not in bounce_addresses]
removed = len(all_recipients) - len(cleaned)
print(f"除外: {removed}件 / 配信対象: {len(cleaned)}件")
# 50件ずつ配信先アドレスを登録
for recipient in cleaned:
requests.post(
f"{BASE}/deliveries/{delivery_id}/emails",
headers={**HEADERS, "Content-Type": "application/json"},
json={
"email": recipient["email"],
"insert_code": recipient.get("insert_code", [])
}
)
# 使用例
all_recipients = [
{"email": "user1@example.jp", "insert_code": [{"key": "__name__", "value": "田中"}]},
{"email": "invalid@example.jp", "insert_code": [{"key": "__name__", "value": "不明"}]},
# ...
]
clean_and_send(all_recipients, bounce_addresses, delivery_id=123)
大量の宛先がある場合: 1件ずつの登録ではなく、CSVファイルを生成してCSV一括登録API(POST /deliveries/{delivery_id}/emails/import)を使うのが効率的です。
配信ログと組み合わせた分析
エラー停止リストだけでなく、配信ログAPIも併用することで、より詳細な分析ができます。
from collections import Counter
def analyze_errors(delivery_id):
"""配信のエラー傾向を分析"""
logs = []
anchor = None
while True:
params = {"delivery_id": delivery_id, "status[]": ["HARDERROR", "SOFTERROR", "DROP"], "count": 1000}
if anchor:
params["anchor"] = anchor
res = requests.get(f"{BASE}/logs/mails/results", headers=HEADERS, params=params)
data = res.json()["data"]
if not data:
break
logs.extend(data)
anchor = data[-1]["maillog_id"]
# ステータス別集計
status_count = Counter(log["status"] for log in logs)
code_count = Counter(log["last_response_code"] for log in logs)
print(f"=== 配信ID {delivery_id} のエラー分析 ===")
print(f"ハードエラー: {status_count.get('HARDERROR', 0)}件")
print(f"ソフトエラー: {status_count.get('SOFTERROR', 0)}件")
print(f"ドロップ: {status_count.get('DROP', 0)}件")
print(f"\nレスポンスコード別")
for code, count in code_count.most_common():
print(f" {code}: {count}件")
return logs
analyze_errors(delivery_id=123)
出力例
=== 配信ID 123 のエラー分析 ===
ハードエラー: 15件
ソフトエラー: 8件
ドロップ: 12件
レスポンスコード別
554: 12件
550: 10件
421: 5件
552: 3件
554 が多い場合、エラー停止中のアドレスに配信している=配信リストのクリーニングが不足しているということです。
定期運用のポイント
| 項目 | 推奨 |
|---|---|
| エラー停止リストの取得頻度 | 週1回 or 配信前 |
| ハードエラーアドレスの扱い | 即座にリストから除外 |
| ソフトエラーアドレスの扱い | 3回連続なら除外を検討 |
| 554(errors)の監視 | 多発する場合はリストクリーニングを強化 |
| エラー停止期間 | 最大2週間で自動解除される |
まとめ
blastengineのエラー停止機能は自動でバウンスアドレスへの配信を止めてくれますが、停止期間は最大2週間で解除されます。エラー停止リストAPIで定期的にバウンスアドレスを取得し、自前の配信リストからも除外する運用を組み合わせることで、送信元の評価を維持し、メールの到達率を高く保てます。