Build Email Agents with Claude's Native Tool Use

from anthropic import Anthropic
import requests

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

email_tools = [
    {
        "name": "send_email",
        "description": "Send an email through MultiMail. In gated_send mode, this creates a pending draft for human approval.",
        "input_schema": {
            "type": "object",
            "properties": {
                "mailbox_id": {"type": "string", "description": "The mailbox to send from"},
                "to": {"type": "string", "description": "Recipient email address"},
                "subject": {"type": "string", "description": "Email subject line"},
                "body": {"type": "string", "description": "Email body content"}
            },
            "required": ["mailbox_id", "to", "subject", "body"]
        }
    },
    {
        "name": "check_inbox",
        "description": "Check the inbox for a given mailbox. Returns recent messages.",
        "input_schema": {
            "type": "object",
            "properties": {
                "mailbox_id": {"type": "string", "description": "The mailbox to check"},
                "limit": {"type": "integer", "description": "Max messages to return", "default": 10}
            },
            "required": ["mailbox_id"]
        }
    }
]

client = Anthropic()

def execute_tool(name, args):
    if name == "send_email":
        resp = requests.post(f"{MULTIMAIL_API}/send", headers=MM_HEADERS, json=args)
    elif name == "check_inbox":
        resp = requests.get(
            f"{MULTIMAIL_API}/mailboxes/{args[&"cm">#039;mailbox_id']}/inbox",
            headers=MM_HEADERS, params={"limit": args.get("limit", 10)}
        )
    else:
        return {"error": f"Unknown tool: {name}"}
    return resp.json()

def run_email_agent(user_message, mailbox_id):
    messages = [{"role": "user", "content": user_message}]
    while True:
        response = client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=1024,
            system=f"You are an email assistant. Mailbox ID: {mailbox_id}. "
                   f"Emails are sent in gated_send mode — they queue for human approval.",
            tools=email_tools,
            messages=messages
        )
        if response.stop_reason == "tool_use":
            tool_block = next(b for b in response.content if b.type == "tool_use")
            result = execute_tool(tool_block.name, tool_block.input)
            messages.append({"role": "assistant", "content": response.content})
            messages.append({"role": "user", "content": [
                {"type": "tool_result", "tool_use_id": tool_block.id,
                 "content": str(result)}
            ]})
        else:
            return response.content[0].text

reply_tools = [
    {
        "name": "get_thread",
        "description": "Get all messages in an email thread for context.",
        "input_schema": {
            "type": "object",
            "properties": {
                "thread_id": {"type": "string", "description": "Thread ID to retrieve"}
            },
            "required": ["thread_id"]
        }
    },
    {
        "name": "reply_email",
        "description": "Reply to an email by message ID within its thread.",
        "input_schema": {
            "type": "object",
            "properties": {
                "message_id": {"type": "string", "description": "Message ID to reply to"},
                "body": {"type": "string", "description": "Reply body content"}
            },
            "required": ["message_id", "body"]
        }
    }
]

"cm"># Combine all tools for a full-featured email agent
all_tools = email_tools + reply_tools

result = run_email_agent(
    "Check my inbox and reply to the latest message from Sarah",
    mailbox_id="mbx_abc123"
)

def run_streaming_agent(user_message, mailbox_id):
    messages = [{"role": "user", "content": user_message}]
    with client.messages.stream(
        model="claude-sonnet-4-20250514",
        max_tokens=1024,
        system=f"You are an email assistant for mailbox {mailbox_id}.",
        tools=email_tools,
        messages=messages
    ) as stream:
        for event in stream:
            if hasattr(event, &"cm">#039;type'):
                if event.type == &"cm">#039;content_block_start':
                    if hasattr(event.content_block, &"cm">#039;name'):
                        print(f"Calling tool: {event.content_block.name}")
                elif event.type == &"cm">#039;text':
                    print(event.text, end=&"cm">#039;', flush=True)

        response = stream.get_final_message()
    return response