How to Build a Multi-Agent Email Automation System with Claude API

What You'll Build

You're going to build a fully working multi-agent email automation system using Python and the Claude API — one that segments your contact list, writes personalized emails for each segment, and schedules delivery, all without you touching it after setup. This isn't a toy demo. It's a production-ready pattern you can drop into a real business workflow today.

By the end of this tutorial, three specialized Claude agents are working together: a segmentation agent that reads your contacts and groups them, a content generation agent that writes tailored emails using tool calls, and a scheduling agent that queues and timestamps everything for delivery. Think of it as your first real Claude API multi-agent system.

📦 Full Source Code Note
The complete, working code for this project is built step by step in the sections below. Each step adds a new piece, and by Step 5 you'll have everything assembled into one runnable script. Copy each block as you go, or jump to Step 5 to see the full orchestration loop that ties it all together.

Prerequisites

Python 3.10 or higher installed
An Anthropic API key (get one at console.anthropic.com)
anthropic Python SDK installed (pip install anthropic)
Basic familiarity with Python classes and async concepts
python-dotenv for environment variable management (pip install python-dotenv)

Step 1: Set Up Your Claude API Keys and Environment

First things first — let's get your environment wired up so Claude can actually talk to your code. Create a .env file in your project root and add your API key there. Never hardcode keys in source files.

.env

ANTHROPIC_API_KEY=sk-ant-your-key-here
[email protected]
MAX_EMAILS_PER_BATCH=50

Now install your dependencies and verify the SDK connects cleanly before writing any agent code.

setup_check.py

import os
from dotenv import load_dotenv
import anthropic

load_dotenv()

client = anthropic.Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))

# Quick ping to confirm auth works before building anything on top of it
response = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=64,
    messages=[{"role": "user", "content": "Reply with: API connection successful."}]
)

print(response.content[0].text)

Run this and you should see API connection successful. printed in your terminal. If you get an auth error, double-check that your .env file is in the same directory you're running from and that your key has no trailing spaces.

Step 2: Create the Email Segmentation Agent

The segmentation agent is the first brain in the pipeline. You hand it a raw list of contacts and it figures out how to group them — by industry, behavior, purchase history, whatever makes sense for your data. This is where Claude's reasoning really starts to shine compared to a dumb regex approach.

The agent takes a plain list of contact dictionaries and returns structured segments. I'm keeping the contact schema simple here so you can swap in your real CRM data later.

segmentation_agent.py

import os
import json
from dotenv import load_dotenv
import anthropic

load_dotenv()


class EmailSegmentationAgent:
    """
    Analyzes a contact list and groups contacts into meaningful segments
    for targeted email campaigns. Uses Claude to reason about segmentation
    rather than hard-coded rules.
    """

    def __init__(self):
        self.client = anthropic.Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))
        self.model = "claude-sonnet-4-6"

    def segment_contacts(self, contacts: list[dict]) -> dict:
        """
        Takes a list of contact dicts and returns a segmented structure.

        Args:
            contacts: List of dicts with keys: name, email, industry, last_purchase_days_ago, total_spend

        Returns:
            Dict with segment names as keys and lists of contacts as values
        """
        contacts_json = json.dumps(contacts, indent=2)

        prompt = f"""You are an expert email marketing strategist.
Analyze the following contact list and segment them into 3-4 distinct groups
based on their industry, recency, and spend patterns.

Return ONLY a valid JSON object where each key is a segment name (snake_case)
and each value is a list of email addresses belonging to that segment.
Include a brief one-sentence description for each segment using a key called
"_description" within each segment object.

Format your response as:
{{
  "segment_name": {{
    "_description": "Who this segment is and why they're grouped together",
    "contacts": ["[email protected]", "[email protected]"]
  }}
}}

Contact list:
{contacts_json}"""

        response = self.client.messages.create(
            model=self.model,
            max_tokens=1024,
            messages=[{"role": "user", "content": prompt}]
        )

        raw_text = response.content[0].text.strip()

        # Strip markdown code fences if Claude wraps the JSON
        if raw_text.startswith("```"):
            raw_text = raw_text.split("```")[1]
            if raw_text.startswith("json"):
                raw_text = raw_text[4:]

        segments = json.loads(raw_text.strip())
        return segments


