イクバル・アバドゥラ
PythonからKaiMail HTTP APIでカスタムドメインメールを送信する
サーバーレス環境や厳しいネットワーク制限下で、メール送信に苦戦していませんか？KaiMail HTTP APIを活用すれば、Pythonから簡単にカスタムドメインメールを送信でき、SMTPの制約を回避できます。実装からエラーハンドリングまで、実践的なステップを詳しく解説します。
7月にKaiMailのHTTP Sending APIをリリースしたとき、私は話はそれで終わりだと思っていた。JSONをPOSTして、tracking IDを受け取って、それだけの話だ。
ところが、自分で使い始めてみると、同じような摩擦に何度も遭遇した。401はパスワードの間違いだった。403はメールボックスの登録を忘れたせいだった。クォータエラーは本番環境で初めて気づいた。毎回思った。「こういうのが全部壊れるパターンも含めて、一つのガイドにまとまってあるべきだ。」
これが、そのガイドである。
HTTP APIを使う理由
ポート587のSMTPは標準だ。それが許可されている場所ではどこでも動く。しかし「どこでも」という言葉が実はかなり重い。
制限の厳しいVPC内のAWS Lambda、Netlify Functions、一部のDigitalOcean droplet、そして意外と多いマネージドホストは、アウトバウンドSMTPを単純にブロックしている。ブロックされていなくても、ミリ秒単位で課金されるサーバーレス環境では、SMTPハンドシェイクがタイムアウトすることもある。
HTTP APIは、同じパイプラインへの別の入り口だ。メッセージは同じくDKIM署名され、同じ検証を通り、同じ配信ログに残る。違いは話しかけ方だけだ。STARTTLS越しのMIMEではなく、HTTPS越しのJSONだ。
コードがwebhookにPOSTできるなら、メールも送信できる。
必要なもの
コードを書く前に、以下を確認する。
- SMTP送信が有効なKaiMailの有料プラン。
- KaiMailに登録済みのカスタムドメイン。DKIMとSPFのレコードが公開済みであること。
- そのドメイン上に登録済みのメールボックス。送信に使う完全なアドレス。
- KaiMailダッシュボードから取得したSMTPパスワード。これはログインパスワードではない。プロフィールページに表示される別の認証情報だ。
ステップ3がないと、すべての送信が mailbox_not_registered で失敗する。私自身、これで20分ほどデバッグした。
最小限の送信例
APIは https://kaimail.net/api/v1 にある。HTTP Basic認証を使い、ユーザー名はKaiMailアカウントのメールアドレス、パスワードはSMTPパスワードだ。
最小限の動作例は以下の通りだ：
import os
import requests
API_BASE = "https://kaimail.net/api/v1"
AUTH = (os.environ["KAIMAIL_USER"], os.environ["KAIMAIL_PASS"])
resp = requests.post(
f"{API_BASE}/email/send",
auth=AUTH,
json={
"from": "Support <[email protected]>",
"to": ["[email protected]"],
"subject": "Welcome aboard",
"text": "Hello — thanks for signing up.",
},
timeout=30,
)
print(resp.status_code)
print(resp.json())
保存して、KAIMAIL_USER と KAIMAIL_PASS を環境変数に設定し、実行する。設定が正しければ 202 Accepted が返る：
{
"request_id": "0f6d2c9be5a34f7f9f0c1d2e3a4b5c6d",
"data": {
"tracking_id": "k2vq81xw3n7p5d09smc4",
"status": "queued",
"recipients": ["[email protected]"]
}
}
202 はメッセージが検証を通り、キューに入れたことを意味する。配信されたわけではない。配信状況を知るには、ステータスエンドポイントをポーリングする必要がある。
HTMLメールの送信
text フィールドはプレーンテキストだ。HTMLが必要なら html フィールドを追加する。両方を指定すると、サーバー側で自動的に multipart/alternative メッセージが構築される。
import os
import requests
API_BASE = "https://kaimail.net/api/v1"
AUTH = (os.environ["KAIMAIL_USER"], os.environ["KAIMAIL_PASS"])
resp = requests.post(
f"{API_BASE}/email/send",
auth=AUTH,
json={
"from": "Support <[email protected]>",
"to": ["[email protected]"],
"subject": "Welcome aboard",
"text": "Hello — thanks for signing up.",
"html": "<p>Hello — thanks for <strong>signing up</strong>.</p>",
},
timeout=30,
)
print(resp.status_code)
print(resp.json())
複数宛先への送信
to フィールドは単一の文字列でも文字列の配列でも受け付ける。1メッセージあたり最大100宛先まで。
import os
import requests
API_BASE = "https://kaimail.net/api/v1"
AUTH = (os.environ["KAIMAIL_USER"], os.environ["KAIMAIL_PASS"])
resp = requests.post(
f"{API_BASE}/email/send",
auth=AUTH,
json={
"from": "Support <[email protected]>",
"to": ["[email protected]", "[email protected]", "[email protected]"],
"subject": "Welcome aboard",
"text": "Hello — thanks for signing up.",
},
timeout=30,
)
print(resp.status_code)
print(resp.json())
重要：すべての宛先がメッセージの To: ヘッダーに表示される。Bccのセマンティクスが必要な場合、つまり宛先がヘッダーに表示されずにメッセージを受け取る場合は、send-raw エンドポイントが必要だ。
send-raw で添付ファイルを送信する
添付ファイル、カスタムヘッダー、MIME構造の完全な制御が必要な場合は POST /api/v1/email/send-raw を使う。完全なRFC 822メッセージをbase64エンコードしたものと、エンベロープ宛先を渡す。
スクリプトと同じディレクトリに invoice.pdf を置き、実行する：
import os
import base64
import requests
from email.message import EmailMessage
API_BASE = "https://kaimail.net/api/v1"
AUTH = (os.environ["KAIMAIL_USER"], os.environ["KAIMAIL_PASS"])
msg = EmailMessage()
msg["From"] = "Billing <[email protected]>"
msg["To"] = "[email protected]"
msg["Subject"] = "Your invoice"
msg.set_content("Invoice attached.")
with open("invoice.pdf", "rb") as f:
msg.add_attachment(
f.read(),
maintype="application",
subtype="pdf",
filename="invoice.pdf",
)
resp = requests.post(
f"{API_BASE}/email/send-raw",
auth=AUTH,
json={
"raw_message": base64.b64encode(msg.as_bytes()).decode("ascii"),
"to": ["[email protected]"],
"envelope_from": "[email protected]",
},
timeout=30,
)
print(resp.status_code)
print(resp.json())
ここでの to はSMTPエンベロープ、つまり実際の配信アドレスだ。メッセージ内の To: ヘッダーは表示用のみ。この分離がBccの仕組みだ：エンベロープの to 配列にアドレスを含め、メッセージヘッダーからは省略する。
送信後のステータス確認
tracking_id を取得したら、GET /api/v1/email/<tracking_id> をポーリングして状況を確認する。送信レスポンスからIDを貼り付ける：
import os
import requests
API_BASE = "https://kaimail.net/api/v1"
AUTH = (os.environ["KAIMAIL_USER"], os.environ["KAIMAIL_PASS"])
tracking_id = "k2vq81xw3n7p5d09smc4"
status_resp = requests.get(
f"{API_BASE}/email/{tracking_id}",
auth=AUTH,
timeout=30,
)
print(status_resp.status_code)
print(status_resp.json())
成功した送信は以下のようになる：
{
"request_id": "6b1a2c3d4e5f60718293a4b5c6d7e8f9",
"data": {
"tracking_id": "k2vq81xw3n7p5d09smc4",
"status": "sent",
"envelope_from": "[email protected]",
"recipients": ["[email protected]"],
"subject": "Welcome aboard",
"created_time": "2026-07-20T04:12:33Z",
"processed_time": "2026-07-20T04:12:41Z",
"error": null
}
}
ステータス遷移は単純だ：queued → sending → sent。あるいは一時的なエラー後に3回リトライしても失敗した場合、queued → sending → failed となる。
sent は「KaiMailのメールサーバーが配信を受け入れた」を意味する。相手の受信箱に届いたわけではない。バウンスや宛先ごとの結果は、Logsダッシュボードで確認する。
クォータの確認
バッチ送信の前に、残りクォータを確認する：
import os
import requests
API_BASE = "https://kaimail.net/api/v1"
AUTH = (os.environ["KAIMAIL_USER"], os.environ["KAIMAIL_PASS"])
usage_resp = requests.get(
f"{API_BASE}/usage",
auth=AUTH,
timeout=30,
)
print(usage_resp.status_code)
print(usage_resp.json())
レスポンス：
{
"request_id": "9c8b7a6d5e4f30211203f4e5d6c7b8a9",
"data": {
"smtp_sending": true,
"sent_this_period": 42,
"quota": 1000,
"remaining": 958,
"period_days": 30
}
}
クォータは宛先ごとにカウントされる。1通を10人に送ると10単位を消費する。キューに入ったが未送信のメッセージもカウントされるため、キューに入れただけで上限を超えることはない。
エラーハンドリング：実際に遭遇するエラー
APIは機械可読なエラーコードを返す。以下は実際によく出るものだ。
invalid_credentials (401)
パスワードが間違っている。SMTPパスワードを使う。ダッシュボードのログインパスワードではない。両者は意図的に別々のシークレットになっており、API認証情報をローテーションしてもWebインターフェースに締め出されないようにしている。
sending_not_allowed (403)
プランにアウトバウンド送信が含まれていない。Plus、Pro、Businessにアップグレードする。
domain_not_owned (403)
送信元ドメインがKaiMailアカウントに登録されていない。所有していないアドレスからは送信できない。先にダッシュボードでドメインを追加する。
mailbox_not_registered (403)
ドメインは所有しているが、送信に使った具体的なアドレスが登録済みのメールボックスではない。これは私も引っかかった。yourdomain.com を追加して [email protected] から送信しようとしたが、hello メールボックスを実際には作成していなかった。ダッシュボードでメールボックスを作成してから試す。
quota_exceeded (429)
月間上限に達した。/usage を確認してバックオフする。大きなバッチを送る前にクォータを確認する習慣をつける。
invalid_request (400)
フィールドの欠落または不正な形式。よくある原因：from の欠落、to の欠落、text も html もない、不正なメールアドレス、100宛先を超えている。エラーメッセージにどのフィールドが失敗したかが含まれる。
invalid_message (400)
send-raw のみ。raw_message が有効なbase64ではないか、デコード後に空のメッセージになる。標準的なパディング付きbase64を使うことを確認する。
message_too_large (413)
デコード後のメッセージがプランのサイズ上限を超えている。base64はペイロードを約33%膨張させるが、制限はデコード後のバイト数に適用される。
再利用可能なPythonクライアント
プロジェクトにドロップできる小さなモジュールだ。APIをラップし、エラーを処理し、ステータスをポーリングする：
import os
import time
from typing import Literal
import requests
class KaiMailClient:
def __init__(self, username: str, password: str, base_url: str = "https://kaimail.net/api/v1"):
self.base_url = base_url.rstrip("/")
self.auth = (username, password)
def _post(self, path: str, payload: dict) -> dict:
resp = requests.post(
f"{self.base_url}{path}",
auth=self.auth,
json=payload,
timeout=30,
)
data = resp.json()
if resp.status_code != 202:
raise KaiMailError(data["error"]["code"], data["error"]["message"], data["request_id"])
return data
def _get(self, path: str) -> dict:
resp = requests.get(f"{self.base_url}{path}", auth=self.auth, timeout=30)
resp.raise_for_status()
return resp.json()
def send(
self,
from_addr: str,
to_addrs: list[str],
subject: str,
text: str | None = None,
html: str | None = None,
) -> str:
if not text and not html:
raise ValueError("Provide at least one of text or html")
payload = {
"from": from_addr,
"to": to_addrs,
"subject": subject,
}
if text:
payload["text"] = text
if html:
payload["html"] = html
data = self._post("/email/send", payload)
return data["data"]["tracking_id"]
def send_raw(self, raw_message_b64: str, to_addrs: list[str], envelope_from: str | None = None) -> str:
payload = {"raw_message": raw_message_b64, "to": to_addrs}
if envelope_from:
payload["envelope_from"] = envelope_from
data = self._post("/email/send-raw", payload)
return data["data"]["tracking_id"]
def get_status(self, tracking_id: str) -> dict:
return self._get(f"/email/{tracking_id}")
def wait_for_status(
self,
tracking_id: str,
desired: Literal["sent", "failed"] = "sent",
max_attempts: int = 10,
delay: float = 2.0,
) -> dict:
for _ in range(max_attempts):
data = self.get_status(tracking_id)
status = data["data"]["status"]
if status in (desired, "failed"):
return data
time.sleep(delay)
raise TimeoutError(f"Status did not reach {desired} after {max_attempts} attempts")
def usage(self) -> dict:
return self._get("/usage")
class KaiMailError(Exception):
def __init__(self, code: str, message: str, request_id: str):
self.code = code
self.message = message
self.request_id = request_id
super().__init__(f"[{code}] {message} (request_id: {request_id})")
# Example usage
if __name__ == "__main__":
client = KaiMailClient(
os.environ["KAIMAIL_USER"],
os.environ["KAIMAIL_PASS"],
)
tracking_id = client.send(
from_addr="Support <[email protected]>",
to_addrs=["[email protected]"],
subject="Welcome aboard",
text="Hello — thanks for signing up.",
)
print(f"Queued: {tracking_id}")
result = client.wait_for_status(tracking_id)
print(f"Final status: {result['data']['status']}")
インテグレーションチェックリスト
このAPIを呼ぶコードをリリースする前に確認するべき項目：
- ドメインがKaiMailに登録済み。DKIMレコードが公開されている。
- 送信に使うアドレス用のメールボックスが作成済み。
- SMTPパスワードが安全に保存されている。環境変数やシークレットマネージャーを使い、ソースコードには含めない。
- エラーハンドリングは
error.codeを使って分岐しており、メッセージテキストに依存していない。 - 大きなバッチを送る前にクォータを確認している。
tracking_idは後からデバッグで問い合わせできるようにどこかに保存している。- ステータスポーリングにタイムアウトがあり、永遠にハングアップしない。
send-rawを使う場合、base64エンコーディングにパディングが含まれており、デコード後のサイズがプランの制限に収まっていることを確認する。
SMTPを使うべき場合
HTTP APIはSMTPの代替ではない。SMTPが不便または不可能な場合の代替手段だ。
SMTPを使うのは：デスクトップのメールクライアントから送信する場合、メーリングリストから送信する場合、すでにネイティブにSMTPを話すアプリケーションがある場合。
HTTPを使うのは：ポート587をブロックしているプラットフォーム、サーバーレス環境、JSONの方がMIME生成より楽な場合、プログラムから送信ステータスやクォータにアクセスしたい場合。
次のステップ
HTTP APIは安定しており、私たち自身も本番で使っている。予期しない問題が発生した場合、すべてのレスポンスに request_id が含まれる。問い合わせの際にそれを伝えれば、正確に何が起きたかを追跡できる。
試してみてほしい。1つのPOSTから始めて、小さなクライアントにラップして、そこから構築していく。
参考リンク：