Python SDK

The Python SDK provides both synchronous (MailsClient) and asynchronous (AsyncMailsClient) clients with full type hints and 12 methods covering the entire mails-agent API.

Installation

pip install mails-agent

Requires Python 3.8+. The async client requires aiohttp (installed automatically).

Quick Start

from mails_agent import MailsClient

client = MailsClient(
    api_url="https://mails-worker.genedai.workers.dev",
    token="mk_your_api_key",
    mailbox="[email protected]",
    hosted=True,
)

# Send an email
client.send(
    to=["[email protected]"],
    subject="Hello from Python",
    text="This is the body"
)

# Check inbox
emails = client.get_inbox(limit=5)
for email in emails:
    print(f"{email.from_address}: {email.subject}")

# Wait for a verification code
result = client.wait_for_code(timeout=60)
if result:
    print(f"Code: {result.code}")

Client Initialization

Sync Client

from mails_agent import MailsClient

client = MailsClient(
    api_url="https://mails-worker.genedai.workers.dev",    # required: API base URL
    token="mk_your_api_key",                               # required: API key
    mailbox="[email protected]",                            # required: your mailbox address
    timeout=60.0,                                           # optional: request timeout in seconds
    hosted=True,                                            # optional: use /v1/* routes (hosted mode)
)

Async Client

from mails_agent import AsyncMailsClient

client = AsyncMailsClient(
    api_url="https://mails-worker.genedai.workers.dev",
    token="mk_your_api_key",
    mailbox="[email protected]",
    timeout=60.0,
    hosted=True,
)

# Use with async/await
async def main():
    emails = await client.get_inbox()
    result = await client.wait_for_code(timeout=60)
    await client.close()  # or use as async context manager

# As async context manager
async with AsyncMailsClient(api_url="...", token="...", mailbox="...") as client:
    emails = await client.get_inbox()

Methods Reference

All 12 methods are available on both MailsClient (sync) and AsyncMailsClient (async). The async versions have the same signature but return awaitables.

send()

Send an email.

result = client.send(
    to=["[email protected]"],         # list of recipients (or single string)
    subject="Hello",                  # subject line
    text="Plain text body",          # plain text (optional if html provided)
    html="<h1>Hello</h1>",            # HTML body (optional if text provided)
    reply_to="[email protected]",    # optional reply-to address
    headers={"X-Custom": "value"},   # optional custom headers
    attachments=[                     # optional file attachments
        {"filename": "report.pdf", "content": b"...", "content_type": "application/pdf"}
    ],
)
# Returns: SendResult(id="a1b2c3d4-...", provider_id="re_abc123")

get_inbox()

List emails in the mailbox.

emails = client.get_inbox(
    limit=20,            # max results (default 20)
    offset=0,            # pagination offset
)
# Returns: [{"id": "msg_...", "from_address": "...", "subject": "...", ...}, ...]

search()

Semantic search across emails.

results = client.search(
    query="verification code from GitHub",
    limit=10,
)
# Returns: [Email(...), ...] — list of Email objects

get_email()

Get a single email by ID.

email = client.get_email(email_id="msg_abc123")
# Returns: Email(id="msg_abc123", from_address="...", subject="...", body_text="...", ...)

delete_email()

Delete a single email by ID.

result = client.delete_email(email_id="msg_abc123")
# Returns: True if deleted, False if not found

wait_for_code()

Block until a verification code arrives. Polls the inbox and extracts codes automatically (supports EN, ZH, JA, KO).

result = client.wait_for_code(
    timeout=60,              # max seconds to wait
    since="2026-04-01T12:00:00Z",  # optional: only consider emails after this time
)
# Returns: VerificationCode(code="482916", from_address="[email protected]", subject="...", id="...", received_at="...")
# Returns: None on timeout

    
get_threads()
    
List email threads.
    
threads = client.get_threads(
    limit=20,
    offset=0,
)
# Returns: [{"thread_id": "thr_...", "subject": "...", "message_count": 3, ...}, ...]

    
get_thread()
    
Get all messages in a thread.
    
emails = client.get_thread(thread_id="thr_abc123")
# Returns: [Email(...), Email(...), ...] — list of Email objects in chronological order

    
get_attachment()
    
Download an email attachment.
    
data = client.get_attachment(attachment_id="att_def456")
# Returns: bytes (raw file content)

    
get_me()
    
Get information about the authenticated key and mailbox.
    
info = client.get_me()
# Returns: MeInfo(worker="mails-worker", mailbox="[email protected]", send=True)

    
get_stats()
    
Get mailbox statistics.
    
stats = client.get_stats()
# Returns: {"mailbox": "...", "total_emails": 142, "inbound": 97, "outbound": 45, "emails_this_month": 42}

    
extract()
    
Extract structured data from an email using AI. The type parameter selects the extraction template.
    
data = client.extract(
    message_id="msg_abc123",
    type="order",   # one of: order, shipping, calendar, receipt, code
)
# Returns: {"order_id": "ORD-12345", "total": 99.99, "items": [...]}

    
Error Handling
    
from mails_agent import MailsClient, MailsError, AuthenticationError, NotFoundError

client = MailsClient(api_key="mk_your_api_key")

try:
    email = client.get_email(id="msg_nonexistent")
except AuthenticationError:
    print("Invalid API key")
except NotFoundError:
    print("Email not found")
except MailsError as e:
    print(f"API error: {e.status_code} {e.message}")

    
Common Patterns

    
Agent Self-Signup Flow
    
from mails_agent import MailsClient

client = MailsClient(api_url="...", token="mk_your_api_key", mailbox="[email protected]", hosted=True)

# 1. Sign up on a website with your mailbox address
#    (your agent fills in the form with [email protected])

# 2. Wait for the verification code
result = client.wait_for_code(timeout=120)
if result:
    print(f"Got code: {result.code}")

# 3. Submit the code to complete signup
#    (your agent enters the code on the website)

    
Process Incoming Emails
    
from mails_agent import MailsClient

client = MailsClient(api_url="...", token="mk_your_api_key", mailbox="[email protected]", hosted=True)

# Poll for new emails
emails = client.get_inbox()
for email in emails:
    print(f"From: {email.from_address}")
    print(f"Subject: {email.subject}")
    print(f"Body: {email.body_text[:200]}")

    # Reply
    client.send(
        to=[email.from_address],
        subject=f"Re: {email.subject}",
        text="Thanks for your email!",
    )

    
Extract Data from Order Confirmation
    
from mails_agent import MailsClient

client = MailsClient(api_url="...", token="mk_your_api_key", mailbox="[email protected]", hosted=True)

# Search for order confirmations
results = client.search(query="order confirmation")
if results:
    data = client.extract(
        email_id=results[0].id,
        extract_type="order",
    )
    print(f"Order {data['extraction']['order_id']}: ${data['extraction']['total']}")

    
Constructor Parameters
    
Both MailsClient and AsyncMailsClient accept the same constructor parameters:
    

      
Parameter
Required
Description
      
api_url
Yes
Worker API base URL (e.g. https://mails-worker.genedai.workers.dev)
      
token
Yes
API key or worker token for authentication
      
mailbox
Yes
Your email address (e.g. [email protected])
      
timeout
No
Request timeout in seconds (default: 60)
      
hosted
No
When True, use /v1/* routes (hosted mode). Default: False