# --- Sample usage ---
if __name__ == "__main__":
    sample_contacts = [
        {"name": "Maria Gonzalez", "email": "[email protected]", "industry": "real_estate",
         "last_purchase_days_ago": 12, "total_spend": 4800},
        {"name": "Dr. James Park", "email": "[email protected]", "industry": "healthcare",
         "last_purchase_days_ago": 45, "total_spend": 12000},
        {"name": "Tony Russo", "email": "[email protected]", "industry": "restaurant",
         "last_purchase_days_ago": 7, "total_spend": 950},
        {"name": "Karen Liu", "email": "[email protected]", "industry": "real_estate",
         "last_purchase_days_ago": 90, "total_spend": 2100},
        {"name": "Sam Becker", "email": "[email protected]", "industry": "automotive",
         "last_purchase_days_ago": 3, "total_spend": 8700},
        {"name": "Dr. Priya Nair", "email": "[email protected]", "industry": "healthcare",
         "last_purchase_days_ago": 180, "total_spend": 500},
        {"name": "Luis Fernandez", "email": "[email protected]", "industry": "restaurant",
         "last_purchase_days_ago": 22, "total_spend": 3200},
        {"name": "Beth Caldwell", "email": "[email protected]", "industry": "real_estate",
         "last_purchase_days_ago": 5, "total_spend": 15000},
    ]

    agent = EmailSegmentationAgent()
    segments = agent.segment_contacts(sample_contacts)
    print(json.dumps(segments, indent=2))

Here's what the output looks like when you run this against the sample data above:

Sample Output — segmentation_agent.py

{
  "high_value_active": {
    "_description": "High spenders who purchased recently — prime candidates for upsell campaigns.",
    "contacts": ["[email protected]", "[email protected]", "[email protected]"]
  },
  "healthcare_segment": {
    "_description": "Healthcare professionals grouped for industry-specific messaging.",
    "contacts": ["[email protected]", "[email protected]"]
  },
  "restaurant_owners": {
    "_description": "Food and beverage operators with mixed recency — nurture with ROI-focused content.",
    "contacts": ["[email protected]", "[email protected]"]
  },
  "win_back": {
    "_description": "Contacts with low spend or high recency gaps who need re-engagement.",
    "contacts": ["[email protected]", "[email protected]"]
  }
}

Step 3: Build the Content Generation Agent

This is the most interesting agent in the stack. The content generation agent uses Claude's tool use feature to pull in segment-specific context before writing each email. That means it can look up tone guidelines, compliance notes, or industry-specific talking points before drafting anything.

I'm defining two tools here: one that retrieves segment guidelines and one that formats the final email as structured JSON. This pattern keeps the output consistent so the scheduling agent downstream doesn't have to guess at the format.

content_agent.py

import os
import json
from dotenv import load_dotenv
import anthropic

load_dotenv()

# Simulated knowledge base your tool call will read from
SEGMENT_GUIDELINES = {
    "high_value_active": {
        "tone": "exclusive, grateful, forward-looking",
        "cta": "Schedule a strategy call",
        "avoid": "generic discounts — they respond to access and exclusivity"
    },
    "healthcare_segment": {
        "tone": "professional, compliance-aware, outcome-focused",
        "cta": "Download our HIPAA-ready AI overview",
        "avoid": "casual language, unsubstantiated health claims"
    },
    "restaurant_owners": {
        "tone": "energetic, practical, ROI-first",
        "cta": "See how restaurants in Naples are cutting labor costs",
        "avoid": "overly technical jargon"
    },
    "win_back": {
        "tone": "warm, low-pressure, curious",
        "cta": "Let's reconnect — 15 minutes, no pitch",
        "avoid": "guilt-tripping or urgency tactics"
    }
}


def get_segment_guidelines(segment_name: str) -> dict:
    """Simulates a tool that fetches writing guidelines for a given segment."""
    return SEGMENT_GUIDELINES.get(segment_name, {
        "tone": "professional and friendly",
        "cta": "Learn more about our services",
        "avoid": "nothing specific"
    })


def format_email_output(subject: str, body: str, cta_link: str) -> dict:
    """Structures the generated email into a consistent dict for the scheduler."""
    return {
        "subject": subject,
        "body": body,
        "cta_link": cta_link
    }


