---
title: "Give an OpenAI Agents SDK agent an email tool"
description: "Give an OpenAI Agents SDK agent an email tool: MCPServerStreamableHttp to Unitpost or one function tool, with approvals and idempotency keys."
url: https://www.unitpost.com/docs/ai/openai-agents-sdk
section: MCP
updated: 2026-11-02
---
# Give an OpenAI Agents SDK agent an email tool

## 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)

## OpenAI Agents SDK

> Give an OpenAI Agents SDK agent an email tool: MCPServerStreamableHttp to Unitpost or one function tool, with approvals and idempotency keys.

Connect an Agents SDK agent (Python) to the Unitpost MCP server and filter it down to the send tool, or give it one function tool that calls the REST API. Approvals pause the run until a person says yes.

### 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. Connect MCPServerStreamableHttp

Install `openai-agents`. `MCPServerStreamableHttp` sends your key as a Bearer header, and `create_static_tool_filter` leaves the agent one tool: `email_send`.

```python
import asyncio
import os

from agents import Agent, Runner
from agents.mcp import MCPServerStreamableHttp, create_static_tool_filter


async def main() -> None:
    async with MCPServerStreamableHttp(
        name="unitpost",
        params={
            "url": "https://mcp.unitpost.com/mcp?toolsets=sending",
            "headers": {"Authorization": f"Bearer {os.environ['UNITPOST_API_KEY']}"},
        },
        tool_filter=create_static_tool_filter(allowed_tool_names=["email_send"]),
    ) as server:
        agent = Agent(
            name="Mailer",
            instructions="Send email only when asked. Pass a new idempotency_key per message.",
            mcp_servers=[server],
        )
        result = await Runner.run(agent, "Email ada@example.com from hi@yourdomain.com: the report is ready.")
        print(result.final_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 function tool

No MCP server needed: one `@tool` function that posts to `/api/v1/email`. The SDK builds the schema from the signature and docstring.

```python
import json
import os

import httpx
from agents import Agent
from agents.decorators import tool

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


@tool
async def send_email(sender: str, to: str, subject: str, text: str, idempotency_key: str) -> str:
    """Send one email through Unitpost.

    Args:
        sender: From address on a verified domain, e.g. "Acme <hi@acme.com>".
        to: One recipient address.
        subject: Subject line.
        text: Plain-text body.
        idempotency_key: A new value for each message; the same value when
            retrying this one, 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": ...}}


agent = Agent(name="Mailer", instructions="Send email only when asked.", tools=[send_email])
```

> 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 the Agents SDK, `@tool(needs_approval=True)` on the function, or `require_approval={"email_send": "always"}` on the MCP server, pauses the run at each send. Approve or reject the pending calls, then resume.

```python
from agents import Runner


async def run_with_approval(agent, prompt: str) -> str:
    result = await Runner.run(agent, prompt)
    while result.interruptions:
        state = result.to_state()
        for item in result.interruptions:
            answer = input(f"Send {item.name} with {item.arguments}? [y/N] ")
            if answer.strip().lower() == "y":
                state.approve(item, always_approve=False)
            else:
                state.reject(item)
        result = await Runner.run(agent, state)
    return result.final_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 a LangChain agent](https://www.unitpost.com/docs/ai/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.
- [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.