Webhookなしでメールを待つ方法：コードで書くポーリングループ

プッシュとプル、それぞれのコスト

自分のコードがメッセージの到着を知る方法は、2つしかありません。相手が知らせてくれるか、自分で尋ねるかです。それ以外のもの — waitForを持つクライアントライブラリ、受信箱を「ストリーミングする」SDK、ブロックするテストヘルパー — はどれも、仕組みを隠しただけのこの2つのどちらかにすぎません。デバッグを迫られる前に、自分が今どちらを手にしているのかを知っておく価値があります。

提供されているもののほぼすべては、この4つの組み合わせでカバーできます。

Webhook

メールが届くたびに、サービス側があなたが所有するアドレスへHTTPリクエストを送ります。これは考えられる中で最も安い待ち方です — 何かをするべき瞬間まで、何もしなくていいのです — その代わりに必要になるのが、パブリックインターネット上のアドレス、メールが届いた瞬間に立ち上がっているリスナー、そのリクエストが本当に相手からのものであることを証明する共有シークレット、そして自分のリスナーがそこにいなかったときにどうするかという答えを自分で用意することです。

ロングポーリング

あなたがリクエストを送り、サーバーはメールが届くか、タイムアウトになるまでそれを開いたままにします。あなたに必要なのは外向きの接続だけですが、サーバー側は待っている人数分のワーカーを消費します — だからこそ、これを提供するサービスはどこも、待つ時間の長さと同時に待てる人数の両方に上限を設けています。

普通のポーリング

あなたが繰り返し尋ね、どのリクエストにもその場にあるものがすぐに返ってきます。ルーターの内側にあるノートPCからでも、外部からの着信経路を持たないCIランナーからでも、他人のサンドボックスの中で動くエージェントからでも機能する、唯一の形です — そして、これこそがこのガイド全体のテーマです。

メールボックスのプロトコル

IMAPにはIDLEがあり、これは別の顔をしたロングポーリングです。接続を開いたままにしておき、新しいメールが届くとサーバー側がそれをその接続上で知らせてきます。これは本物のプッシュに近い仕組みですが、認証情報付きのメールボックス、ソケットを開いたまま保持し、切れたら再接続できるクライアント、そしてそのコマンドに応じてくれるサーバーを必要とします — メッセージ1通のためだけの仕事にしては、かなり大掛かりな仕組みです。

並べてみると、この選択はエレガントさの話ではなく、それぞれが、あなたのコードが動いているマシンに何を要求するかという話であることが分かります。

誰もが最初に書くループ

それは4行で書けて、書いたその日はうまく動きますが、その問題のひとつひとつは、あとになって、別の場所で表面化します。真夜中3時に動くパイプラインの中で、11分間「考え中」のままになっているエージェントの中で、あなたのループがその枠を握ったままにしているせいで同僚に429を返してくるメールボックスの中で。

import time
import requests

while True:
    r = requests.get("https://grabmail.io/api/v1/mailbox",
                     params={"address": "signup-42@grabmail.io"})
    if r.json()["messages"]:
        break
    time.sleep(1)

これには5つの問題があり、明らかなのは最初の1つだけです。

決してあきらめない

期限がないため、メッセージが本当に届かないとき — フォームがアドレスを拒否した、送信側のキューが詰まっている、誰かがドメインを打ち間違えた — このループは失敗しません。ハングするのです。ハングするジョブは失敗するジョブより始末が悪く、ログは理由を語ることなくそこで終わってしまいます。

試行回数を数えて、それを秒だと思い込む

パス数に上限を付けたとしても、「1秒」を30回試行しても30秒にはなりません。各パスにはリクエストのコストもかかり、400msかかるリクエストは、あなたの30秒を42秒にしてしまいます。リトライを1つ加えれば、その足し算はもはや足し算ではなくなります。

すべてのランナーが同じタイミングで尋ねる

同じパイプラインから20個のジョブを起動すると、それらは一斉にポーリングします。互いに数ミリ秒の差で始まり、全員が同じ1秒間スリープするからです。ピークは平均の20倍になり、拒否されるのはそのピークのほうです。

自分宛てではなく、最新のメッセージを取ってしまう

一覧の最初のエントリは、そのメールボックスの一番上にあるものにすぎません。公開アドレスなら誰か他人のメールかもしれず、再利用したアドレスなら先週のものかもしれません。最初に見つけたメッセージで終了するループは、本来待っていたメッセージが届く前に、平気で終了してしまいます。