class ContentGenerationAgent:
    """
    Generates personalized email content for each segment using Claude's
    tool use to pull in segment-specific writing guidelines first.
    """

    def __init__(self):
        self.client = anthropic.Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))
        self.model = "claude-sonnet-4-6"

        # Tool definitions Claude will choose from during generation
        self.tools = [
            {
                "name": "get_segment_guidelines",
                "description": (
                    "Retrieves tone, CTA guidance, and content restrictions "
                    "for a specific audience segment before drafting an email."
                ),
                "input_schema": {
                    "type": "object",
                    "properties": {
                        "segment_name": {
                            "type": "string",
                            "description": "The snake_case name of the segment (e.g. high_value_active)"
                        }
                    },
                    "required": ["segment_name"]
                }
            },
            {
                "name": "format_email_output",
                "description": "Formats the final email into a structured object once writing is complete.",
                "input_schema": {
                    "type": "object",
                    "properties": {
                        "subject": {"type": "string", "description": "Email subject line"},
                        "body": {"type": "string", "description": "Full email body text"},
                        "cta_link": {"type": "string", "description": "The call-to-action URL"}
                    },
                    "required": ["subject", "body", "cta_link"]
                }
            }
        ]

    def _handle_tool_call(self, tool_name: str, tool_input: dict) -> str:
        """Routes tool calls from Claude to the correct local function."""
        if tool_name == "get_segment_guidelines":
            result = get_segment_guidelines(tool_input["segment_name"])
            return json.dumps(result)
        elif tool_name == "format_email_output":
            result = format_email_output(
                tool_input["subject"],
                tool_input["body"],
                tool_input["cta_link"]
            )
            return json.dumps(result)
        return json.dumps({"error": "Unknown tool"})

    def generate_email(self, segment_name: str, segment_description: str,
                       sender_name: str = "Chris at Naples AI") -> dict:
        """
        Runs the full tool-use loop to generate a personalized email for a segment.

        Returns:
            Dict with subject, body, and cta_link keys
        """
        messages = [
            {
                "role": "user",
                "content": (
                    f"You are writing a personalized marketing email for the '{segment_name}' segment.\n"
                    f"Segment description: {segment_description}\n\n"
                    f"Steps to follow:\n"
                    f"1. First, call get_segment_guidelines to retrieve the tone and CTA for this segment.\n"
                    f"2. Write a compelling email (150-200 words) from {sender_name} at Naples AI, "
                    f"a Southwest Florida AI agency that builds custom AI solutions for local businesses.\n"
                    f"3. Call format_email_output with your final subject, body, and a relevant CTA link "
                    f"(use https://calendly.com/chris-mejias-naplesaiagency/30min as the base URL).\n\n"
                    f"Begin by calling the guidelines tool."
                )
            }
        ]

        final_email = {}

        # Agentic loop — keeps running until Claude stops calling tools
        while True:
            response = self.client.messages.create(
                model=self.model,
                max_tokens=1024,
                tools=self.tools,
                messages=messages
            )

            # Append Claude's response to the conversation history
            messages.append({"role": "assistant", "content": response.content})

            if response.stop_reason == "end_turn":
                break

            if response.stop_reason == "tool_use":
                tool_results = []
                for block in response.content:
                    if block.type == "tool_use":
                        tool_result = self._handle_tool_call(block.name, block.input)

                        # Capture the formatted email when that tool fires
                        if block.name == "format_email_output":
                            final_email = json.loads(tool_result)

                        tool_results.append({
                            "type": "tool_result",
                            "tool_use_id": block.id,
                            "content": tool_result
                        })

                messages.append({"role": "user", "content": tool_results})
            else:
                break

        return final_email


# --- Sample usage ---
if __name__ == "__main__":
    agent = ContentGenerationAgent()
    email = agent.generate_email(
        segment_name="restaurant_owners",
        segment_description="Food and beverage operators with mixed recency — nurture with ROI-focused content."
    )
    print(json.dumps(email, indent=2))

Sample Output — content_agent.py

