API и автоматизация

Python: приём писем через API одноразового ящика, без ключа

Шестьдесят строк Python, одна зависимость и никакого API-ключа: адрес, который вы придумываете сами, ожидание с дедлайном, письмо в виде dict. Вот модуль на requests, та же логика на httpx для asyncio, постраничное чтение загруженного ящика, сохранение вложения, фикстура pytest и шесть ошибок, которые проявляются с первого же запуска без присмотра.

  • Средний
  • 23 мин на чтение
Свободно свёрнутая серая трубка, из открытого конца которой синий конверт выезжает на маленький серый поднос

API глазами Python

На серверной стороне ничего не нужно устанавливать и не к чему аутентифицироваться: ящик на публичном домене может прочитать любой, кто знает его адрес, — по обычному HTTPS, в виде JSON. Вся поверхность API — это три вызова:

GET /api/v1/mailbox?address=…
Всё, что ждёт по адресу, от новых к старым, в виде списка сводок. Пустой ящик — это 200 с count: 0, а не 404. limit ограничивает один ответ (1–200, по умолчанию 50), а before листает дальше.
GET /api/v1/message/{id}?mailbox=…
Одно письмо целиком: отправитель, получатель, тема, дата, часть с обычным текстом, HTML-часть (или null) и список вложений с готовым URL у каждого.
DELETE /api/v1/message/{id}?mailbox=…
Удаляет письмо сразу, а не через 5 дней. Идемпотентно: повторное удаление всё равно отвечает 200.

Список писем выглядит так. alias — это второй адрес на отдельном домене, который доставляет почту в тот же ящик, но не может использоваться для его чтения, — тот, что стоит отдавать сайту, когда не хочется, чтобы он мог открыть ящик.

GET /api/v1/mailbox — ответ
{
  "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" }
  ]
}

Модуль

Один файл, один класс, и requests — единственная зависимость. Он намеренно скучный: сессия, цикл с дедлайном и единственный по-настоящему правильный вид повтора — сон при 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

Использовать его — четыре строки. Выведите адрес, используйте его там, где требуется адрес, и ждите:

первый запуск
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

То же самое на httpx, для asyncio

Когда скрипту нужно следить за несколькими ящиками одновременно — пять регистраций в одном пакетном задании, агент, жонглирующий аккаунтами, — синхронный цикл превращает ожидания в последовательность. httpx.AsyncClient делает тот же класс awaitable, а asyncio.gather ждёт их все вместе:

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())

Пять ящиков, каждый опрашиваемый раз в секунду, — это пять запросов в секунду, с большим запасом до потолка в 1200 в минуту на клиента. При двадцати с лишним одновременных ящиках вы упрётесь в этот потолок, и ветка 429 начнёт засыпать — это правильное поведение, а не сбой.

Загруженный ящик: постраничное чтение через before

Список писем возвращает не больше 200 сводок за раз. Ящик, который получает больше, — например, catch-all-адрес на вашем собственном домене, собирающий возвраты за целый день, — читается постранично: передайте значение next из одного ответа как параметр before в следующем запросе и остановитесь, когда next станет null.

все письма, сколько бы страниц ни было
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"])

Курсор — это id самого старого письма, которое у вас уже есть, поэтому страница остаётся стабильной, даже пока сверху приходит новая почта. В руководстве «Автоматизация ящика из скрипта» курсор разобран подробнее, вместе с расписанием и сроком хранения.

Вложения на диск

У каждого письма перечислены вложения с именем файла, заявленным типом, размером в байтах и URL. URL уже содержит параметр ?mailbox=, так что его можно запрашивать как есть. Ответ всегда приходит как application/octet-stream с заголовком Content-Disposition: attachment, независимо от того, каким типом файл пометил отправитель, — настоящий тип лежит в поле mime в JSON, где он остаётся данными, а не инструкцией.

сохранить все вложения письма
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)

В виде фикстуры pytest

Фикстура, возвращающая новый Inbox, даёт каждому тесту собственный ящик, а это самое важное свойство почтового теста: ни один прогон никогда не сможет прочитать письмо из другого прогона. Шаблон извлечения привязан к формулировке письма, а не к принципу «шесть цифр», по причинам, изложенным в «OTP-коды в автотестах».

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

Фикстура client — это что угодно, что предоставляет ваш фреймворк: тестовый клиент Flask, тестовый клиент Django, httpx.Client, направленный на запущенный сервер. Почтовой стороне это безразлично. Запуск набора тестов на раннере CI добавляет правило для исходящего трафика и таймаут — оба вопроса разобраны в руководстве по GitHub Actions.

