CrewAI Integration Guide

Give your CrewAI agents a governed email mailbox. Send, receive, and track email with built-in policy enforcement.

CrewAI agents can send email by calling any HTTP endpoint through a custom tool. The problem is that nothing governs what they send: no deduplication, no cooldown windows, no suppression checks, no audit trail for the crew's decisions.

This guide shows how to give CrewAI agents a managed mailbox. Every email proposed by any agent in your crew runs through the Molted policy engine first - 20+ rules evaluated in under a second. Policy runs at the infrastructure layer, so no agent in the crew can override it regardless of what instructions it receives.

Prerequisites

A Molted account with an API key (sign up at molted.email/signup)
A verified sending domain (see Domains)
CrewAI installed: pip install crewai
Requests installed: pip install requests

1. Create a mailbox for your crew

Each crew should have its own mailbox (or one per agent role if you want separate policies). Log in to the portal, go to Mailboxes, and create one - or use the API:

curl

curl -X POST https://api.molted.email/v1/me/mailboxes \
  -H "Authorization: Bearer YOUR_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "Sales Crew",
    "emailAddress": "outreach@yourdomain.com"
  }'

Note the mailboxId - you will pass it in each send request.

2. Define the send tool

CrewAI tools are Python classes that inherit from BaseTool. The send tool wraps the Molted send endpoint and surfaces the policy decision back to the agent.

tools/send_email.py

import os
import requests
from crewai.tools import BaseTool
from pydantic import BaseModel, Field
from typing import Optional


class SendEmailInput(BaseModel):
    to: str = Field(..., description="Recipient email address")
    subject: str = Field(..., description="Email subject line")
    body: str = Field(..., description="Email body in plain text or HTML")
    dedupe_key: Optional[str] = Field(
        None,
        description="Unique key to prevent duplicate sends. Defaults to recipient+subject.",
    )


class SendEmailTool(BaseTool):
    name: str = "send_email"
    description: str = (
        "Send an email to a contact. The email is checked against policy rules "
        "before it is delivered. If blocked, the tool returns the reason — "
        "do not retry a blocked send."
    )
    args_schema: type[BaseModel] = SendEmailInput

    def _run(self, to: str, subject: str, body: str, dedupe_key: Optional[str] = None) -> str:
        api_key = os.environ["MOLTED_API_KEY"]
        tenant_id = os.environ["MOLTED_TENANT_ID"]

        payload = {
            "tenantId": tenant_id,
            "recipientEmail": to,
            "templateId": "_default",
            "dedupeKey": dedupe_key or f"{to}-{subject}",
            "agentId": "crewai-agent",
            "payload": {
                "subject": subject,
                "html": body,
                "text": body,
            },
        }

        response = requests.post(
            "https://api.molted.email/v1/agent/send/request",
            headers={
                "Authorization": f"Bearer {api_key}",
                "Content-Type": "application/json",
            },
            json=payload,
            timeout=10,
        )
        data = response.json()

        if data.get("status") == "blocked":
            return (
                f"Email blocked by policy. Reason: {data.get('blockReason')}. "
                f"Decision trace: {data.get('requestId')}. Do not retry."
            )

        return (
            f"Email queued successfully. "
            f"requestId={data.get('requestId')}, status={data.get('status')}"
        )

The tool returns the policy decision to the agent. When a send is blocked, the agent sees the reason and can inform its crew rather than silently failing or retrying.

3. Add a check-status tool (optional)

Agents can query delivery status of a previous send:

tools/check_email_status.py

import os
import requests
from crewai.tools import BaseTool
from pydantic import BaseModel, Field


class CheckStatusInput(BaseModel):
    request_id: str = Field(..., description="The requestId returned when the email was sent")


class CheckEmailStatusTool(BaseTool):
    name: str = "check_email_status"
    description: str = "Check the delivery status of a previously queued email."
    args_schema: type[BaseModel] = CheckStatusInput

    def _run(self, request_id: str) -> str:
        api_key = os.environ["MOLTED_API_KEY"]

        response = requests.get(
            f"https://api.molted.email/v1/agent/send/{request_id}/status",
            headers={"Authorization": f"Bearer {api_key}"},
            timeout=10,
        )
        data = response.json()

        return (
            f"Status: {data.get('status')}. "
            f"Provider: {data.get('provider', 'unknown')}. "
            f"Last event: {data.get('lastEvent', 'none')}."
        )

4. Wire the tools into a crew

crew.py

import os
from crewai import Agent, Task, Crew, Process
from tools.send_email import SendEmailTool
from tools.check_email_status import CheckEmailStatusTool

send_email = SendEmailTool()
check_status = CheckEmailStatusTool()

# An agent that writes and sends outreach emails
outreach_agent = Agent(
    role="Outreach Specialist",
    goal="Send personalized, policy-compliant emails to prospects",
    backstory=(
        "You craft personalized outreach emails and send them through the "
        "company mailbox. Every email you send is checked against policy rules "
        "before delivery. If a send is blocked, you report the reason to the "
        "crew manager and do not retry."
    ),
    tools=[send_email, check_status],
    verbose=True,
)