{
  "subject": "How Naples restaurants are trimming 10+ hours of weekly labor with AI",
  "body": "Hey Luis,\n\nRunning a restaurant in Southwest Florida is no joke — labor costs,\nno-shows, and last-minute schedule changes eat into margins every single week.\n\nAt Naples AI, we've helped local restaurant owners automate the busywork:\nfrom intelligent scheduling and inventory alerts to AI-powered menu pricing\nthat responds to demand in real time.\n\nThe owners we work with aren't tech people — they're operators who wanted\nless chaos and more time on the floor. Most see results within the first 30 days.\n\nIf you've got 15 minutes, I'd love to show you what this looks like\nfor a place your size. No pitch, just a real conversation.\n\nChris\nNaples AI",
  "cta_link": "https://calendly.com/chris-mejias-naplesaiagency/30min"
}

Step 4: Design the Scheduling and Delivery Agent

The scheduling agent is the last specialist in the chain. It takes the generated emails and figures out the optimal send time for each segment, then writes everything to a queue. In a real deployment you'd connect this to SendGrid, Mailgun, or AWS SES — but I'm keeping the delivery layer as a JSON queue here so you can swap in whatever sending infrastructure you already use.

scheduling_agent.py

import os
import json
from datetime import datetime, timedelta
from zoneinfo import ZoneInfo
from dotenv import load_dotenv
import anthropic

load_dotenv()

# Eastern Time — correct for Southwest Florida
NAPLES_TZ = ZoneInfo("America/New_York")


class SchedulingAgent:
    """
    Determines optimal send times for each email segment and writes
    scheduled jobs to a delivery queue (JSON file or extensible to a real queue).
    """

    def __init__(self, queue_path: str = "email_queue.json"):
        self.client = anthropic.Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))
        self.model = "claude-sonnet-4-6"
        self.queue_path = queue_path
        self.queue: list[dict] = []

    def _get_send_time_recommendation(self, segment_name: str, segment_description: str) -> str:
        """
        Asks Claude to recommend an optimal send day and time window
        based on segment behavior patterns.
        """
        prompt = (
            f"You are an email deliverability expert.\n"
            f"Given this audience segment — '{segment_name}': {segment_description}\n\n"
            f"Recommend the single best day of the week and time window (in Eastern Time) "
            f"to send a marketing email to maximize open rates.\n"
            f"Reply in this exact format and nothing else:\n"
            f"DAY: \nTIME:  ET\nREASON: "
        )

        response = self.client.messages.create(
            model=self.model,
            max_tokens=128,
            messages=[{"role": "user", "content": prompt}]
        )

        return response.content[0].text.strip()

    def _parse_send_time(self, recommendation: str) -> datetime:
        """
        Parses Claude's recommendation string into a concrete datetime object
        set to the next occurrence of that day/time from now.
        """
        lines = {line.split(":")[0].strip(): line.split(":", 1)[1].strip()
                 for line in recommendation.split("\n") if ":" in line}

        day_name = lines.get("DAY", "Tuesday")
        time_str = lines.get("TIME", "10:00 ET").replace(" ET", "").strip()

        days_of_week = ["Monday", "Tuesday", "Wednesday", "Thursday", "Friday", "Saturday", "Sunday"]
        target_weekday = days_of_week.index(day_name) if day_name in days_of_week else 1

        hour, minute = map(int, time_str.split(":"))
        now = datetime.now(tz=NAPLES_TZ)

        # Find the next occurrence of the target weekday
        days_ahead = (target_weekday - now.weekday()) % 7
        if days_ahead == 0 and now.hour >= hour:
            days_ahead = 7  # Push to next week if today's window already passed

        send_date = now + timedelta(days=days_ahead)
        send_datetime = send_date.replace(hour=hour, minute=minute, second=0, microsecond=0)
        return send_datetime

    def schedule_email(self, segment_name: str, segment_description: str,
                       email_content: dict, recipient_emails: list[str]) -> dict:
        """
        Schedules a generated email for a segment and appends it to the queue.

        Returns:
            The scheduled job dict that was added to the queue
        """
        recommendation = self._get_send_time_recommendation(segment_name, segment_description)
        send_at = self._parse_send_time(recommendation)

        job = {
            "job_id": f"{segment_name}_{int(send_at.timestamp())}",
            "segment": segment_name,
            "recipients": recipient_emails,
            "send_at": send_at.isoformat(),
            "subject": email_content.get("subject", ""),
            "body": email_content.get("body", ""),
            "cta_link": email_content.get("cta