API & otomatisasi

Python: menerima email lewat API kotak masuk sekali pakai

Enam puluh baris Python, satu dependency, dan tanpa kunci API: sebuah alamat yang Anda buat sendiri, penantian dengan batas waktu, pesannya sebagai sebuah dict. Berikut modulnya di atas requests, versi yang sama di atas httpx untuk asyncio, paging melewati kotak surat yang sibuk, menyimpan sebuah lampiran, sebuah fixture pytest, dan enam kesalahan yang muncul pertama kali skrip ini berjalan tanpa pengawasan.

  • Menengah
  • 23 menit baca
Tabung abu-abu yang menggulung longgar, dengan amplop biru keluar dari ujungnya yang terbuka ke nampan abu-abu kecil

API-nya, sebagaimana dilihat Python

Tidak ada yang perlu diinstal di sisi server dan tidak ada yang perlu diautentikasi: sebuah kotak surat pada domain publik bisa dibaca siapa pun yang tahu alamatnya, lewat HTTPS biasa, sebagai JSON. Keseluruhan permukaannya adalah tiga panggilan:

GET /api/v1/mailbox?address=…
Semua yang menunggu di sebuah alamat, terbaru lebih dulu, sebagai sebuah daftar ringkasan. Kotak surat kosong adalah 200 dengan count: 0 — bukan pernah 404. limit membatasi satu response (1–200, default 50) dan before melakukan paging melewatinya.
GET /api/v1/message/{id}?mailbox=…
Satu pesan secara lengkap: pengirim, penerima, subjek, tanggal, bagian teks polos, bagian HTML (atau null), dan sebuah daftar lampiran yang masing-masing sudah punya URL siap pakai.
DELETE /api/v1/message/{id}?mailbox=…
Menghapusnya sekarang, bukan menunggu 5 hari. Idempoten: menghapus dua kali tetap menjawab 200.

Sebuah listing terlihat seperti ini. alias adalah alamat kedua pada domain terpisah yang mengantarkan surat ke kotak surat yang sama dan tidak bisa dipakai untuk membacanya — inilah yang diberikan ke sebuah situs kalau Anda tidak ingin situs itu bisa membuka kotak masuknya.

GET /api/v1/mailbox — response-nya
{
  "address": "py-3f9a1c2e@grabmail.io",
  "alias": "k7m2p9x4q1wz@example.net",
  "count": 1,
  "next": null,
  "messages": [
    { "id": "01JR8W2K4Q", "from": "noreply@example.com", "subject": "Your verification code",
      "date": "2026-09-01T18:31:07Z", "seen": false, "attachments": 0, "expires_at": "2026-09-06T18:31:07Z" }
  ]
}

Modulnya

Satu file, satu class, dan requests sebagai satu-satunya dependency. Ini sengaja dibuat membosankan: sebuah session, sebuah loop dengan batas waktu, dan satu-satunya retry yang pernah benar — sleep saat menerima 429.

grabmail.py
"""grabmail.py — a disposable inbox from Python. Three endpoints, no key, no account."""
from __future__ import annotations

import secrets
import time
from pathlib import Path

import requests

API = "https://grabmail.io/api/v1"
DOMAIN = "grabmail.io"


def fresh_address(prefix: str = "py") -> str:
    """A mailbox nothing else is using. Nothing has to be created first."""
    return f"{prefix}-{secrets.token_hex(4)}@{DOMAIN}"


