Auto-Reply Email Agent

Fresh 🌱

Build a simple agent that automatically responds to incoming emails with personalized messages

Overview

Learn how to build an email auto-reply agent that automatically responds to incoming emails. This beginner-friendly example demonstrates the core concepts of building with AgentMail: receiving webhooks, processing email events, and sending automated replies.

What You'll Build

By the end of this guide, you'll have a working auto-reply agent that:

1. Receives incoming emails to a dedicated AgentMail inbox
2. Processes webhook events in real-time
3. Extracts sender information (name and email)
4. Generates personalized replies using a template
5. Sends automated responses back to the sender

Here's what the user experience looks like:

User sends email → Agentmail inbox receives → Agent processes it → User gets reply
                         ↓
              "Hi John, thank you for your email!
               I've received your message and will
               get back to you within 24 hours..."

Prerequisites

Before you begin, make sure you have:

Required:

• Python 3.8 or higher installed
• An AgentMail account and API key
• An ngrok account (free tier works)

Project Setup

Step 1: Create Project Directory

Create a new directory for your agent:

mkdir auto-reply-agent
cd auto-reply-agent

Step 2: Create the Agent Code

Create a file named agent.py and paste the following code:

"""
Auto-Reply Email Agent

A simple example showing how to build an email auto-reply bot with AgentMail.
This agent automatically responds to incoming emails with personalized messages.
"""

import os
from dotenv import load_dotenv

# Load environment variables before importing AgentMail
load_dotenv()

from flask import Flask, request, Response
import ngrok
from agentmail import AgentMail
import threading

# Configuration
PORT = 8080
INBOX_USERNAME = os.getenv("INBOX_USERNAME", "auto-reply")
WEBHOOK_DOMAIN = os.getenv("WEBHOOK_DOMAIN")

# Initialize Flask app and AgentMail client
app = Flask(__name__)
client = AgentMail()
processed_messages = set()  # Track processed message IDs to prevent duplicates

def setup_agentmail():
    """Create inbox and webhook with idempotency."""
    print("Setting up AgentMail infrastructure...")

    # Create inbox (or get existing one)
    try:
        inbox = client.inboxes.create(
            username=INBOX_USERNAME,
            client_id=f"{INBOX_USERNAME}-inbox"
        )
        print(f"✓ Inbox created: {inbox.inbox_id}")
    except Exception as e:
        if "already exists" in str(e).lower():
            inbox_id = f"{INBOX_USERNAME}@agentmail.to"
            class SimpleInbox:
                def __init__(self, inbox_id):
                    self.inbox_id = inbox_id
            inbox = SimpleInbox(inbox_id)
            print(f"✓ Using existing inbox: {inbox.inbox_id}")
        else:
            raise

    # Start ngrok tunnel
    listener = ngrok.forward(PORT, domain=WEBHOOK_DOMAIN, authtoken_from_env=True)

    # Create webhook (or get existing one)
    try:
        webhook = client.webhooks.create(
            url=f"{listener.url()}/webhook/agentmail",
            event_types=["message.received"],
            inbox_ids=[inbox.inbox_id],
            client_id=f"{INBOX_USERNAME}-webhook"
        )
        print(f"✓ Webhook created")
    except Exception as e:
        if "already exists" in str(e).lower():
            print(f"Webhook already exists")
        else:
            raise

    print(f"\n✓ Setup complete!")
    print(f"Inbox: {inbox.inbox_id}")
    print(f"Webhook: {listener.url()}/webhook/agentmail\n")

    return inbox, listener

def generate_reply(sender_name, subject):
    """Generate auto-reply message using a template."""
    return (
        f"Hi {sender_name},\n\n"
        f"Thank you for your email! I've received your message and will get back to you within 24 hours.\n\n"
        f"If your matter is urgent, please reply with \"URGENT\" in the subject line.\n\n"
        f"Best regards,\n"
        f"Auto-Reply Agent"
    )