# A manager agent that orchestrates the outreach
manager_agent = Agent(
    role="Outreach Manager",
    goal="Coordinate the outreach campaign and track delivery",
    backstory=(
        "You manage the outreach campaign. You receive leads, assign sends to "
        "the outreach specialist, and track which emails were delivered, blocked, "
        "or pending."
    ),
    verbose=True,
)

send_task = Task(
    description=(
        "Send a trial welcome email to {recipient_email}. "
        "Subject: 'Welcome to the trial'. "
        "Body: 'Thanks for signing up - your 14-day trial starts today. "
        "Reply to this email if you have questions.' "
        "Use a dedupeKey of 'welcome-{recipient_email}' to prevent duplicates."
    ),
    expected_output=(
        "Confirmation that the email was queued with a requestId, "
        "or the block reason if it was rejected by policy."
    ),
    agent=outreach_agent,
)

crew = Crew(
    agents=[manager_agent, outreach_agent],
    tasks=[send_task],
    process=Process.sequential,
    verbose=True,
)

result = crew.kickoff(inputs={"recipient_email": "alice@example.com"})
print(result)

5. Per-agent registration for multi-agent crews

If your crew has multiple agents that send email independently, register each one. This gives you per-agent rate limits and a per-agent attribution trail in the decision trace:

register_agents.py

import os
import requests

api_key = os.environ["MOLTED_API_KEY"]
tenant_id = os.environ["MOLTED_TENANT_ID"]

agents = [
    {"name": "outreach-agent", "config": {"humanizer_enabled": True, "humanizer_style": "professional"}},
    {"name": "support-agent", "config": {"humanizer_enabled": False}},
    {"name": "churn-agent", "config": {"humanizer_enabled": True, "humanizer_style": "friendly"}},
]

for agent in agents:
    response = requests.post(
        "https://api.molted.email/v1/agent/register",
        headers={"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"},
        json={"tenantId": tenant_id, **agent},
    )
    data = response.json()
    print(f"Registered {agent['name']}: agentId={data['id']}")

Pass the registered agentId in each send request instead of the hardcoded string:

payload = {
    "tenantId": tenant_id,
    "agentId": registered_agent_id,  # from registration
    ...
}

6. Handle policy blocks in your crew

Common block reasons and how your agents should respond:

Reason	What it means	Recommended agent behavior
`duplicate_send`	Same `dedupeKey` within cooldown window	Inform crew, do not retry
`rate_limit_exceeded`	Mailbox or tenant hit send rate limit	Report to manager agent, pause outreach
`suppressed_recipient`	Contact has unsubscribed or bounced	Remove from target list, do not retry
`cooldown_active`	Per-recipient cooldown in effect	Schedule for later, report ETA if available
`risk_budget_exceeded`	Agent risk budget exhausted for this period	Escalate to manager, stop sends until reset

Add explicit block-handling instructions to your agent's backstory:

outreach_agent = Agent(
    role="Outreach Specialist",
    backstory=(
        "...When a send is blocked:\n"
        "- 'duplicate_send': the email was already sent recently. "
        "  Inform the crew manager. Do not retry.\n"
        "- 'suppressed_recipient': this contact has opted out. "
        "  Mark them as suppressed and skip.\n"
        "- 'rate_limit_exceeded': pause and notify the manager. "
        "  Do not retry immediately.\n"
        "- Any other block: report the reason and requestId to the manager."
    ),
    ...
)

7. Async crews (LangGraph-style flows)

For async or flow-based CrewAI setups, the send tool works the same way since it uses blocking HTTP. For non-blocking sends in an async context:

import asyncio
import httpx

async def send_email_async(to: str, subject: str, body: str, dedupe_key: str) -> dict:
    async with httpx.AsyncClient() as client:
        response = await client.post(
            "https://api.molted.email/v1/agent/send/request",
            headers={
                "Authorization": f"Bearer {os.environ['MOLTED_API_KEY']}",
                "Content-Type": "application/json",
            },
            json={
                "tenantId": os.environ["MOLTED_TENANT_ID"],
                "recipientEmail": to,
                "templateId": "_default",
                "dedupeKey": dedupe_key,
                "payload": {"subject": subject, "html": body, "text": body},
            },
            timeout=10.0,
        )
        return response.json()

What gets enforced

When any agent in your crew calls send_email, the mailbox evaluates 20+ policy rules before anything leaves:

Suppression - has this recipient opted out or hard-bounced?
Deduplication - has the same dedupeKey been used within the cooldown window?
Cooldown - is there an active cooldown for this recipient?
Rate limits - has the mailbox hit its per-minute, per-hour, or per-day budget?
Risk budget - has the agent exhausted its risk allocation?
Consent - does the contact have a valid consent record?

If all rules pass, the email is delivered. If any rule fires, the send is blocked and the decision trace records exactly which rule triggered and why. The agent receives the blockReason and can respond accordingly.

No agent in your crew - regardless of its instructions or the LLM powering it - can override these checks. They run at the infrastructure layer.

Quickstart - send your first policy-checked email
LangChain Integration Guide - LangChain-specific setup
Policy Simulation - test policy rules against draft sends
Reactive Agent Guide - build agents that respond to inbound email
Agent Coordination - multi-agent leases and consensus
Autonomy Levels - require human approval for high-risk sends

On this page