---
title: "Send email from a LangChain agent"
description: "Send email from a LangChain agent: load Unitpost's MCP tools with MCPAdapter or add one @tool, then gate sends with human-in-the-loop middleware."
url: https://www.unitpost.com/docs/ai/langchain
section: MCP
updated: 2026-11-02
---
# Send email from a LangChain 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)

## LangChain

> Send email from a LangChain agent: load Unitpost's MCP tools with MCPAdapter or add one @tool, then gate sends with human-in-the-loop middleware.

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.

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

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

## Related

- [Send email from a Mastra agent](https://www.unitpost.com/docs/ai/mastra): Send email from a Mastra agent: connect MCPClient to Unitpost or add one createTool send tool, with tool approval and idempotency keys.
- [Send email from a CrewAI agent](https://www.unitpost.com/docs/ai/crewai): Send email from a CrewAI agent: add Unitpost with MCPServerHTTP or one @tool, filter to the send tool, and gate sends with a tool hook.
- [Send email from a Pydantic AI agent](https://www.unitpost.com/docs/ai/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.