Python Email SDK: Build an Agent That Reads and Replies

The Robotomail Python email SDK gives your agent a real mailbox in a few lines: a synchronous Robotomail client for scripts and request handlers, an AsyncRobotomail client for asyncio apps, and dictionary responses so you can log a payload without wrestling with model classes. This tutorial builds a complete loop, list mailboxes, send a message with a friendly From name, receive an inbound reply over a webhook, and send a human-approved response.

If you want the raw HTTP background first, we already covered sending mail from Python with plain requests and the same flow side by side in Node. This post is about client ergonomics and the receive half, which is the part most SMTP tutorials skip.

Install the Python email SDK

The SDK targets Python 3.10 and newer. The five Robotomail SDKs are public MIT releases on GitHub. Package-registry publishing is not finished, so install from the tested v0.1.0 tag rather than guessing at a PyPI name:

pip install "robotomail @ git+https://github.com/robotomail/robotomail-python.git@v0.1.0"

That command needs git on the machine, which matters in slim container images. Pin the tag in your lockfile or requirements file so CI builds stay reproducible.

Authentication is a server-side API key in the environment:

export ROBOTOMAIL_API_KEY="rm_your_api_key"

The client reads ROBOTOMAIL_API_KEY automatically when you construct it with no arguments. Existing mailbox-scoped keys work fine for sending and reading in one mailbox; account-level operations and attachment uploads need a full-access key. The scope rules are in the authentication docs.

First request: list the account's mailboxes

Start with the mailbox the account already has. Use the client as a context manager so the underlying HTTP connection pool closes cleanly:

from robotomail import Robotomail

with Robotomail() as mail:  # reads ROBOTOMAIL_API_KEY
    result = mail.list_mailboxes()
    print([box["fullAddress"] for box in result["mailboxes"]])

Responses are plain dictionaries with camelCase keys, matching the REST API in the HTTP reference. That means print(result) is a useful debugging step, and you can hand the payload straight to json.dumps in a log line.

Copy the mailbox id you want to work with into your config. If you want a dedicated address for this agent instead of reusing the default one, the CLI is the fastest route:

npx @robotomail/cli mailbox create shopping-agent

Send with a display name

Sends need three fields: to, subject, and bodyText. Add displayName when you want the recipient's client to show something friendlier than the raw local part, and attachments when you have uploaded attachment IDs.

import os
from robotomail import Robotomail

MAILBOX_ID = os.environ["ROBOTOMAIL_MAILBOX_ID"]  # e.g. mbx_shopping

with Robotomail() as mail:
    sent = mail.send_message(
        MAILBOX_ID,
        to=["vendor@example.com"],
        subject="Order #A-4521 delivery update?",
        body_text="Hi, checking on the status of order A-4521.",
        display_name="Acme Procurement Bot",
    )
    print(sent)

Two operational notes we have learned the hard way. First, the SDK does not retry sends for you. A timeout is ambiguous, not a failure, so list the mailbox's messages and check whether the send landed before you send again. Blind retries are how agents send the same chase email three times. Second, keep the display name honest about the fact that a bot is writing. Recipients reply differently when they know, and you get fewer manual-review escalations later.

Receiving is a webhook, not a stream

Here is the part that trips people up when they move from a send-only library to a full mailbox. An event stream is a nice debugging tool, but streaming alone does not schedule your agent.

A stream only delivers while your process is connected. If the connection drops, and it will drop, the SDK does not reconnect for you, so messages that arrived during the gap are not pushed to you again. A long-lived SSE loop also couples your agent's availability to its inbox: no process running, no work queued. Deploys, scale-to-zero, and OOM kills all become silent message loss.

Webhooks invert that. Robotomail delivers inbound mail to an HTTP endpoint you register as its own resource, your handler writes the message into your own queue or table, and the agent runs when the queue has work. Registering a webhook is covered in the webhooks concept guide. Use the stream for a terminal window while you develop, and the webhook for anything you page on.

The inbound payload is { event, timestamp, data } with snake_case data fields:

import json
from flask import Flask, request

app = Flask(__name__)

@app.post("/hooks/robotomail")
def inbound():
    payload = request.get_json(force=True)
    if payload["event"] != "message.received":
        return "", 204

    data = payload["data"]
    enqueue_for_agent({
        "message_id": data["message_id"],      # RFC Message-ID
        "mailbox_id": data["mailbox_id"],
        "from": data["from"],
        "subject": data["subject"],
        "body_text": data["body_text"],
        "thread_id": data["thread_id"],
        "received_at": data["received_at"],
    })
    return "", 202

