Email-Capable Crews with MultiMail

import requests
from crewai.tools import tool

MULTIMAIL_API = "https://api.multimail.dev/v1"
HEADERS = {"Authorization": "Bearer mm_live_your_api_key"}

@tool("Send Email")
def send_email(to: str, subject: str, body: str, mailbox_id: str) -> str:
    """Send an email via MultiMail. In gated_send mode, the email is queued for human approval before delivery."""
    resp = requests.post(f"{MULTIMAIL_API}/send", headers=HEADERS, json={
        "mailbox_id": mailbox_id,
        "to": to,
        "subject": subject,
        "body": body
    })
    return str(resp.json())

@tool("Check Inbox")
def check_inbox(mailbox_id: str) -> str:
    """Check the inbox for new emails in a given mailbox."""
    resp = requests.get(
        f"{MULTIMAIL_API}/mailboxes/{mailbox_id}/inbox",
        headers=HEADERS,
        params={"limit": 10}
    )
    return str(resp.json())

@tool("Reply to Email")
def reply_email(message_id: str, body: str) -> str:
    """Reply to an email within its thread."""
    resp = requests.post(f"{MULTIMAIL_API}/reply", headers=HEADERS, json={
        "message_id": message_id,
        "body": body
    })
    return str(resp.json())

from crewai import Agent, Task, Crew

researcher = Agent(
    role="Lead Researcher",
    goal="Find the right contacts and gather context for outreach",
    backstory="You are an expert at finding decision-makers and understanding their needs.",
    tools=[check_inbox, search_contacts]
)

outreach_agent = Agent(
    role="Outreach Specialist",
    goal="Draft and send personalized emails that get responses",
    backstory="You craft compelling, concise emails. The mailbox uses gated_send mode, so a human reviews each email before delivery.",
    tools=[send_email, reply_email, check_inbox]
)

research_task = Task(
    description="Check the inbox for any replies to our previous outreach. Summarize who responded and what they said.",
    expected_output="A summary of inbox activity with contact names and response content.",
    agent=researcher
)

outreach_task = Task(
    description="Based on the research summary, draft follow-up emails to contacts who replied positively. Send each email via the send_email tool.",
    expected_output="Confirmation that follow-up emails have been queued for approval.",
    agent=outreach_agent
)

crew = Crew(
    agents=[researcher, outreach_agent],
    tasks=[research_task, outreach_task],
    verbose=True
)

result = crew.kickoff()

@tool("Get Thread")
def get_thread(thread_id: str) -> str:
    """Retrieve the full email thread for context."""
    resp = requests.get(
        f"{MULTIMAIL_API}/threads/{thread_id}",
        headers=HEADERS
    )
    return str(resp.json())

support_agent = Agent(
    role="Customer Support Agent",
    goal="Respond helpfully to customer emails in the support inbox",
    backstory="You handle customer support. Always be helpful and professional. All replies go through gated_send for manager review.",
    tools=[check_inbox, get_thread, reply_email]
)

support_task = Task(
    description="Check the support inbox (mailbox ID: mbx_support). For each unread email, read the full thread and draft a helpful reply.",
    expected_output="All customer emails have been replied to with helpful responses queued for approval.",
    agent=support_agent
)

support_crew = Crew(
    agents=[support_agent],
    tasks=[support_task],
    verbose=True
)

support_crew.kickoff()

@tool("Search Contacts")
def search_contacts(query: str) -> str:
    """Search the contact database by name or email."""
    resp = requests.get(
        f"{MULTIMAIL_API}/contacts",
        headers=HEADERS,
        params={"q": query}
    )
    return str(resp.json())

@tool("Add Contact")
def add_contact(email: str, name: str, tags: str = "") -> str:
    """Add a new contact to the database."""
    resp = requests.post(f"{MULTIMAIL_API}/contacts", headers=HEADERS, json={
        "email": email,
        "name": name,
        "tags": tags.split(",") if tags else []
    })
    return str(resp.json())