def process_and_reply(message_id, inbox_id, from_field, subject, message):
    """Process incoming message and send reply in background."""
    # Extract sender email and name
    if '<' in from_field and '>' in from_field:
        sender_email = from_field.split('<')[1].split('>')[0].strip()
        sender_name = from_field.split('<')[0].strip()
        if not sender_name or ',' in sender_name:
            sender_name = sender_email.split('@')[0].title()
    else:
        sender_email = from_field.strip()
        sender_name = sender_email.split('@')[0].title() if '@' in sender_email else 'Friend'

    # Log incoming email
    print(f"Processing email from {sender_email}: {subject}")

    # Generate and send auto-reply
    try:
        reply_text = generate_reply(sender_name, subject)
        client.inboxes.messages.reply(
            inbox_id=inbox_id,
            message_id=message_id,
            to=[sender_email],
            text=reply_text
        )
        print(f"Auto-reply sent to {sender_email}\n")
    except Exception as e:
        print(f"Error: {e}\n")

@app.route('/webhook/agentmail', methods=['POST'])
def receive_webhook():
    """Webhook endpoint to receive incoming email notifications."""
    payload = request.json
    event_type = payload.get('type') or payload.get('event_type')

    # Ignore outgoing messages
    if event_type == 'message.sent':
        return Response(status=200)

    message = payload.get('message', {})
    message_id = message.get('message_id')
    inbox_id = message.get('inbox_id')
    from_field = message.get('from_', '') or message.get('from', '')

    # Validate required fields
    if not message_id or not inbox_id or not from_field:
        return Response(status=200)

    # prevent duplicate
    if message_id in processed_messages:
        return Response(status=200)
    processed_messages.add(message_id)

    subject = message.get('subject', '(no subject)')

    # Process in background thread and return immediately
    thread = threading.Thread(
        target=process_and_reply,
        args=(message_id, inbox_id, from_field, subject, message)
    )
    thread.daemon = True
    thread.start()

    return Response(status=200)

if __name__ == '__main__':
    print("\n" + "="*60)
    print("AUTO-REPLY EMAIL AGENT")
    print("="*60 + "\n")

    inbox, listener = setup_agentmail()

    print(f"Agent is ready!")
    print(f"Send emails to: {inbox.inbox_id}")
    print(f"\nWaiting for incoming emails...\n")

    app.run(port=PORT)

Step 3: Create Requirements File

Create a file named requirements.txt:

agentmail
flask>=3.0.0
ngrok>=1.0.0
python-dotenv>=1.0.0

Step 4: Install Dependencies

Install the required Python packages:

pip install -r requirements.txt

Step 5: Configure Environment Variables

Create a .env file with your credentials:

# AgentMail Configuration
AGENTMAIL_API_KEY=your_agentmail_api_key_here

# Ngrok Configuration
NGROK_AUTHTOKEN=your_ngrok_authtoken_here
WEBHOOK_DOMAIN=your-name.ngrok-free.app

# Inbox Settings
INBOX_USERNAME=auto-reply

Code Walkthrough

Let's understand how the agent works by breaking down the key components.

Architecture Overview

┌─────────────┐
│  Someone    │
│  sends      │ ──────► ┌──────────────┐
│  email      │         │  AgentMail   │
└─────────────┘         │   Inbox      │
                        └──────┬───────┘
                               │ Webhook
                               ▼
                        ┌──────────────┐
                        │   Ngrok      │
                        │   Tunnel     │
                        └──────┬───────┘
                               │
                               ▼
                        ┌──────────────┐
                        │  Your Flask  │
                        │    Server    │
                        └──────┬───────┘
                               │
                               ▼
                        ┌──────────────┐
                        │   Generate   │
                        │   & Send     │
                        │   Reply      │
                        └──────────────┘

1. Initialization

import os
from dotenv import load_dotenv

# Load environment variables FIRST
load_dotenv()

from flask import Flask, request, Response
import ngrok
from agentmail import AgentMail

# Initialize the AgentMail client
client = AgentMail()  # Reads AGENTMAIL_API_KEY from environment
app = Flask(__name__)

Key points:

• Load .env variables before importing AgentMail
• AgentMail SDK automatically reads AGENTMAIL_API_KEY from environment
• Flask creates the web server for receiving webhooks

2. Setting Up Infrastructure

The setup_agentmail() function creates your inbox and webhook:

def setup_agentmail():
    """Create inbox and webhook with idempotency."""

    # Create inbox (or get existing one)
    try:
        inbox = client.inboxes.create(
            username=INBOX_USERNAME,
            client_id=f"{INBOX_USERNAME}-inbox"  # ← Idempotency key
        )
        print(f"✓ Inbox created: {inbox.inbox_id}")
    except Exception as e:
        if "already exists" in str(e).lower():
            # Inbox already exists, that's fine!
            inbox_id = f"{INBOX_USERNAME}@agentmail.to"
            inbox = SimpleInbox(inbox_id)