どの応答も成功として扱ってしまう

429や404の応答からメッセージ一覧を読み出そうとすると、原因から3フレームも離れた場所でエラーが発生し、500から読み出そうとすると、何のエラーも起きないことさえあります。ステータスコードは最後に見るものではなく、最初に見るべきものです。

回数ではなく、時計で止める

期限は、最初のリクエストより前に、モノトニックな時計 — マシンが時刻を補正しても後ろに戻らない時計 — から一度だけ取得し、各パスの先頭でそれと比較してください。そうしておけば、ループの中の他のすべては、待ち時間の長さを変えることなく自由に変更できます。間隔を広げても、拒否をリトライしても、フィルタを1つ増やしても、90秒はやはり90秒のままです。

どのくらい待てば十分かは、あなたについての問いではなく、送信側についての問いです。フォームへの応答として機械が生成するメールは、たいてい1桁の秒数で届きます。滞留したキュー、グレーリスティングを行う受信側、1時間ごとのバッチ処理は、まるで違う桁の話であり、どんな間隔を選んでも、それが早く届くようにはなりません。

期限はまた、エラーメッセージを置くのに正直な場所でもあります。「『Confirm your email』に一致するものは、signup-42@grabmail.io宛てに90秒以内に届きませんでした」は、アドレス、フィルタ、そして枠を名指ししており、これは何が起きたかを突き止めるのに必要な4つのうち3つに当たります。4つ目 — 実際に何が届いたか — も出力しておく価値があります。ループが見て、そして拒否した件名の一覧があれば、「不安定だ」は一読で「件名が変わった」に変わります。

どのくらいの頻度で尋ね、いつ間隔を広げるか

下限は、そのサービスが許す値であり、ここではアドレスごとに1秒に1リクエストです。これは遠慮してほしいという意味ではありません — 1秒に1回のポーリングはまさに想定された使い方であり、日次のクォータも月次のクォータも、管理すべきバーストクレジットもありません — ただしこれは下限であり、同じ1秒の中で2回尋ねるループは、速い答えの代わりに2回目に対する429を受け取ります。

一定の間隔

毎回1秒、期限まで変わりません。10秒で終わる待機には十分すぎるほど良く、単一のランナー上の単一のテストにとっては正しい既定値です。唯一の欠点は、メールが来ないことがはっきりしたあとも、長々と同じペースで尋ね続けてしまうことです。

広がっていく間隔

メッセージがまだ配送中である可能性が高い間は1秒のままにし、そこから2秒、4秒、8秒と倍にしていき、上限を設けます。届くのが遅いメッセージに対しては多少のレイテンシを払いますが、どうせ失敗するはずだった待機では、リクエストの大半を節約できます。必ず上限を設けてください。上限なしに倍々にしていく間隔は、2分の期限の後半をほとんどスリープに費やしてしまいます。

ジッターは、足すだけで、引いてはならない

各パスをランダムな量だけずらして、20個のランナーが同じタイミングで尋ねるのを避けてください。よくあるやり方 — ゼロから間隔までの範囲でランダムな値を取る — は、ここでは間違いです。その範囲の半分が、1秒という下限を下回ってしまうからです。代わりに、ランダムな量を上に積んでください。間隔は最小値であり、ジッターはパスを遅らせる方向にだけ働くべきです。

自分では選べない停止時間

答えが429のとき、その間隔はRetry-Afterが示す値そのものであり、拒否されたそのパスは試行として数えるべきではありません。それを1回として数えてしまうと、スロットリングされているループは、メールボックスを一度も読むことなく、期限のすべてを拒否の収集に費やすことになります。

最初の15秒は1秒間隔、そこから8秒を上限に倍々にしていき、その上にジッターを載せる — これで、このガイドに出てくるほぼすべての待機が、6行に収まります。

def delay(attempt: int) -> float:
    # One second while the message is probably still in flight, then
    # wider. Never below a second: the list endpoint allows one call
    # per second, per address, so jitter is added and never taken off.
    step = 1.0 if attempt < 15 else min(8.0, 2.0 ** (attempt - 14))
    return step + random.uniform(0.0, step / 2)

指数にはオフセットが入っており、間隔が広がり始めるのは最初のパスからではなく、固定区間が終わったあとです。このオフセットがなければ、遅れて届くサインアップメールが届くころには間隔はすでに8秒に達してしまい、本来12秒で済むはずの待機が20秒かかってしまいます。

