Send email from a CrewAI agent

Add the Unitpost MCP server to a CrewAI agent's mcps and filter it to the send tool, or give the agent one tool that calls the REST API. A pre-tool-call hook checks recipients and asks a person before anything goes out.

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

   Add MCPServerHTTP to the agent

   Install crewai. MCPServerHTTP connects over streamable HTTP with your key in headers, and create_static_tool_filter keeps one tool: email_send.

   Python

   import os
   
   from crewai import Agent, Crew, Task
   from crewai.mcp import MCPServerHTTP
   from crewai.mcp.filters import create_static_tool_filter
   
   mailer = Agent(
       role="Mailer",
       goal="Send the emails the task asks for, and nothing else",
       backstory="Sends transactional email through Unitpost.",
       mcps=[
           MCPServerHTTP(
               url="https://mcp.unitpost.com/mcp?toolsets=sending",
               headers={"Authorization": f"Bearer {os.environ['UNITPOST_API_KEY']}"},
               streamable=True,
               tool_filter=create_static_tool_filter(allowed_tool_names=["email_send"]),
           ),
       ],
   )
   
   task = Task(
       description="Email ada@example.com from hi@yourdomain.com: the report is ready. Pass an idempotency_key.",
       expected_output="The id of the sent email",
       agent=mailer,
   )
   
   print(Crew(agents=[mailer], tasks=[task]).kickoff())

   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 @tool function that posts to /api/v1/email. Pass it in the agent's tools.

   Python

   import json
   import os
   
   import httpx
   from crewai.tools import tool
   
   ALLOWED = {"ada@example.com"}  # who this agent may email
   
   
   @tool("send_email")
   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})
       res = httpx.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 CrewAI, a PRE_TOOL_CALL hook runs before the tool: raise HookAborted to block the call, and the agent reads why. CrewAI prefixes MCP tool names with the server's name, so scope the hook to the name your crew lists for email_send.

   Python

   from crewai.hooks import HookAborted, InterceptionPoint, ToolCallHookContext, on
   
   ALLOWED = {"ada@example.com"}
   
   
   @on(InterceptionPoint.PRE_TOOL_CALL, tools=["send_email"])
   def approve_send(ctx: ToolCallHookContext) -> None:
       if ctx.tool_input.get("to") not in ALLOWED:
           raise HookAborted(reason="recipient is not on the allowlist", source="email-policy")
       answer = ctx.request_human_input(
           prompt=f"Send this email? {ctx.tool_input}",
           default_message="Type 'yes' to send:",
       )
       if answer.lower() != "yes":
           raise HookAborted(reason="declined by a person", source="email-policy")

   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.