# Pydantic AI Email Integration: Typed Send and Receive Published: August 28, 2026 Give a Pydantic AI agent its own inbox: typed send tools, inbound webhooks that trigger agent runs, and validated reply models with real code. 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: ```bash npx @robotomail/cli mailbox create support-triage ``` Or over the API: ```bash 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. ```python 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: ```python 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. ```python 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 ```python 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. ```python 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](https://robotomail.com).