Send email from a LangChain agent

Load the Unitpost send tool into a LangChain agent (Python) with the built-in MCP adapter, or write one tool that calls the REST API. Human-in-the-loop middleware pauses the agent before each send.

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.

Load the tools with MCPAdapter

Install langchain[mcp] (1.4 or later; langchain.mcp is in beta). A FastMCP Client with auth set to the key sends it as a Bearer token. Keep only email_send from the list.

Python

import asyncio
import os

from fastmcp.client import Client
from langchain.agents import create_agent
from langchain.mcp import MCPAdapter


async def main():
    client = Client("https://mcp.unitpost.com/mcp?toolsets=sending", auth=os.environ["UNITPOST_API_KEY"])
    async with MCPAdapter(client) as adapter:
        tools = [t for t in await adapter.list_tools() if t.name == "email_send"]
        agent = create_agent("claude-sonnet-5", tools)
        result = await agent.ainvoke(
            {"messages": [{"role": "user", "content": "Email ada@example.com from hi@yourdomain.com: the report is ready."}]}
        )
        print(result["messages"][-1].content)


asyncio.run(main())

Still on langchain-mcp-adapters? LangChain's migration guide replaces MultiServerMCPClient with MCPAdapter. 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.

Or add one REST send tool

No MCP needed: one @tool function that posts to /api/v1/email. The docstring becomes the tool description the model reads.

Python

import json
import os

import httpx
from langchain.tools import tool

ALLOWED = {"ada@example.com"}  # who this agent may email


@tool
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
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 LangChain, HumanInTheLoopMiddleware pauses the agent before the send tool runs (send_email here, email_send over MCP). It needs a checkpointer and a thread ID; resume with a decision.
Python
from langchain.agents import create_agent from langchain.agents.middleware import HumanInTheLoopMiddleware from langgraph.checkpoint.memory import InMemorySaver from langgraph.types import Command agent = create_agent( "claude-sonnet-5", [send_email], middleware=[ HumanInTheLoopMiddleware( interrupt_on={"send_email": {"allowed_decisions": ["approve", "reject"]}}, ) ], checkpointer=InMemorySaver(), ) config = {"configurable": {"thread_id": "mail-1"}} paused = agent.invoke({"messages": [{"role": "user", "content": "Email ada@example.com: the report is ready."}]}, config) print(paused["__interrupt__"]) # the pending send: show it to a person agent.invoke(Command(resume={"decisions": [{"type": "approve"}]}), config)
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.

Send email from a LangChain agent

Create a send-only key

Load the tools with MCPAdapter

Or add one REST send tool

Gate each send

Verify the send

Search docs and guides