これらはいずれも、最初のリクエストには当てはまりません。スリープする前に、まず即座に尋ねてください。ループが始まった時点ですでにメールボックスに入っていたメッセージ — 待機が始まる前に何かをトリガーした場合の、ごく普通のケースです — に気づくために、1秒分のレイテンシを払う必要はないはずです。

拒否応答を読み解く

一覧エンドポイントからの応答はすべてJSONであり、メールボックスではない応答は、どれも同じ形をしています。分岐の基準にすべき、安定したerrorスラッグと、いつでも文言が変わりうる説明文であるmessageです。速すぎることに対する拒否には、ヘッダーも1つ付いてきます。

HTTP/1.1 429 Too Many Requests
Retry-After: 1
Content-Type: application/json; charset=utf-8

{"error":"rate_limited","message":"one request per second, per address"}

Retry-Afterは整数の秒数で、これは本当の数字です — ドキュメントに書かれた定数からではなく、このアドレスの枠に実際にどれだけ残っているかから取られています。ちょうどその秒数だけスリープするのが、最も礼儀正しく、かつ最も速い方法です。それより短く眠れば再び拒否され、長く眠ればその分の時間を無駄にします。ここに、ポーリングループが出会いうるすべての応答と、それぞれの応答が本当に求めているものをまとめます。

この7つのうち2つは「止まれ」を意味しており、この2つこそ声を大にして言う価値があります。unknown_domainを「まだ届いていない」として扱うループは、サービスが最初の40ミリ秒で伝えていたことを証明するために、90秒をまるごと費やしてしまいます。

どのメッセージが自分宛てか

メールボックスはキューではなく、その中の最新のものが、必ずしも自分が待っているものだとは限りません。公開ドメインでは、アドレスを推測できる人なら誰でもそこに送信できますし、テストスイートでは同じアドレスが実行のたびに再利用されることが多く、1回のサインアップがウェルカムメールと確認メールという2通を送ってきて、そのうちコードを含むのは片方だけということもよくあります。この解決策が基準点であり、それはメールを発生させる操作より前に取っておかなければなりません。

1. フォームを送信する前に、limit=1でメールボックスを一覧表示し、最新メッセージのidを控えておきます。空であれば何も控えません。そのidが基準点です。
2. 実際にその操作を行う — フォームを送信する、エンドポイントを呼ぶ、ボタンを押す。
3. 一覧をポーリングする。メッセージは新しい順に返ってくるので、上から順に見ていき、基準点に達した瞬間に止めてください。そこから下はすべて自分の操作より古いものであり、読まずに無視できます。
4. それより上にあるものを送信者、件名、あるいは両方で絞り込みます。部分文字列で通常は十分ですが、ローカライズされない部分を選んでください — 「Confirm your email」に一致させているテストは、テスト対象のアカウントが別の言語に切り替わった日に失敗します。
5. そして、そのときになって初めて開きます。一覧には短いpreviewはあっても本文はなく、お目当てのコードはたいていその先にあります。もう1回リクエストすれば、メッセージ全体が手に入り、そのコストは一覧とは別の、はるかに大きな枠から差し引かれます。

シェルでは、この2つの読み取りはこのようになります — まず基準点、それからポーリングです。

curl -sG https://grabmail.io/api/v1/mailbox \
  --data-urlencode "address=signup-42@grabmail.io" --data-urlencode "limit=1"
curl -sG https://grabmail.io/api/v1/mailbox \
  --data-urlencode "address=signup-42@grabmail.io" --data-urlencode "limit=25"

どちらのリクエストも、アドレスを丸ごと名指ししています。ここではアドレスそのものがメールボックスであり、セッションもなく、あなたの代わりに保持されるカーソルもなく、1回の呼び出しについて次の呼び出しが覚えていることも何もないからです。これはまた、同じアドレスを2か所から同時に見張っても安全である理由でもあります — 読み取りは何も消費しないので、同じメールボックスを見ている2つのループはどちらもすべてのメッセージを見ることができ、どちらかが相手の足元からメッセージを奪うこともありません。

ちょうど1回だけ処理する