Why client_id?

The client_id parameter makes this operation idempotent - you can run it multiple times without creating duplicates. If the inbox already exists, AgentMail returns the existing one.

    # Start ngrok tunnel
    listener = ngrok.forward(PORT, domain=WEBHOOK_DOMAIN, authtoken_from_env=True)

    # Create webhook
    webhook = client.webhooks.create(
        url=f"{listener.url()}/webhook/agentmail",
        event_types=["message.received"],  # ← Only subscribe to incoming emails
        client_id=f"{INBOX_USERNAME}-webhook"  # ← Idempotency
    )

What's happening:

1. Ngrok creates a public URL that forwards to localhost:8080
2. We register a webhook with AgentMail
3. AgentMail will POST to this URL when emails arrive

3. Processing Webhooks

The webhook endpoint receives incoming email notifications:

@app.route('/webhook/agentmail', methods=['POST'])
def receive_webhook():
    """Webhook endpoint to receive incoming email notifications."""
    payload = request.json
    event_type = payload.get('type') or payload.get('event_type')

    # Ignore outgoing messages (prevents infinite loops!)
    if event_type == 'message.sent':
        return Response(status=200)

Why ignore message.sent?

When your agent sends a reply, AgentMail triggers a message.sent webhook. If we don't filter this out, the agent would treat its own replies as new emails and respond to itself infinitely!

By returning 200, we tell AgentMail "I received this webhook successfully, but I'm choosing not to process it."

4. Extracting Email Data

    message = payload.get('message', {})
    message_id = message.get('message_id')
    inbox_id = message.get('inbox_id')
    from_field = message.get('from_', '') or message.get('from', '')

    # Validate required fields
    if not message_id or not inbox_id or not from_field:
        return Response(status=200)  # Gracefully skip incomplete data

Webhook payload structure:

{
  "type": "message.received",
  "message": {
    "message_id": "<abc123@agentmail.to>",
    "inbox_id": "auto-reply@agentmail.to",
    "from_": "John Doe <john@example.com>",
    "subject": "Hello",
    "text": "Email body content..."
  }
}

5. Parsing Sender Information

Email addresses can come in different formats. We handle both:

# Extract sender email and name
if '<' in from_field and '>' in from_field:
    # Format: "John Doe <john@example.com>"
    sender_email = from_field.split('<')[1].split('>')[0].strip()
    sender_name = from_field.split('<')[0].strip()
    if not sender_name or ',' in sender_name:
        # Name is empty or has comma, use email username
        sender_name = sender_email.split('@')[0].title()
else:
    # Format: "john@example.com"
    sender_email = from_field.strip()
    sender_name = sender_email.split('@')[0].title()

Examples:

• "John Doe <john@example.com>" → name: "John Doe", email: "john@example.com"
• "john@example.com" → name: "John", email: "john@example.com"
• "Last, First <name@example.com>" → name: "Name", email: "name@example.com"

6. Generating the Reply

def generate_reply(sender_name, subject):
    """Generate auto-reply message using a template."""
    return (
        f"Hi {sender_name},\n\n"
        f"Thank you for your email! I've received your message and will get back to you within 24 hours.\n\n"
        f"If your matter is urgent, please reply with \"URGENT\" in the subject line.\n\n"
        f"Best regards,\n"
        f"Auto-Reply Agent"
    )

This simple template-based approach requires no AI or external APIs. The reply is personalized with the sender's name.

7. Sending the Reply

try:
    reply_text = generate_reply(sender_name, subject)
    client.inboxes.messages.reply(
        inbox_id=inbox_id,
        message_id=message_id,
        to=[sender_email],  # ← Must be a list!
        text=reply_text
    )
    print(f"Auto-reply sent to {sender_email}\n")
except Exception as e:
    print(f"Error: {e}\n")

return Response(status=200)  # Always return 200 to acknowledge webhook

Important details:

• to parameter must be a list of email addresses
• message_id links the reply to the original email (threading)
• Always return 200 status to acknowledge the webhook
• Errors are logged but don't crash the server

Why always return 200?

Even if sending the reply fails, we return 200 to AgentMail. This tells AgentMail "I received and processed this webhook." If we returned an error status, AgentMail would retry sending the webhook multiple times, which isn't helpful for application errors.

Running the Agent

Start the agent:

python agent.py

