Pydantic AI Email Integration: Typed Send and Receive

A Pydantic AI email integration has three moving parts: a mailbox with a real address that the agent owns, a typed tool that posts outbound messages to the mail API, and a webhook endpoint that turns each inbound message into an agent run. Pydantic AI is a good fit here because everything email touches, the tool arguments, the inbound payload, and the reply the model produces, benefits from being a validated model instead of a loose dict.

This guide shows the full loop with Robotomail: provision the mailbox, register a webhook, define send_email as a typed tool, validate the inbound JSON, and constrain the agent's reply with output_type so it cannot send half-formed mail.

The shape of the integration

inbound email -> webhook POST -> InboundEvent (validated)
             -> agent.run(...) with output_type=DraftReply
             -> send_email tool -> POST /v1/mailboxes/{id}/messages

Two rules keep this stable in production:

1. The webhook handler validates and acknowledges fast, then runs the agent out of band. Agent runs take seconds to minutes; webhook delivery does not wait that long.
2. Sending is a tool call with a Pydantic model as its argument schema. That gives you address validation, length caps, and a single choke point for policy checks before anything leaves the building.

Step 1: give the agent a mailbox

Create the mailbox first so you have an ID to pass into the agent as a dependency. CLI:

npx @robotomail/cli mailbox create support-triage

Or over the API:

curl -X POST https://api.robotomail.com/v1/mailboxes \
  -H "Authorization: Bearer rm_your_api_key" \
  -H "Content-Type: application/json" \
  -d '{"address": "support-triage"}'

That returns a mailbox with an address like support-triage@robotomail.co, which can send and receive immediately. Use your own domain later by adding a domainId when you create the mailbox. If you are still deciding how identity should map to agents, one mailbox per agent instance is usually the right default, and the broader tradeoffs are covered in our guide on how to give your AI agent an email address.

Webhooks are registered as their own resource against your account, pointed at a URL you control. See the webhooks concept docs for the registration call and signature verification details.

Step 2: a typed send tool for Pydantic AI

Pydantic AI turns your function signature into the tool schema the model sees. Use a BaseModel parameter so the model has to produce valid recipients and a bounded subject line.

import os
import httpx
from pydantic import BaseModel, EmailStr, Field
from pydantic_ai import Agent, RunContext

API_KEY = os.environ["ROBOTOMAIL_API_KEY"]
AUTH = {"Authorization": f"Bearer {API_KEY}"}


class MailboxDeps(BaseModel):
    mailbox_id: str
    mailbox_address: str


class SendEmail(BaseModel):
    """Arguments for sending one email from the agent's own mailbox."""
    to: list[EmailStr] = Field(max_length=5)
    subject: str = Field(max_length=200)
    body_text: str = Field(max_length=8000)
    in_reply_to: str | None = None


class SendResult(BaseModel):
    ok: bool
    detail: str


agent = Agent(
    "openai:gpt-4o",
    deps_type=MailboxDeps,
    system_prompt=(
        "You handle email from your own mailbox. Reply only to the sender of the "
        "message you were given. Never invent order numbers, prices, or dates. "
        "If you cannot answer with the context provided, set escalate=true."
    ),
)


@agent.tool
async def send_email(ctx: RunContext[MailboxDeps], params: SendEmail) -> SendResult:
    payload: dict = {
        "to": [str(a) for a in params.to],
        "subject": params.subject,
        "bodyText": params.body_text,
    }
    if params.in_reply_to:
        payload["inReplyTo"] = params.in_reply_to

    async with httpx.AsyncClient(timeout=20.0) as client:
        response = await client.post(
            f"https://api.robotomail.com/v1/mailboxes/{ctx.deps.mailbox_id}/messages",
            headers={**AUTH, "Content-Type": "application/json"},
            json=payload,
        )
    if response.status_code >= 400:
        return SendResult(ok=False, detail=f"send failed: {response.status_code}")
    return SendResult(ok=True, detail="accepted")

Two details worth copying. First, RunContext[MailboxDeps] keeps the mailbox ID out of the prompt, so the model cannot redirect sends to another mailbox. Second, returning a SendResult instead of raising on a 4xx lets the agent see the failure and decide what to do, while transport errors still surface as exceptions you can retry.

Add an allowlist check inside the tool if the agent should only ever mail known contacts:

    allowed = {"vendor@example.com", "billing@example.com"}
    if not set(str(a) for a in params.to) <= allowed:
        return SendResult(ok=False, detail="recipient not allowed")

Step 3: validate the inbound webhook payload

Robotomail delivers inbound mail as { event, timestamp, data } with snake_case fields under data. Model it once and you get typed access everywhere downstream.

from datetime import datetime
from pydantic import BaseModel, ConfigDict, EmailStr, Field


class InboundMessage(BaseModel):
    model_config = ConfigDict(populate_by_name=True)

    message_id: str
    mailbox_id: str
    mailbox_address: str
    sender: EmailStr = Field(alias="from")
    subject: str = ""
    body_text: str = ""
    thread_id: str
    received_at: datetime