class Inbox:
    def __init__(self, address: str | None = None, session: requests.Session | None = None):
        self.address = address or fresh_address()
        self.http = session or requests.Session()

    def _get(self, url: str, **params) -> requests.Response:
        """One GET, with the only retry that is ever right: waiting out a 429."""
        while True:
            r = self.http.get(url, params=params, timeout=15)
            if r.status_code == 429:
                time.sleep(float(r.headers.get("Retry-After", 1)))
                continue
            r.raise_for_status()
            return r

    def list(self, limit: int = 50, before: str | None = None) -> dict:
        params = {"address": self.address, "limit": limit}
        if before:
            params["before"] = before
        return self._get(f"{API}/mailbox", **params).json()

    def wait_for(self, timeout: float = 60, subject_contains: str | None = None,
                 from_contains: str | None = None) -> dict:
        """Block until a matching message arrives, then return it in full."""
        deadline = time.monotonic() + timeout
        while time.monotonic() < deadline:
            for m in self.list()["messages"]:
                if subject_contains and subject_contains.lower() not in m["subject"].lower():
                    continue
                if from_contains and from_contains.lower() not in m["from"].lower():
                    continue
                return self.read(m["id"])
            time.sleep(1)                       # one read a second, never throttled
        raise TimeoutError(f"no message for {self.address} within {timeout}s")

    def read(self, message_id: str) -> dict:
        return self._get(f"{API}/message/{message_id}", mailbox=self.address).json()

    def delete(self, message_id: str) -> None:
        """Optional and idempotent: everything expires on its own."""
        self.http.delete(f"{API}/message/{message_id}", params={"mailbox": self.address}, timeout=15)

    def download(self, attachment: dict, into: Path) -> Path:
        """Save one entry of a message's `attachments` list. The URL carries its own ?mailbox=."""
        into.mkdir(parents=True, exist_ok=True)
        target = into / attachment["filename"]
        with self.http.get("https://grabmail.io" + attachment["url"], stream=True, timeout=60) as r:
            r.raise_for_status()
            with target.open("wb") as f:
                for chunk in r.iter_content(65536):
                    f.write(chunk)
        return target

Memakainya cukup empat baris. Cetak alamatnya, pakai di mana pun sebuah alamat diminta, lalu tunggu:

percobaan pertama
from grabmail import Inbox

inbox = Inbox()
print("sign up with:", inbox.address)

message = inbox.wait_for(subject_contains="code")
print(message["subject"])
print(message["text"])          # the plain-text part; message["html"] is the HTML part or None

Versi yang sama di atas httpx, untuk asyncio

Saat sebuah skrip harus mengawasi beberapa kotak masuk sekaligus — lima pendaftaran dalam satu batch job, seorang agen yang mengelola banyak akun sekaligus — loop yang sinkron akan membuat penantiannya berurutan satu per satu. httpx.AsyncClient membuat class yang sama menjadi awaitable, dan asyncio.gather menunggu semuanya sekaligus bersamaan:

grabmail_async.py
"""grabmail_async.py — the same inbox for asyncio, on httpx."""
from __future__ import annotations

import asyncio
import time

import httpx

from grabmail import API, fresh_address


class AsyncInbox:
    def __init__(self, address: str | None = None, client: httpx.AsyncClient | None = None):
        self.address = address or fresh_address()
        self.http = client or httpx.AsyncClient(timeout=15)

    async def _get(self, url: str, **params) -> httpx.Response:
        while True:
            r = await self.http.get(url, params=params)
            if r.status_code == 429:
                await asyncio.sleep(float(r.headers.get("Retry-After", 1)))
                continue
            r.raise_for_status()
            return r

    async def list(self, limit: int = 50, before: str | None = None) -> dict:
        params = {"address": self.address, "limit": limit, **({"before": before} if before else {})}
        return (await self._get(f"{API}/mailbox", **params)).json()

    async def wait_for(self, timeout: float = 60, subject_contains: str | None = None) -> dict:
        deadline = time.monotonic() + timeout
        while time.monotonic() < deadline:
            for m in (await self.list())["messages"]:
                if not subject_contains or subject_contains.lower() in m["subject"].lower():
                    return await self.read(m["id"])
            await asyncio.sleep(1)
        raise TimeoutError(f"no message for {self.address} within {timeout}s")

    async def read(self, message_id: str) -> dict:
        return (await self._get(f"{API}/message/{message_id}", mailbox=self.address)).json()


async def main() -> None:
    inboxes = [AsyncInbox() for _ in range(5)]          # five sign-ups, waited on together
    for i in inboxes:
        print("sign up with:", i.address)
    messages = await asyncio.gather(*(i.wait_for(subject_contains="code") for i in inboxes))
    for m in messages:
        print(m["to"], "->", m["subject"])

asyncio.run(main())

Lima kotak masuk yang masing-masing di-polling sekali per detik berarti lima request per detik, jauh di bawah batas atas per klien sebesar 1200 per menit. Melewati dua puluh kotak masuk sekaligus Anda akan mencapai batas atasnya, dan cabang 429 akan mulai sleep — dan itu perilaku yang benar, bukan sebuah kegagalan.

Kotak surat yang sibuk: paging dengan before

