mailgw-python

Async Python client for the mail.gw API.

It helps you:

• list available domains
• create and manage temporary mail accounts
• read mailbox messages
• listen for new messages in real time (Mercure/SSE webhook-like stream)
• fetch raw source of a message

Requirements

• Python 3.10+

Installation

Poetry

poetry add mailgw-python

pip (from PyPI)

pip install mailgw-python

pip (specific version)

pip install "mailgw-python==0.1.0"

pip (from TestPyPI)

pip install --index-url https://test.pypi.org/simple/ --extra-index-url https://pypi.org/simple mailgw-python

pip (from GitHub)

pip install "git+https://github.com/<your-user>/<your-repo>.git"

pip (from local source)

pip install .

Client Lifecycle

You can create the client directly:

client = MailGWClient()

async with MailGWClient() is also supported and is usually preferred, because the client connection closes automatically.

Quick Start

import asyncio
import uuid

from mailgw import MailGWClient


async def main():
    client = MailGWClient()

    domains = await client.domains.get()
    domain = domains.hydra_member[0].domain

    address = f"{uuid.uuid4()}@{domain}"
    password = str(uuid.uuid4())

    account = await client.accounts.create(address, password)
    token_data = await client.token.get_token(address, password)
    token = token_data.token

    messages = await client.messages.reads(token)
    print(f"Mailbox has {messages.total_items} messages")

    await client.accounts.delete_by_id(account.id, token)


if __name__ == "__main__":
    asyncio.run(main())

API Usage Examples

1) Get all domains

import asyncio

from mailgw import MailGWClient


async def main():
    client = MailGWClient()

    domains = await client.domains.get(page=1)
    for item in domains.hydra_member or []:
        print(item.domain, item.is_active)


if __name__ == "__main__":
    asyncio.run(main())

2) Get one domain by ID

import asyncio

from mailgw import MailGWClient


async def main():
    client = MailGWClient()
    domain = await client.domains.get_by_id("your-domain-id")
    print(domain)


if __name__ == "__main__":
    asyncio.run(main())

3) Create one account

import asyncio
import uuid

from mailgw import MailGWClient


async def main():
    client = MailGWClient()

    domains = await client.domains.get()
    domain = domains.hydra_member[0].domain

    address = f"{uuid.uuid4()}@{domain}"
    password = str(uuid.uuid4())

    account = await client.accounts.create(address, password)
    print("Account ID:", account.id)
    print("Address:", account.address)


if __name__ == "__main__":
    asyncio.run(main())

4) Authenticate and get token

import asyncio

from mailgw import MailGWClient


async def main():
    client = MailGWClient()
    token_data = await client.token.get_token(
        address="user@your-domain.tld",
        password="your-password",
    )
    print("Token:", token_data.token)


if __name__ == "__main__":
    asyncio.run(main())

5) Get current account (/me)

import asyncio

from mailgw import MailGWClient


async def main():
    client = MailGWClient()
    me = await client.accounts.me("your-jwt-token")
    print(me.id, me.address)


if __name__ == "__main__":
    asyncio.run(main())

6) Get all messages

import asyncio

from mailgw import MailGWClient


async def main():
    client = MailGWClient()
    messages = await client.messages.reads("your-jwt-token")
    print("Total:", messages.total_items)
    for message in messages.messages:
        print(message.id, message.subject)


if __name__ == "__main__":
    asyncio.run(main())

7) Get one message by ID

import asyncio

from mailgw import MailGWClient


async def main():
    client = MailGWClient()
    message = await client.messages.read_by_id(
        message_id="your-message-id",
        token="your-jwt-token",
    )
    print(message.subject)
    print(message.text)


if __name__ == "__main__":
    asyncio.run(main())

8) Get raw message source

import asyncio

from mailgw import MailGWClient


async def main():
    client = MailGWClient()
    source = await client.sources.get_source_message(
        message_id="your-message-id",
        token="your-jwt-token",
    )
    print(source.data)


if __name__ == "__main__":
    asyncio.run(main())

9) Receive messages via webhook-like stream (SSE)

listen_messages subscribes to the Mail.gw Mercure topic for your account and yields new Message events.

import asyncio
import httpx

from mailgw import MailGWClient


async def main():
    account_id = "your-account-id"
    token = "your-jwt-token"
    client = MailGWClient()

    try:
        async for event in client.messages.listen_messages(account_id, token):
            full_message = await client.messages.read_by_id(event.id, token)
            print("New message:", full_message.subject)
            break
    except httpx.ReadTimeout:
        print("No new messages within timeout window")


if __name__ == "__main__":
    asyncio.run(main())

10) Forward new messages to your own webhook URL

If you need classic HTTP webhooks, subscribe via SSE and forward each event to your endpoint.

import asyncio
import httpx

from mailgw import MailGWClient


async def main():
    account_id = "your-account-id"
    token = "your-jwt-token"
    webhook_url = "https://your-service.example/webhooks/mailgw"
    client = MailGWClient()

    async with httpx.AsyncClient(timeout=10.0) as webhook_client:
        async for event in client.messages.listen_messages(account_id, token):
            message = await client.messages.read_by_id(event.id, token)
            await webhook_client.post(
                webhook_url,
                json={
                    "id": message.id,
                    "subject": message.subject,
                    "intro": message.intro,
                    "text": message.text,
                    "created_at": message.created_at,
                },
            )


if __name__ == "__main__":
    asyncio.run(main())

Resource Methods

client.domains

• get(page: int = 1)
• get_by_id(domain_id: str)

client.accounts

• create(address: str, password: str)
• me(token: str)
• read_by_id(account_id: str, token: str)
• delete_by_id(account_id: str, token: str)

client.token

• get_token(address: str, password: str)

client.messages

• reads(token: str)
• read_by_id(message_id: str, token: str)
• listen_messages(account_id: str, token: str, read_timeout: float | None = 120.0)

client.sources

• get_source_message(message_id: str, token: str)

Error Handling

All API errors raise mailgw.exceptions.APIError.

import asyncio

from mailgw import MailGWClient
from mailgw.exceptions import APIError


async def main():
    client = MailGWClient()
    try:
        await client.accounts.me("invalid-token")
    except APIError as e:
        print(f"Status: {e.status_code}")
        print(f"Message: {e.message}")


if __name__ == "__main__":
    asyncio.run(main())

Notes

• You can use client = MailGWClient() directly.
• async with MailGWClient() is usually preferred because the client connection is closed automatically.
• You can override request timeout: MailGWClient(timeout=5.0).
• Real-time updates use Mercure Server-Sent Events (SSE).