Send email from a Pydantic AI agent

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. 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.

   Shell

   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. 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. 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. 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. 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.