You should see output like this:

============================================================
AUTO-REPLY EMAIL AGENT
============================================================

Setting up AgentMail infrastructure...
✓ Inbox created: auto-reply@agentmail.to
✓ Webhook created

✓ Setup complete!
  Inbox: auto-reply@agentmail.to
  Webhook: https://your-name.ngrok-free.app/webhook/agentmail

Agent is ready!
Send emails to: auto-reply@agentmail.to
Reply mode: Template-based

Waiting for incoming emails...

 * Running on http://127.0.0.1:8080

Success! Your agent is now running and ready to receive emails.

Leave this terminal window open - closing it will stop the agent.

Testing Your Agent

Let's verify everything works by sending a test email.

Send a Test Email

1. Open your personal email (Gmail, Outlook, etc.)

2. Compose a new email:

   To: auto-reply@agentmail.to
   Subject: Testing my auto-reply agent
   Body: Hi there! This is a test message.

3. Send the email

Watch the Magic Happen

In your terminal, you should see:

Email from youremail@gmail.com: Testing my auto-reply agent
Auto-reply sent to youremail@gmail.com

Check Your Inbox

Within seconds, you should receive an automated reply:

Hi Youremail,

Thank you for your email! I've received your message and will get back to you within 24 hours.

If your matter is urgent, please reply with "URGENT" in the subject line.

Best regards,
Auto-Reply Agent

It works! You just built and tested your first AgentMail agent.

The agent extracted your name, personalized the message, and sent an instant reply.

Customization

You can customize the auto-reply message by editing the generate_reply() function in agent.py.

The function has access to:

• sender_name - The sender's name extracted from their email
• subject - The subject line of the email

Simply modify the text in the return statement to change what your agent replies with.

Troubleshooting

Common Issues

Problem: Python dependencies not installed.

Solution:

pip install -r requirements.txt

If using a virtual environment, make sure it's activated first:

source venv/bin/activate  # macOS/Linux
venv\Scripts\activate     # Windows

Problem: Invalid or missing API key.

Solutions:

1. Check your .env file has the correct AGENTMAIL_API_KEY
2. Verify the API key is valid in your AgentMail Dashboard
3. Make sure there are no extra spaces or quotes around the key
4. Ensure .env is in the same directory as agent.py

Test your API key:

from agentmail import AgentMail
client = AgentMail()
print(client.inboxes.list())  # Should succeed

Problem: Invalid or missing ngrok auth token.

Solutions:

1. Get your auth token from ngrok dashboard
2. Update NGROK_AUTHTOKEN in .env
3. Verify the token has no extra spaces

Alternatively, configure ngrok globally:

ngrok config add-authtoken YOUR_TOKEN

Checklist:

• Is the agent running? (python agent.py should show "Waiting for incoming emails...")
• Is ngrok tunnel active? (Check console output for webhook URL)
• Did you send email to the correct inbox? (Check console for inbox address)
• Is the webhook URL accessible? Test with: curl https://your-domain.ngrok-free.app/webhook/agentmail

Debug steps:

1. Add logging to see webhook payloads:

@app.route('/webhook/agentmail', methods=['POST'])
def receive_webhook():
    payload = request.json
    print(f"📨 Received webhook: {json.dumps(payload, indent=2)}")
    # ... rest of code ...

2. Check ngrok dashboard for webhook requests:

   • Visit ngrok dashboard
   • View request logs to see if webhooks are arriving

3. Verify webhook is registered:

client = AgentMail()
webhooks = client.webhooks.list()
print(webhooks)

Problem: Another process is using port 8080.

Solution 1: Kill the process using the port

# macOS/Linux
lsof -ti:8080 | xargs kill -9

# Windows
netstat -ano | findstr :8080
taskkill /PID <PID> /F

Solution 2: Use a different port

PORT = 8081  # Change in agent.py

Problem: Webhook received but no reply sent.

Debug steps:

1. Check console for error messages
2. Verify the reply API parameters:

print(f"Sending to: {sender_email}")
print(f"From inbox: {inbox_id}")
print(f"Reply text: {reply_text[:50]}...")

3. Test the reply API directly:

client.inboxes.messages.reply(
    inbox_id="your-inbox@agentmail.to",
    message_id="test-message-id",
    to=["test@example.com"],
    text="Test reply"
)

4. Ensure to is a list (common mistake):

#  Wrong
to=sender_email

#  Correct
to=[sender_email]