Return a 2xx fast and do the LLM work out of band. Keep message_id around: that RFC Message-ID is what you pass as inReplyTo so your reply threads correctly in the recipient's client.

The same workflow with AsyncRobotomail

If your agent already runs under asyncio, FastAPI, or a framework-driven event loop, swap the client and await the calls. The method names and dictionary shapes are identical:

import asyncio
from robotomail import AsyncRobotomail

async def main() -> None:
    async with AsyncRobotomail() as mail:
        result = await mail.list_mailboxes()
        print([box["fullAddress"] for box in result["mailboxes"]])

asyncio.run(main())

Construct one client per process and reuse it, the same way you would a database pool. Creating an AsyncRobotomail inside a hot request handler throws away connection reuse. The async with form is right for scripts and for FastAPI lifespan setup; in long-running services, open the client at startup and close it at shutdown.

Gate the reply behind a human

For anything that commits money, promises a date, or touches a customer relationship, draft and then wait. The send call stays the same, you just move it behind an approval record.

from robotomail import Robotomail

def draft_reply(inbound: dict) -> dict:
    body = generate_reply(inbound["body_text"])  # your model call
    return {
        "mailbox_id": inbound["mailbox_id"],
        "to": [inbound["from"]],
        "subject": f"Re: {inbound['subject']}",
        "body_text": body,
        "in_reply_to": inbound["message_id"],
    }

def send_approved(draft: dict) -> dict:
    with Robotomail() as mail:
        return mail.send_message(
            draft["mailbox_id"],
            to=draft["to"],
            subject=draft["subject"],
            body_text=draft["body_text"],
            display_name="Acme Procurement Bot",
            in_reply_to=draft["in_reply_to"],
        )

Store the draft, show it in whatever review surface your team already uses, and call send_approved only when someone clicks approve. Because in_reply_to is captured at draft time, a reply approved an hour later still lands in the right thread. The end-to-end receive-then-reply flow, including threading rules, is written up in the receive and reply guide.

A reasonable progression: human review on everything for the first week, then auto-send only the intents you have measured, like status acknowledgements, and keep review on anything that quotes a price.

Troubleshooting

ModuleNotFoundError: robotomail after pip install. Usually a second interpreter. Run python -m pip install ... with the exact interpreter that runs your app, and check the virtualenv is active.

The install command fails with a git error. The package installs from a GitHub tag, so the build environment needs git. Add it to your Dockerfile build stage or vendor a wheel you build yourself.

401 on construction or first call. The client reads ROBOTOMAIL_API_KEY. Shells that export the variable in an interactive profile do not pass it to systemd units or CI jobs. Echo it in the same process that fails.

403 on an account-level call with a working send key. Mailbox-scoped keys cannot list account resources or upload attachments. Issue a full-access key for those paths.

A send times out and you do not know if it went out. Do not loop. List the mailbox's messages, look for your subject and recipient, and only resend if it is absent.

Replies arrive as new threads in Gmail. You passed an internal identifier instead of the RFC Message-ID. Use the message_id value from the inbound payload.

RuntimeError: Event loop is closed on shutdown. An AsyncRobotomail outlived its loop. Close it inside the same loop that created it, usually via async with or a framework lifespan hook.

FAQ

Is the Python email SDK on PyPI?

Not as of this writing. Install from the tested v0.1.0 GitHub tag with the pip command above, and check the SDK docs for current packaging status before you change your install line.

Can one client handle several mailboxes?

Yes. The mailbox ID is a parameter on message calls, not on the client, so one Robotomail instance can serve a whole fleet of agent mailboxes, provided the API key's scope covers them.

Do I need the async client?

Only if you are already in an event loop. The synchronous client is simpler for scripts, cron jobs, and WSGI handlers. Mixing the sync client into asyncio code will block your loop, so pick by runtime, not by preference.

How do I handle attachments?

Upload the file to get an attachment ID, then pass the IDs in attachments on the send call. Attachment uploads require a full-access key.

Ready to point the SDK at a real inbox? Create a mailbox and start with Robotomail, or read the Python SDK reference first.