Sebuah listing mengembalikan paling banyak 200 ringkasan. Kotak surat yang menerima lebih dari itu — misalnya sebuah alamat catch-all pada domain Anda sendiri yang menampung bounce selama sehari penuh — dibaca halaman demi halaman: sertakan nilai next dari satu response sebagai parameter before pada request berikutnya, dan berhenti saat next bernilai null.

semua pesan, berapa pun banyaknya halaman
from collections.abc import Iterator


def all_messages(inbox: Inbox) -> Iterator[dict]:
    """Every summary in the mailbox, newest first, however many pages it takes."""
    before = None
    while True:
        page = inbox.list(limit=200, before=before)
        yield from page["messages"]
        before = page["next"]
        if not before:
            return


for m in all_messages(inbox):
    print(m["date"], m["from"], m["subject"], "expires", m["expires_at"])

Cursor-nya adalah id dari pesan tertua yang sudah Anda punya, sehingga satu halaman tetap stabil meski ada email baru yang tiba di bagian atas. Mengotomasi kotak masuk lewat skrip membahas cursor ini lebih detail, beserta penjadwalan dan retensi.

Lampiran ke disk

Setiap pesan mendaftar lampirannya dengan sebuah nama file, tipe yang dideklarasikan, ukuran dalam byte, dan sebuah URL. URL-nya sudah membawa parameter ?mailbox=, jadi tinggal diambil apa adanya. Response-nya selalu application/octet-stream dengan header Content-Disposition: attachment, apa pun label yang diberikan pengirim pada filenya — tipe yang sebenarnya ada di field mime dalam JSON-nya, di mana itu berupa data, bukan sebuah instruksi.

menyimpan setiap lampiran dari sebuah pesan
from pathlib import Path

message = inbox.wait_for(subject_contains="invoice")
for a in message["attachments"]:
    print(a["filename"], a["mime"], a["size"], "bytes")
    path = inbox.download(a, into=Path("downloads") / message["id"])
    print("saved to", path)

Sebagai sebuah fixture pytest

Sebuah fixture yang mengembalikan Inbox baru memberi setiap pengujian kotak suratnya sendiri, dan itulah satu sifat paling penting dari sebuah pengujian email: tidak ada run yang bisa membaca pesan dari run sebelumnya. Pola ekstraksinya dipatok pada kata-kata template, bukan pada “enam digit”, untuk alasan-alasan yang dijelaskan di kode OTP dalam pengujian otomatis.

conftest.py
# conftest.py
import pytest

from grabmail import Inbox


@pytest.fixture
def inbox() -> Inbox:
    """A brand-new mailbox for this test, and this test only."""
    return Inbox()
test_signup.py
# test_signup.py
import re

CODE = re.compile(r"code is\D{0,12}(\d{6})", re.I)      # anchored on YOUR template's wording


def test_signup_confirmation(client, inbox):
    client.post("/signup", data={"email": inbox.address, "password": "hunter2hunter2"})

    message = inbox.wait_for(subject_contains="confirm")
    code = CODE.search(f"{message.get('text') or ''} {message.get('html') or ''}").group(1)

    response = client.post("/confirm", data={"email": inbox.address, "code": code})
    assert response.status_code == 200

Fixture client adalah apa pun yang disediakan framework Anda — test client milik Flask, milik Django, sebuah httpx.Client yang diarahkan ke server yang sedang berjalan. Sisi kotak suratnya tidak peduli soal itu. Menjalankan suite-nya pada runner CI menambahkan sebuah aturan egress dan sebuah timeout, keduanya dibahas di panduan GitHub Actions.

Kesalahan yang muncul pertama kali berjalan tanpa pengawasan

Tidak satu pun dari ini rusak di laptop. Semuanya rusak pada suatu malam Selasa di dalam sebuah cron job.

GejalaPenyebabPerbaikan
Lolos setiap saat, bahkan saat pengirimnya rusakAlamat yang sama di setiap run; polling pertama menemukan pesan dari run sebelumnya.fresh_address() per run. Inilah yang benar-benar penting.
429 di dalam log, lalu crashLoop tanpa sleep, atau dua skrip yang melakukan polling ke satu alamat yang sama.Satu kali baca per detik per alamat; sleep sesuai Retry-After; satu alamat per skrip.
Timeout di hari yang lambat, lolos saat diulangHitungan retry alih-alih batas waktu, atau batas waktu yang lebih singkat daripada antrean pengirimnya.Batas waktu time.monotonic(), enam puluh detik untuk email transaksional.
404 dari /mailboxDomainnya tidak di-hosting di sini — salah ketik, atau domain Anda sendiri yang belum punya record MX.Periksa alamatnya; untuk domain Anda sendiri, periksa apakah MX-nya menunjuk ke smtp.grabmail.io.
Membaca pesan yang salahMengambil pesan terbaru padahal alurnya mengirim dua pesan.Lakukan filter dengan subject_contains atau from_contains.
Berfungsi selama seminggu, lalu 404 pada sebuah pesanSebuah id pesan yang tersimpan sudah lebih tua dari 5 hari.Tidak ada yang bertahan melewati 5 hari. Ambil ulang, jangan di-cache.