リトライされたポーリングは、同じメッセージを2回見ることがあり、これは珍しい出来事ではありません。サーバーが応答し、本文が届く前に接続が切れ、HTTPクライアントがリトライし、その2回目の応答には最初の応答が既に運んでいたメッセージが含まれている、という具合です。メッセージに対して行うのがリンクのクリックであれ、支払いの確認であれ、チャンネルへの投稿であれ、それを2回行うことは、あなたのプロセスの外にまで影響が及ぶバグになります。

処理済みのidを覚えておく

1つの関数の中で生まれて終わる待機であれば、メモリ上のidの集合で十分です。再起動をまたいで生き延びる必要があるもの — スケジュール実行のジョブが空にし続けるメールボックスや、積み残しを片付けているエージェント — については、それと一緒に生き延びる場所に書き留めておく必要があります。

削除はべき等

メッセージの削除は、2回目も1回目と同じように200を返すので、リトライされた削除が失敗のように見えることは決してなく、特別扱いも要りません。削除は行動する前ではなく、あとにしてください。その間にクラッシュしても、失われるのは再読み込みで済むもの — 取り戻せるコスト — であって、取り戻せないメッセージそのものではありません。

ここのidは、送信者のMessage-IDではない

APIにあるidは、こちら側が発行したものです。1つのメールボックスに限定されたスコープを持ち、メッセージが期限切れになれば存在しなくなります。Message-IDヘッダーは送信者のものであり、メッセージとともに運ばれます。同じメールを2つのシステムの間で一致させたいなら必要なのはこちらのほうです — その見つけ方はヘッダーのガイドにあります。

1つのコードを待ってメールボックスを使い捨てるだけのテストには、これらは何も必要ありません。しかしループが無人で動く瞬間から、そのすべてが必要になります。これが防ぐ失敗は、失敗のようには見えないからです — それは、作業が2回、しかも正しく行われたように見えるのです。

待つことをサーバーに任せられるとき

ここには、自分でループを書かなくてよい場所がひとつだけあります。それは、ループを持つ余裕のない呼び出し元のために存在しています。AIエージェントは、確認するたびに1ターン分のコストを払うため、「まだ何もない」と9回答えるツールは、9ターン分の空振りです。MCPサーバーのwait_for_messageは、代わりにリクエストを開いたまま保持し、こちら側でポーリングを行い、応答を1回だけ返します — メッセージそのもの、あるいは待ったが何も来なかったという簡潔な報告のどちらかです。

$ curl -sX POST https://grabmail.io/mcp -H "Content-Type: application/json" -d '{
  "jsonrpc":"2.0","id":1,"method":"tools/call",
  "params":{"name":"wait_for_message","arguments":{
    "address":"signup-42@grabmail.io","subject_contains":"code",
    "timeout_seconds":25}}}'

これを使ってものを組み立てる前に、知っておく価値のある4つのことがあります。

待つのは最大25秒

timeout_secondsで短くするよう頼むことはできますが、それより長くすることは決してできません。この上限は適当に決められたものではありません。待っている1件ごとに、眠っているだけのワーカーが1つ消費され、数分間開いたままのリクエストは、それが返ってくるよりずっと前に、誰かのプロキシのタイムアウトに引っかかって死んでしまうリクエストだからです。

入り口でフィルタする

from_contains、subject_contains、since_idは、前のセクションと同じ3つの判断を、サーバー側で行っているだけです。since_idが基準点であり、それはどこよりもここで重要になります。これがなければ、この呼び出しはメールボックスに既に入っていたものをそのまま即座に返してしまいます。

タイムアウトは応答であり、エラーではない

何も届かないと、実際に待った時間とともにtimed_outが設定された状態で返ってきて、待ち続けるにはもう一度呼び出せばよいと、はっきりそう述べます。サインアップメールへの待機として2〜3回の呼び出しは普通のことです — それがそのままループであり、90ターンではなく3ターンで済みます。

待機できるのは8件までで、順番待ちの列はない

そのすべてが使用中のとき、呼び出しはほかの7件のエージェントの後ろに並ぶのではなく、その場ですぐに、その旨を返してきます。これは正しい失敗の形です。「待機が多すぎる」と告げられたエージェントはメールボックスを一覧表示して先へ進めますが、列に並んで待つだけのエージェントにできることは、ただ座っていることだけです。

