---
title: "Send email from a Pydantic AI agent"
description: "Send email from a Pydantic AI agent: attach Unitpost as an MCPToolset or add one tool, and require approval before each send goes out."
url: https://www.unitpost.com/docs/ai/pydantic-ai
section: MCP
updated: 2026-11-02
---
# Send email from a Pydantic AI agent

## Connection settings

- **Server URL**: https://mcp.unitpost.com/mcp
- **Transport**: Streamable HTTP
- **Auth (recommended)**: OAuth 2.1 — add the URL, approve in the browser (scopes match API capabilities; tokens refresh automatically)
- **Auth (legacy)**: Authorization: Bearer pk_live_YOUR_KEY (workspace API key from Settings → API keys; for clients that can only paste a key)
- **Tools**: one per API operation (email_send, email_campaigns_send, …) (the default set)
- **Toolsets (optional)**: https://mcp.unitpost.com/mcp?toolsets=default,campaigns (or the X-MCP-Toolsets header; add ?readonly=1 for read-only tools)

## Pydantic AI

> Send email from a Pydantic AI agent: attach Unitpost as an MCPToolset or add one tool, and require approval before each send goes out.

Attach the Unitpost MCP server to a Pydantic AI agent as a toolset filtered to the send tool, or register one tool that calls the REST API. Deferred approvals end the run until a person decides.

### 1. Create a send-only key

Open https://www.unitpost.com → Settings → API keys and create a key with Sending only permissions. On a paid plan, open Advanced and keep only `emails:send`, so the key can send email and do nothing else. Copy the `pk_live_…` value (shown once) into the environment your agent runs in.

```bash
export UNITPOST_API_KEY=pk_live_YOUR_KEY
```

> A Sending only key can also send campaigns and read contacts, segments, templates and domains. Keep the key in your server code: the model never needs to see it.

### 2. Attach an MCPToolset

Install `pydantic-ai-slim[mcp]`. `MCPToolset` takes the key as `auth` and sends it as a Bearer token; `.filtered()` keeps one tool: `email_send`.

```python
import asyncio
import os

from pydantic_ai import Agent
from pydantic_ai.mcp import MCPToolset

unitpost = MCPToolset("https://mcp.unitpost.com/mcp?toolsets=sending", auth=os.environ["UNITPOST_API_KEY"]).filtered(
    lambda ctx, tool_def: tool_def.name == "email_send"
)
agent = Agent("openai:gpt-6.1-sol", toolsets=[unitpost])


async def main():
    result = await agent.run("Email ada@example.com from hi@yourdomain.com: the report is ready.")
    print(result.output)


asyncio.run(main())
```

> `email_send` takes an `idempotency_key` beside `body`. Tell the agent to pass a new one per message and the same one on a retry: the server's instructions say so too.

### 3. Or add one REST send tool

No MCP needed: one `@agent.tool_plain` function that posts to `/api/v1/email`. `requires_approval=True` defers every call to it until it's approved.

```python
import json
import os

import httpx
from pydantic_ai import Agent, DeferredToolRequests

ALLOWED = {"ada@example.com"}  # who this agent may email
agent = Agent("openai:gpt-6.1-sol", output_type=[str, DeferredToolRequests])


@agent.tool_plain(requires_approval=True)
async def send_email(sender: str, to: str, subject: str, text: str, idempotency_key: str) -> str:
    """Send one email through Unitpost. sender must be on a verified domain.

    idempotency_key: a new value for each message, and the same value when
    retrying that message, so a retry never sends twice.
    """
    if to not in ALLOWED:
        return f"Refused: {to} is not on this agent's allowlist."
    body = json.dumps({"from": sender, "to": to, "subject": subject, "text": text})
    async with httpx.AsyncClient() as http:
        res = await http.post(
            "https://www.unitpost.com/api/v1/email",
            content=body,
            headers={
                "Authorization": f"Bearer {os.environ['UNITPOST_API_KEY']}",
                "Content-Type": "application/json",
                "User-Agent": "my-agent/1.0",
                "Idempotency-Key": idempotency_key,  # the model's: a retry reuses it
            },
        )
    return res.text  # {"id": ...} or {"error": {"code": ..., "message": ...}}
```

> The model passes `idempotency_key`: a new one per message, and the same one when it retries that message, so a lost response never becomes a second email. A key made inside the tool would change on every retry. Never derive it from the content, or two separate identical notifications become one. `User-Agent` is required: a request without one is refused with `missing_user_agent` (403).

### 4. Gate each send

Every send is real mail to a real person. Send only transactional mail or mail people opted in to: the Acceptable Use Policy rules out cold outreach. Check each recipient against an allowlist, or have a person approve the send, before the call goes out. Pass an idempotency key, and the same key on a retry: for 24 hours a retry returns the first result instead of sending twice. In Pydantic AI, `requires_approval=True` on the tool, or `.approval_required()` on the MCP toolset, ends the run with a `DeferredToolRequests` output. Answer each pending call, then run again with the message history.

```python
from pydantic_ai import DeferredToolRequests, DeferredToolResults

# For MCP: Agent(..., toolsets=[unitpost.approval_required()], output_type=[str, DeferredToolRequests])


async def run_with_approval(prompt: str) -> str:
    result = await agent.run(prompt)
    while isinstance(result.output, DeferredToolRequests):
        approvals = {}
        for call in result.output.approvals:
            answer = input(f"Send {call.tool_name} with {call.args}? [y/N] ")
            approvals[call.tool_call_id] = answer.strip().lower() == "y"
        result = await agent.run(
            message_history=result.all_messages(),
            deferred_tool_results=DeferredToolResults(approvals=approvals),
        )
    return result.output
```

> Errors come back as `{ "error": { "code", "message", "details" } }`. Branch on `error.code` and fix the call instead of resending it: `domain_not_verified` means the `from` domain isn't verified, `validation_error` names the bad fields in `details`, and `rate_limit_exceeded` comes with a `Retry-After` wait. Over MCP the same error arrives under `data`, with a `hint` for the next move.

### 5. Verify the send

Make the first send to an address you own, from an address on a verified domain. A success returns the email's `id` and a `status` of `queued` (`scheduled` when you set `scheduled_at`); over MCP the same object comes back under `data`. Open Emails in the dashboard to see the message, its recipient and its delivery status, or read it back with `email_get` (MCP) or `GET /api/v1/email/{id}`.

```json
{
  "id": "email_…",
  "status": "queued",
  "scheduled_at": null
}
```

> A retry with the same idempotency key returns this same body (REST adds `Idempotent-Replayed: true`) and sends nothing new.

## Related

- [Send email from the Claude Agent SDK](https://www.unitpost.com/docs/ai/claude-agent-sdk): Send email from the Claude Agent SDK: add Unitpost to mcpServers over HTTP, or one in-process tool, and approve each send in canUseTool.
- [Let a Claude API agent send email (MCP connector)](https://www.unitpost.com/docs/ai/claude-api): Let a Claude API agent send email: attach Unitpost with the MCP connector (mcp_servers plus an mcp_toolset) or a client tool you approve.
- [Send email from Cursor](https://www.unitpost.com/docs/ai/cursor): Send email from Cursor's agent through the Unitpost MCP server. Add the server URL, approve with OAuth in the browser, then send a test.