class InboundEvent(BaseModel):
    event: str
    timestamp: datetime | None = None
    data: InboundMessage

from is a Python keyword, so alias it. populate_by_name=True means your own tests can build the object with sender= while the webhook still parses "from".

Step 4: turn the webhook into an agent run

from fastapi import BackgroundTasks, FastAPI, Request, Response

app = FastAPI()


@app.post("/hooks/robotomail")
async def inbound(request: Request, background: BackgroundTasks) -> Response:
    raw = await request.body()
    # verify the signature header before trusting the body, see the webhooks docs
    event = InboundEvent.model_validate_json(raw)

    if event.event != "message.received":
        return Response(status_code=204)

    if already_processed(event.data.message_id):  # your idempotency store
        return Response(status_code=200)

    background.add_task(handle_message, event.data)
    return Response(status_code=200)

Return 200 as soon as the payload validates. Deduplicate on message_id, because any webhook system will occasionally redeliver, and an agent that answers the same email twice looks broken to the human on the other end.

Step 5: constrain the reply with a validated output model

This is where Pydantic AI earns its place. Instead of letting the model free-form a send call, have it return a structured decision that your code inspects before anything is sent.

class DraftReply(BaseModel):
    should_reply: bool
    subject: str = Field(max_length=200)
    body_text: str = Field(max_length=8000)
    escalate: bool = False
    reason: str = Field(max_length=300)


triage = Agent(
    "openai:gpt-4o",
    deps_type=MailboxDeps,
    output_type=DraftReply,
    system_prompt=agent._system_prompts[0] if False else (
        "Draft a reply to the email below. Do not send it. "
        "Set escalate=true when the request involves refunds, legal matters, "
        "or anything you cannot verify from the message itself."
    ),
)


async def handle_message(msg: InboundMessage) -> None:
    deps = MailboxDeps(mailbox_id=msg.mailbox_id, mailbox_address=msg.mailbox_address)
    prompt = (
        f"From: {msg.sender}\nSubject: {msg.subject}\n\n"
        f"---BEGIN UNTRUSTED EMAIL BODY---\n{msg.body_text}\n---END---"
    )
    result = await triage.run(prompt, deps=deps)
    draft = result.output

    if draft.escalate or not draft.should_reply:
        await route_to_human(msg, draft)
        return

    await send_email(
        RunContext(deps=deps, model=triage.model, usage=result.usage(), prompt=prompt),
        SendEmail(
            to=[msg.sender],
            subject=draft.subject,
            body_text=draft.body_text,
            in_reply_to=msg.message_id,
        ),
    )

In real code, call a plain async def send(...) helper and register the same helper as a tool, rather than constructing a RunContext by hand. The point is the separation: the model proposes a DraftReply, your Python decides whether it ships.

in_reply_to is what keeps the conversation in one thread for the recipient's client. Pass the inbound message_id, not the thread_id. Our threading docs explain how thread IDs group the conversation on the Robotomail side while inReplyTo sets the RFC headers on the wire.

Guardrails that matter for email agents

• Treat the body as untrusted input. Email is the most attacker-friendly input channel an agent has. Delimit it in the prompt, never let it choose recipients, and keep the send allowlist in code. We wrote up the failure modes in detail in our post on email prompt injection.
• Cap outbound volume per mailbox per hour. A retry loop plus an autoreply on the other end is how agents generate mail storms. Enforce the cap in the tool, not the prompt.
• Log the tool call and the model output together. When someone asks why the agent sent that, you want the DraftReply and the resulting message ID side by side.
• Never reply to bounces or auto-submitted mail. Check for an Auto-Submitted header or a null-ish sender before you run the agent at all.

FAQ

Do I need SMTP or IMAP credentials for this?

No. Both directions are HTTP: a POST to send, and a webhook POST to receive. That avoids long-lived IMAP connections and OAuth consent screens, which is the main reason agent projects abandon consumer mailboxes.

How do I test the webhook path locally?

Send a real email to the mailbox and tunnel your local port, or replay a saved payload against your handler with InboundEvent.model_validate_json. Because the inbound shape is a Pydantic model, a fixture file plus one unit test covers most regressions. The receive and reply guide walks through the live version.

Can several Pydantic AI agents share one mailbox?

They can, but per-agent mailboxes are easier to reason about: separate addresses, separate webhooks, separate rate limits, and no ambiguity about which agent owns a thread. Use a shared mailbox only when the humans involved expect a single address.

What if the model returns a reply that fails validation?

Pydantic AI feeds the validation error back to the model and retries, up to the retry limit you configure. If it still fails, catch the exception and route the message to a human queue rather than sending something unvalidated.

Ready to wire this up? Create a mailbox, register a webhook, and point your Pydantic AI agent at it: get started with Robotomail.