アドレスごとの制限は、その内部でも変わらず適用されています — こちら側のループも、あなたのループとまったく同じようにレート制限を受けており、サーバー側の待機はその下限を回避する方法ではなく、それをターンで払わずに済ませる方法にすぎません。テストスイートにとっては、これはわざわざ使う価値のあるものではありません。テストはもともとスリープしてよいプロセスであり、テストが書かれている言語でのループのほうが、遠隔のループよりはるかにデバッグしやすいからです。残りのツールについてはMCPガイドで扱っています。

ループの全体を、一度だけ

これまでの内容すべてを1つのファイルに収めます。モノトニックな時計から取った期限、片側だけのジッターで広がる間隔、試行として数えず尊重するRetry-After、何が新しいかを決める基準点、件名でのフィルタ、そして一覧がプレビューしかしていないメッセージを取得するための、追加の1リクエストです。

import random
import time
import requests

API     = "https://grabmail.io/api/v1"
ADDRESS = "signup-42@grabmail.io"


def delay(attempt: int) -> float:
    # One second while the message is probably still in flight, then
    # wider. Never below a second: the list endpoint allows one call
    # per second, per address, so jitter is added and never taken off.
    step = 1.0 if attempt < 15 else min(8.0, 2.0 ** (attempt - 14))
    return step + random.uniform(0.0, step / 2)


def watermark(s):
    # Read this BEFORE the form is submitted. Every id above it
    # afterwards is mail that arrived because of what you did.
    r = s.get(f"{API}/mailbox", params={"address": ADDRESS, "limit": 1})
    r.raise_for_status()
    seen = r.json()["messages"]
    return seen[0]["id"] if seen else None


def wait_for(s, subject, since, timeout=120.0):
    deadline = time.monotonic() + timeout
    attempt  = 0
    rejected = set()

    while time.monotonic() < deadline:
        r = s.get(f"{API}/mailbox", params={"address": ADDRESS, "limit": 25})

        if r.status_code == 429:
            time.sleep(int(r.headers.get("Retry-After", "1")))
            continue                      # refused, so it was not an attempt
        if r.status_code == 200:
            for m in r.json()["messages"]:        # newest first
                if m["id"] == since:
                    break                 # older than the watermark
                if subject.lower() in m["subject"].lower():
                    full = s.get(f"{API}/message/{m['id']}",
                                 params={"mailbox": ADDRESS})
                    full.raise_for_status()
                    return full.json()
                rejected.add(m["subject"])
        elif r.status_code < 500:
            raise RuntimeError(r.json().get("error", r.status_code))
        # a 5xx falls through: transient, and the deadline still governs

        time.sleep(delay(attempt))
        attempt += 1

    raise TimeoutError(
        f"nothing matching {subject!r} at {ADDRESS} in {timeout:.0f}s; "
        f"saw {sorted(rejected) or 'nothing at all'}")

これは意図的に、標準ライブラリと1つのHTTPクライアントだけによる50行あまりに収めています。インストールするものも、設定するものも、どこにもシークレットもありません — それこそが要点です。同じ形が、そのまま変わらずにNodeにも、シェルスクリプトにも、あなたのテストフレームワークが既にリクエストを送るために使っているものにも移せます。

1. 基準点は、メールを発生させる操作の前に読み取ってください。あとで読んではいけません。
2. まず即座に1回尋ね、それから初めてスリープしてください。先にスリープしてはいけません。
3. 期限はモノトニックな時計から取得し、各パスの先頭でそれを確認してください。
4. 間隔はアドレスごとに1秒以上を保ち、ジッターは上方向にだけ加えてください。
5. Retry-Afterが示す秒数だけ正確にスリープし、拒否を試行として数えないでください。
6. errorスラッグで分岐してください。unknown_domainとinvalid_addressは、待てという意味ではなく、止まれという意味です。
7. 送信者か件名で一致を取り、基準点に達したら一覧をたどるのを止めてください。
8. 解析する前にメッセージを開いてください — 一覧が運んでいるのはプレビューであり、本文ではありません。
9. 失敗するときは、アドレス、フィルタ、枠、そしてループが拒否した件名とともに失敗してください。

9つのルールのうち8つは、誰かがログから再構築するしかなかった失敗があったからこそ存在しています。失敗にまつわるものではない唯一の例外が2番目のルールです。最初のスリープより前に1回尋ねておくことで、すでに届いているメッセージへの待機は、1秒ではなく4ミリ秒で済むようになります — これは200件のテストからなるスイート全体では、あとで誰も説明する必要のない3分間の実時間に相当します。

質問