Give a CrewAI agent its own inbox

uv add crewai mcp

import os

from crewai import Agent, Crew, Task
from crewai.mcp import MCPServerHTTP
from crewai.mcp.filters import create_static_tool_filter

shipmail = MCPServerHTTP(
    url="https://shipmail.to/api/mcp",
    headers={"Authorization": f"Bearer {os.environ['SHIPMAIL_API_KEY']}"},
    streamable=True,
    # Read the inbox and write reply drafts. Nothing else.
    tool_filter=create_static_tool_filter(
        allowed_tool_names=[
            "shipmail_list_mailboxes",
            "shipmail_list_mailbox_inbox_messages",
            "shipmail_get_mailbox_inbox_thread",
            "shipmail_create_inbox_reply_draft",
        ]
    ),
)

inbox_agent = Agent(
    role="Inbox assistant for agent@yourdomain.com",
    goal="Answer customer email accurately and briefly",
    backstory="You read the whole conversation before you reply, and you never promise what you cannot check.",
    mcps=[shipmail],
)

triage = Task(
    description="List the unread mail in agent@yourdomain.com and draft a reply to each message that needs one.",
    expected_output="One line per message: the sender, the subject, and whether you drafted a reply.",
    agent=inbox_agent,
)

print(Crew(agents=[inbox_agent], tasks=[triage]).kickoff())

uv add crewai shipmail

import json
import os

from crewai.tools import tool
from shipmail import Shipmail

shipmail = Shipmail(os.environ["SHIPMAIL_API_KEY"])
MAILBOX_ID = os.environ["SHIPMAIL_MAILBOX_ID"]
FROM = os.environ["SHIPMAIL_FROM"]


def _sender(message: dict) -> str | None:
    senders = message["from"] or []
    return senders[0]["email"] if senders else None


@tool("list_inbox")
def list_inbox(unread_only: bool = True, limit: int = 10) -> str:
    """List recent messages in the agent's inbox with their id, conversation_id, sender, subject and preview."""
    params = {"folder_role": "inbox", "limit": limit}
    if unread_only:
        params["not_keyword"] = "$seen"
    page = shipmail.mailboxes.list_inbox_messages(MAILBOX_ID, params)
    return json.dumps(
        [
            {
                "id": m["id"],
                "conversation_id": m["conversation_id"],
                "from": _sender(m),
                "subject": m["subject"],
                "received_at": m["received_at"],
                "preview": m["preview"],
            }
            for m in page["data"]
        ]
    )


@tool("read_conversation")
def read_conversation(conversation_id: str) -> str:
    """Read every message in one conversation."""
    thread = shipmail.mailboxes.get_inbox_thread(MAILBOX_ID, conversation_id)
    return json.dumps(
        [
            {
                "from": _sender(m),
                "received_at": m["received_at"],
                "subject": m["subject"],
                # What the sender newly wrote, without the quoted history.
                "text": m["reply_text"] or m["preview"],
            }
            for m in thread["data"]
        ]
    )


@tool("send_email")
def send_email(to: list[str], subject: str, text: str) -> str:
    """Send a new email from the agent's address."""
    message = shipmail.messages.send({"from": FROM, "to": to, "subject": subject, "text": text})
    return json.dumps({"id": message["id"], "status": message["status"]})


@tool("reply")
def reply(conversation_id: str, to: list[str], text: str) -> str:
    """Reply in an existing conversation."""
    message = shipmail.mailboxes.reply_to_inbox_thread(
        MAILBOX_ID, conversation_id, {"to": to, "text": text}
    )
    return json.dumps({"id": message["id"], "status": message["status"]})

from crewai import Agent, Crew, Task
from crewai.hooks import HookAborted, InterceptionPoint, ToolCallHookContext, on

from email_tools import list_inbox, read_conversation, reply, send_email


# Runs before every send or reply. Anything but "yes" blocks the call.
@on(InterceptionPoint.PRE_TOOL_CALL, tools=["send_email", "reply"])
def approve_send(ctx: ToolCallHookContext) -> None:
    answer = ctx.request_human_input(
        prompt=f"Allow {ctx.tool_name}?",
        default_message=f"Input: {ctx.tool_input}\nType 'yes' to approve:",
    )
    if answer.strip().lower() != "yes":
        raise HookAborted(reason="denied by operator", source="approval-gate")


inbox_agent = Agent(
    role="Inbox assistant for agent@yourdomain.com",
    goal="Answer customer email accurately and briefly",
    backstory="You read the whole conversation before you reply, and you never promise what you cannot check.",
    tools=[list_inbox, read_conversation, send_email, reply],
)

answer_jane = Task(
    description="Reply to the latest email from jane@customer.com and confirm the call on Thursday.",
    expected_output="The reply you sent, or why you did not send one.",
    agent=inbox_agent,
)

print(Crew(agents=[inbox_agent], tasks=[answer_jane]).kickoff())