Congratulations! You've built your first AgentMail agent.

Advanced Feature: AI-Powered Replies

Want to upgrade your agent with intelligent, context-aware responses? You can add AI-powered replies using OpenAI.

Step 1: Install OpenAI

pip install openai

Step 2: Add your OpenAI API key to .env

# AI Configuration
USE_AI_REPLY=true
OPENAI_API_KEY=sk-your_actual_openai_api_key_here

Step 3: Add thread history and AI reply functions to agent.py

After the generate_reply() function, add:

def get_thread_history(thread_id):
    """Fetch conversation history for the thread."""
    try:
        thread = client.threads.get(thread_id=thread_id)
        return thread.messages if hasattr(thread, 'messages') else []
    except Exception as e:
        print(f"Failed to fetch thread history: {e}")
        return []

def format_thread_for_ai(messages):
    """Format thread messages into conversation history for AI."""
    conversation = []

    for msg in messages:
        if hasattr(msg, 'from_'):
            sender = msg.from_
            text = msg.text or msg.html or ""
        else:
            sender = msg.get('from_', '') or msg.get('from', '')
            text = msg.get('text', '') or msg.get('html', '') or msg.get('body', '')

        if '<' in sender and '>' in sender:
            sender = sender.split('<')[1].split('>')[0].strip()

        if text:
            conversation.append(f"From: {sender}\n{text}")

    return "\n\n---\n\n".join(reversed(conversation))

def generate_ai_reply(sender_name, email_body, subject, thread_history=""):
    """Generate AI-powered reply using OpenAI with thread context."""
    try:
        context = f"Email thread history:\n\n{thread_history}\n\n---\n\nLatest message from {sender_name}:\nSubject: {subject}\n{email_body}" if thread_history else f"Subject: {subject}\nFrom: {sender_name}\n{email_body}"

        response = openai_client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[
                {
                    "role": "system",
                    "content": "You are an intelligent email assistant. Read the email thread and respond in a helpful, contextual way. If this is a follow-up in a conversation, acknowledge what was previously discussed. If you can provide helpful information based on the context, do so. If the question requires detailed research or expertise you don't have, acknowledge receipt and set expectations. Be conversational, professional, and concise."
                },
                {
                    "role": "user",
                    "content": f"{context}\n\nGenerate a helpful reply that considers the conversation history. Keep it concise (2-4 sentences) but be actually helpful if you can address their question or continue the conversation meaningfully."
                }
            ],
            max_tokens=250,
            temperature=0.7
        )
        return response.choices[0].message.content
    except Exception as e:
        print(f"AI generation failed, using template: {e}")
        return generate_reply(sender_name, subject)

Step 4: Update the webhook handler

In the receive_webhook() function, add thread_id extraction and replace the reply generation section:

    subject = message.get('subject', '(no subject)')
    thread_id = message.get('thread_id', '')  # Add this line

    # Log incoming email
    print(f"Email from {sender_email}: {subject}")

    # Generate and send auto-reply
    try:
        if USE_AI_REPLY:
            email_body = message.get('text', '') or message.get('body', '')

            # Fetch thread history for context
            thread_history = ""
            if thread_id:
                print(f"Fetching thread history for: {thread_id[:20]}...")
                messages = get_thread_history(thread_id)
                if messages:
                    thread_history = format_thread_for_ai(messages)
                    print(f"Found {len(messages)} messages in thread")

            reply_text = generate_ai_reply(sender_name, email_body, subject, thread_history)
            print("Using AI-generated reply with thread context")
        else:
            reply_text = generate_reply(sender_name, subject)
            print("Using template reply")

        client.inboxes.messages.reply(
            inbox_id=inbox_id,
            message_id=message_id,
            to=[sender_email],
            text=reply_text
        )
        print(f"Auto-reply sent to {sender_email}\n")
    except Exception as e:
        print(f"Error: {e}\n")

How it works:

• Agent fetches entire email thread using client.threads.get(thread_id)
• Thread history is formatted and passed to OpenAI for context-aware replies
• AI can reference previous messages and provide more intelligent responses
• If OpenAI API fails, keep trying for three times

Result:

Your agent now has conversation memory. When replying to follow-up emails, the AI sees the entire conversation history and can provide contextual, intelligent responses that reference previous exchanges instead of generic auto-replies.

---

If you build something cool with AgentMail, we'd love to hear about it. Share in our Discord community!