Sebelum Anda menganggapnya selesai

  • Alamat baru per run, per pengujian, atau per agen — jangan pernah konstanta.
  • Batas waktu monotonic; satu kali baca per detik; 429 ditangani dengan sleep, jangan pernah dilempar sebagai error.
  • Filter pada subjek atau pengirim saat sebuah alur mengirim lebih dari satu pesan.
  • Bagian teksnya di-parse duluan, dengan pola yang dipatok pada kata-kata Anda sendiri.
  • Lampiran diperlakukan sebagai berkas yang tidak tepercaya, disimpan di bawah id pesannya.
  • Tidak ada id pesan yang di-cache melewati hari; tidak ada apa pun di sini yang bertahan lebih lama dari 5 hari.

Itulah keseluruhan client-nya. Modul yang sama dalam TypeScript, untuk Node, Deno, dan Bun, ada di panduan Node.js; bentuk request dan response-nya, lengkap dengan setiap kode status, ada di referensi API, dan ada sebuah dokumen OpenAPI 3.1 untuk siapa pun yang lebih suka men-generate client-nya daripada menulisnya sendiri.

Pertanyaan

Apakah saya butuh kunci API?

Tidak. Domain publiknya tidak butuh kunci, tidak butuh akun, dan tidak butuh header apa pun. Hanya kumpulan domain berbayar yang dijaga agar tidak masuk daftar blokir email sekali pakai yang memakai header Authorization: Bearer, dan selain itu kodenya di atas identik saja.

Bisakah saya men-generate client-nya dari dokumen OpenAPI saja?

Bisa — /openapi.json adalah OpenAPI 3.1 dan generator apa pun akan menghasilkan ketiga panggilan itu. Anda tetap akan membutuhkan loop batas waktu dari panduan ini di sekeliling panggilan listing-nya, karena tidak ada generator yang menuliskan itu untuk Anda.

Berapa banyak kotak surat yang bisa di-polling satu skrip sekaligus?

Dua puluh, dengan nyaman: batas per alamat adalah satu kali baca per detik dan batas atas per klien adalah 1200 request per menit, yaitu dua puluh alamat yang di-polling sekali per detik. Versi async di atas mematuhi keduanya, dan melewati batas atasnya, ia akan sleep saat menerima 429, bukan gagal.

Bisakah saya memakai domain saya sendiri dari Python?

Bisa, tanpa perubahan apa pun pada kode selain konstanta DOMAIN. Satu record MX yang menunjuk ke smtp.grabmail.io dan setiap alamat pada domain itu menjadi kotak surat yang bisa dibaca modul yang sama — pengaturannya ada di sini. Ini jawaban yang tepat saat aplikasi Anda menolak domain sekali pakai publik.

Apakah kotak suratnya bersifat privat selama dipakai skrip saya?

Tidak. Siapa pun yang tahu alamatnya bisa membacanya, baik di domain publik maupun di domain Anda sendiri. Alamat acak yang menyimpan satu kode konfirmasi selama beberapa detik itu baik-baik saja; skrip yang mengarahkan email pelanggan sungguhan ke sana, itu tidak baik.

Apakah ada sesuatu untuk agen AI, bukan skrip?

Ada sebuah server MCP di origin yang sama, tanpa kunci, dengan tool wait_for_message yang menjaga panggilannya tetap terbuka sampai emailnya tiba — bentuk yang dibutuhkan seorang agen, karena setiap polling membebankan token baginya. Kotak masuk yang bisa dibaca agen AI membahas itu.

Coba selagi masih segar

Alamat hanya perlu satu klik, tanpa akun dan tanpa kartu. Semua yang ada di panduan ini langsung berfungsi dengannya.

Selamat datang kembali

Kotak surat dan domain Anda, di satu tempat.