近年、メール配信APIを利用したトランザクションメール運用は一般化しつつあります。その中でもblastengineは、国内サービスならではの高い到達率と分かりやすいAPI仕様を備えており、既存システムへの組み込みも容易です。一方で、初めて利用する際には「認証方式が分かりにくい」「どのヘッダを付与すべきか迷う」といった課題が発生しがちです。
本記事では、blastengine APIの認証方式であるBearer Token認証に焦点を当て、トークン生成方法からHTTPヘッダ設定、運用上の注意点までを実装中心に整理します。PythonおよびcURLの具体的なコード例と、実際に遭遇しやすいエラーへの対応方法も解説するため、この記事だけで認証まわりの実装が完結できるはずです。安全かつ効率的なメール配信実装の参考として活用してください。
blastengineの認証方式概要(Bearer Token)
blastengine APIは、Bearer Token認証方式を採用しています。この認証方式では、APIキー発行時に生成されるBearerTokenをHTTPリクエストのAuthorizationヘッダに設定することで、APIへのアクセスが可能になります。
Bearer認証の特徴
- ユーザー単位の認証: BearerTokenはユーザーごとに異なり、そのユーザーのアクセス権限を保持します
- シンプルな実装: HTTPヘッダに認証トークンを設定するだけで利用可能
- セキュアな通信: 全ての通信はHTTPSプロトコルで暗号化されます
Bearer Token の取得方法
Bearer Tokenは、ログインIDとAPIキーを組み合わせて生成します。
生成手順
- APIキーの発行: blastengine管理画面からAPIキーを発行します
- BearerTokenの生成: 以下の手順でトークンを生成します
生成コマンド
# 1. ログインID + APIキー に SHA256 をかける
# 2. 大文字を全て小文字化する
# 3. base64 エンコードをかける
YOUR_LOGIN_ID=ログインID
YOUR_API_KEY=APIキー
# SHA256ハッシュを生成
YOUR_BEARER_TOKEN=$(echo -n ${YOUR_LOGIN_ID}${YOUR_API_KEY} | shasum -a 256 | awk '{print $1}')
# 小文字化
YOUR_BEARER_TOKEN=$(echo -n ${YOUR_BEARER_TOKEN} | tr A-Z a-z)
# Base64エンコード
YOUR_BEARER_TOKEN=$(echo -n ${YOUR_BEARER_TOKEN} | base64 | tr -d "\n")
echo ${YOUR_BEARER_TOKEN}
APIリクエストへのBearer Token設定(共通ヘッダ)
すべてのAPIリクエストには、以下の共通ヘッダが必要です。
必須ヘッダ
| ヘッダ名 | 説明 | 設定値 |
|---|---|---|
| Authorization | Bearer Token認証情報 | Bearer {YOUR_BEARER_TOKEN} |
| Content-Type | リクエストボディの形式 |
application/json (一部 multipart/form-data) |
| Accept-Language | ロケール情報 |
ja-JP または en-US
|
ヘッダ設定例
Authorization: Bearer YOUR_BEARER_TOKEN
Content-Type: application/json
Accept-Language: ja-JP
cURL のリクエスト例
トランザクションメール送信(基本例)
curl -X POST https://app.engn.jp/api/v1/deliveries/transaction \
-H "Authorization: Bearer YOUR_BEARER_TOKEN" \
-H "Content-Type: application/json" \
-H "Accept-Language: ja-JP" \
-d '{
"from": {
"email": "sender@example.com",
"name": "送信者名"
},
"to": "recipient@example.jp",
"subject": "テスト件名",
"text_part": "テキスト本文です。",
"encode": "UTF-8"
}'
配信一覧の取得
curl -X GET "https://app.engn.jp/api/v1/deliveries?size=10&page=1" \
-H "Authorization: Bearer YOUR_BEARER_TOKEN" \
-H "Content-Type: application/json"
配信詳細の取得
curl -X GET https://app.engn.jp/api/v1/deliveries/1 \
-H "Authorization: Bearer YOUR_BEARER_TOKEN" \
-H "Content-Type: application/json"
Python Requests の実装例
基本的な実装(環境変数使用)
import os
import requests
import json
# 環境変数からBearer Tokenを取得
BEARER_TOKEN = os.environ.get('BLASTENGINE_BEARER_TOKEN')
BASE_URL = 'https://app.engn.jp/api/v1'
# 共通ヘッダの設定
headers = {
'Authorization': f'Bearer {BEARER_TOKEN}',
'Content-Type': 'application/json',
'Accept-Language': 'ja-JP'
}
# トランザクションメール送信
def send_transaction_email():
url = f'{BASE_URL}/deliveries/transaction'
payload = {
'from': {
'email': 'sender@example.com',
'name': '送信者名'
},
'to': 'recipient@example.jp',
'subject': 'テスト件名',
'text_part': 'テキスト本文です。',
'html_part': '<!DOCTYPE html><html><body><h1>HTML本文</h1></body></html>',
'encode': 'UTF-8'
}
try:
response = requests.post(url, headers=headers, json=payload)
response.raise_for_status() # HTTPエラーの場合は例外を発生
result = response.json()
print(f"配信ID: {result['delivery_id']}")
return result
except requests.exceptions.HTTPError as e:
print(f"HTTPエラー: {e}")
print(f"レスポンス: {e.response.text}")
raise
except requests.exceptions.RequestException as e:
print(f"リクエストエラー: {e}")
raise
# 配信一覧の取得
def get_deliveries(size=100, page=1):
url = f'{BASE_URL}/deliveries'
params = {
'size': size,
'page': page,
'sort': 'delivery_time:desc'
}
try:
response = requests.get(url, headers=headers, params=params)
response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as e:
print(f"エラー: {e}")
raise
# 実行例
if __name__ == '__main__':
# メール送信
result = send_transaction_email()
# 配信一覧取得
deliveries = get_deliveries(size=10)
print(f"取得件数: {len(deliveries.get('data', []))}")
レート制限対応版
import time
from requests.adapters import HTTPAdapter
from requests.packages.urllib3.util.retry import Retry
def create_session_with_retry():
"""リトライ機能付きセッションの作成"""
session = requests.Session()
# リトライ設定
retry_strategy = Retry(
total=3, # 最大リトライ回数
backoff_factor=1, # リトライ間隔(秒)
status_forcelist=[429, 500, 502, 503, 504],
allowed_methods=["GET", "POST", "PUT", "DELETE", "PATCH"]
)
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount("https://", adapter)
session.mount("http://", adapter)
return session
# 使用例
session = create_session_with_retry()
def send_email_with_retry():
url = f'{BASE_URL}/deliveries/transaction'
payload = {...} # メールデータ
try:
response = session.post(url, headers=headers, json=payload, timeout=30)
response.raise_for_status()
# レート制限情報の取得
remaining = response.headers.get('X-Rate-Limit-Remaining')
print(f"残りリクエスト数: {remaining}")
return response.json()
except requests.exceptions.HTTPError as e:
if e.response.status_code == 429:
# レート制限エラーの場合
retry_after = e.response.headers.get('X-Rate-Limit-Retry-After-Seconds')
print(f"{retry_after}秒後にリトライしてください")
raise
よくあるエラーと対応方法
401 Unauthorized(認証エラー)
原因:
- Bearer Tokenが間違っている
- Bearer Tokenの生成方法が正しくない
- Authorizationヘッダの形式が間違っている
対処方法:
# 正しい形式
headers = {
'Authorization': f'Bearer {BEARER_TOKEN}' # ○
}
# 間違った形式
headers = {
'Authorization': BEARER_TOKEN # × "Bearer "プレフィックスがない
}
403 Forbidden(権限エラー)
原因:
- APIキーが間違っている
- APIキーが発行されていない
対処方法:
- 管理画面でAPIキーを確認
429 Too Many Requests(レート制限)
原因:
- リクエスト数が制限を超えた(500req/分)
対処方法:
def handle_rate_limit(response):
if response.status_code == 429:
retry_after = int(response.headers.get('X-Rate-Limit-Retry-After-Seconds', 60))
print(f"{retry_after}秒待機します...")
time.sleep(retry_after)
return True
return False
# 使用例
response = requests.post(url, headers=headers, json=payload)
if handle_rate_limit(response):
# リトライ
response = requests.post(url, headers=headers, json=payload)
400 Bad Request(バリデーションエラー)
原因:
- リクエストパラメータが不正
- 必須フィールドが欠けている
- データ形式が間違っている
対処方法:
try:
response = requests.post(url, headers=headers, json=payload)
response.raise_for_status()
except requests.exceptions.HTTPError as e:
if e.response.status_code == 400:
error_detail = e.response.json()
print(f"バリデーションエラー: {error_detail}")
# エラー詳細を確認してリクエストを修正
500 Internal Server Error(サーバーエラー)
原因:
- サーバー側の一時的なエラー
対処方法:
- 時間をおいて再試行
- エラーが継続する場合はサポートに問い合わせ
セキュリティ上の注意点
1. 環境変数での管理
Bearer Tokenは絶対にソースコードに直接記述しないでください。
# ❌ 悪い例
BEARER_TOKEN = "ZjNkMjM0NTY3ODkwYWJjZGVmMTIzNDU2Nzg5MA=="
# ✅ 良い例
import os
BEARER_TOKEN = os.environ.get('BLASTENGINE_BEARER_TOKEN')
2. .envファイルの活用
# .env ファイル
BLASTENGINE_BEARER_TOKEN=your_bearer_token_here
BLASTENGINE_BASE_URL=https://app.engn.jp/api/v1
# Python (python-dotenvを使用)
from dotenv import load_dotenv
import os
load_dotenv()
BEARER_TOKEN = os.getenv('BLASTENGINE_BEARER_TOKEN')
重要: .envファイルは.gitignoreに追加してリポジトリにコミットしないこと
# .gitignore
.env
.env.local
.env.*.local
3. トークンのローテーション
定期的にAPIキーを再発行し、Bearer Tokenを更新することを推奨します。
ローテーション手順:
- 新しいAPIキーを発行
- 新しいBearer Tokenを生成
- アプリケーションの環境変数を更新
4. HTTPS通信の確保
- 本番環境では必ずHTTPSを使用
- 証明書の検証を無効化しない
# ❌ 危険
response = requests.post(url, headers=headers, json=payload, verify=False)
# ✅ 安全
response = requests.post(url, headers=headers, json=payload)
5. ログ出力時の注意
Bearer Tokenをログに出力しないよう注意してください。
import logging
# ❌ 悪い例
logging.info(f"Request headers: {headers}")
# ✅ 良い例
safe_headers = {k: v if k != 'Authorization' else '***MASKED***'
for k, v in headers.items()}
logging.info(f"Request headers: {safe_headers}")
まとめ
-
Bearer Token認証の理解
- ログインID + APIキーから生成(SHA256 → 小文字化 → Base64)
- Authorizationヘッダに
Bearer {token}形式で設定
-
必須ヘッダの設定
- Authorization: Bearer認証トークン
- Content-Type: application/json
- Accept-Language: ja-JP または en-US
-
エラーハンドリング
- HTTPステータスコードに応じた適切な処理
- レート制限(429)への対応
- リトライ処理の実装
-
セキュリティ対策
- 環境変数での管理
- .gitignoreへの登録
- 権限の最小化
- 定期的なトークンローテーション
本記事で紹介した実装例を参考に、安全で効率的なメール配信システムを構築してください。