Ошибки, которые проявляются с первого же запуска без присмотра

Ни одна из них не ломается на ноутбуке. Все они ломаются во вторник ночью в cron-задании.

СимптомПричинаРешение
Проходит каждый раз, даже когда отправитель сломанОдин и тот же адрес при каждом прогоне; первый же опрос находит письмо из прошлого прогона.fresh_address() на каждый прогон. Вот это действительно важно.
429 в логе, затем падениеЦикл без sleep, либо два скрипта опрашивают один адрес.Одно чтение в секунду на адрес; сон на время Retry-After; один адрес на скрипт.
Истекает по таймауту в неудачный день, проходит при повтореСчётчик попыток вместо дедлайна, либо дедлайн короче очереди отправителя.Дедлайн на time.monotonic(), шестьдесят секунд для транзакционного письма.
404 от /mailboxЭтот домен здесь не обслуживается — опечатка, либо у собственного домена отсутствует MX.Проверьте адрес; для собственного домена проверьте, что MX указывает на smtp.grabmail.io.
Читает не то письмоВзял самое новое письмо, когда сценарий отправил два.Фильтруйте через subject_contains или from_contains.
Работает неделю, а потом 404 на письмеСохранённый id письма старше 5 дней.Ничто не переживает 5 дней. Запрашивайте заново, а не кэшируйте.

Прежде чем считать задачу закрытой

  • Новый адрес на каждый прогон, тест или агента — никогда не константа.
  • Дедлайн на монотонных часах; одно чтение в секунду; 429 засыпает, а не выбрасывает исключение.
  • Фильтр по теме или отправителю, когда сценарий отправляет больше одного письма.
  • В первую очередь разбирается текстовая часть, с шаблоном, привязанным к вашей формулировке.
  • Вложения обрабатываются как недоверенные файлы, сохраняются под id письма.
  • Никакого кэширования id письма на несколько дней; здесь ничто не живёт дольше 5 дней.

Вот и весь клиент. Тот же модуль на TypeScript, для Node, Deno и Bun, — в руководстве по Node.js; форматы запросов и ответов, со всеми кодами состояния, — в справочнике по API, а ещё есть документ OpenAPI 3.1 для тех, кто предпочитает сгенерировать клиент, а не писать его вручную.

Вопросы

Нужен ли API-ключ?

Нет. Публичные домены не требуют ни ключа, ни аккаунта, ни заголовка. Заголовок Authorization: Bearer используется только для платного пула доменов, которые держат подальше от чёрных списков одноразовой почты, а в остальном код выше для него точно такой же.

Можно ли вместо этого сгенерировать клиент из документа OpenAPI?

Да — /openapi.json это OpenAPI 3.1, и любой генератор произведёт эти три вызова. Но цикл с дедлайном из этого руководства вокруг вызова списка писем всё равно понадобится, потому что ни один генератор не напишет его за вас.

Сколько ящиков может опрашивать один скрипт одновременно?

Двадцать, с запасом: лимит на адрес — одно чтение в секунду, а потолок на клиента — 1200 запросов в минуту, то есть двадцать адресов, опрашиваемых раз в секунду. Асинхронная версия выше соблюдает оба ограничения, а при превышении потолка засыпает на 429, а не падает.

Можно ли использовать собственный домен из Python?

Да, без каких-либо изменений в коде, кроме константы DOMAIN. Одна MX-запись, указывающая на smtp.grabmail.io, — и любой адрес на домене становится ящиком, который читает тот же самый модуль — настройка здесь. Это правильный ответ, когда ваше приложение отказывает публичным одноразовым доменам.

Приватен ли ящик, пока им пользуется мой скрипт?

Нет. Прочитать его может любой, кто знает адрес, — и на публичном домене, и на вашем собственном. Для случайного адреса, хранящего один код подтверждения несколько секунд, это нормально; для скрипта, направляющего туда настоящую почту клиентов, — нет.

Есть ли что-то для AI-агента, а не для скрипта?

Есть MCP-сервер на том же домене, без ключа, чей инструмент wait_for_message держит вызов открытым, пока не придёт почта, — именно такая форма и нужна агенту, ведь каждый опрос стоит ему токенов. Подробнее в «Ящик, который может прочитать AI-агент».

Читать дальше

Попробуйте, пока свежо

Адрес — это один клик, без аккаунта и карты. Всё из этого руководства сразу заработает на нём.

С возвращением

Ваши ящики и ваши домены в одном месте.