How do I handle inbound emails with my agent?

AgentMail offers three ways to process incoming emails, each suited to different use cases.

1. Webhooks (Recommended for Production)

Configure a webhook URL and AgentMail will send a POST request to your endpoint whenever an email arrives. This is the most reliable approach for production applications.

from flask import Flask, request
from agentmail import AgentMail
app = Flask(__name__)
client = AgentMail()
processed_event_ids = set()  # use a database or cache in production
@app.route("/webhooks", methods=["POST"])
def handle_webhook():
    payload = request.json
    # deliveries are at least once, so skip events you already handled
    if payload["event_id"] in processed_event_ids:
        return "OK", 200
    processed_event_ids.add(payload["event_id"])
    if payload["event_type"] == "message.received":
        message = payload["message"]
        
        # Your agent processes the email here
        reply_text = your_agent.process(message)
        
        # Reply in the same thread
        client.inboxes.messages.reply(
            inbox_id=message["inbox_id"],
            message_id=message["message_id"],
            text=reply_text
        )
    
    return "OK", 200

Register your webhook via the API:

client.webhooks.create(
    url="https://your-domain.ngrok-free.app/webhooks",
    event_types=["message.received"],
)

event_types is required: list every event type the webhook should receive.

Deduplicate on event_id

Webhook delivery is at least once: a retry can arrive after your endpoint already processed the event, so the same event can reach you twice. Every payload carries an event_id that stays the same across retries and across every webhook that receives the event. Record the event_id values you have processed and skip repeats, as in the example above. Do not deduplicate on the svix-id header: it differs between webhooks registered for the same events at different levels (organization, pod, inbox).

If your endpoint goes down

AgentMail retries a failed delivery 8 times over about 27.5 hours. If deliveries keep failing for 5 days, the webhook is disabled automatically and AgentMail does not notify you: the webhook’s enabled field turns false. After you fix the endpoint, re-enable it with client.webhooks.update(webhook_id, enabled=True). Events from while it was down are not redelivered, so catch up by listing messages received since the last event you processed:

response = client.inboxes.messages.list(
    inbox_id="agent@agentmail.to",
    after=last_processed_at,  # datetime of the last event you handled
    ascending=True,
)
for item in response.messages:  # page with next_page_token if there are more
    message = client.inboxes.messages.get(inbox_id=item.inbox_id, message_id=item.message_id)
    your_agent.process(message)

See Recovering missed events for the other options.

For local development, use ngrok to expose your local server. See the Webhook Setup Guide for full instructions.

2. WebSockets (Best for Real-Time, No Public URL)

Stream email events over a persistent connection. No public URL or ngrok needed, which makes this ideal for local development and desktop agents.

import asyncio
from agentmail import AsyncAgentMail, Subscribe, Subscribed, MessageReceivedEvent
client = AsyncAgentMail()
async def main():
    async with client.websockets.connect() as socket:
        await socket.send_subscribe(Subscribe(
            inbox_ids=["agent@agentmail.to"]
        ))
        async for event in socket:
            if isinstance(event, Subscribed):
                print(f"Subscribed to: {event.inbox_ids}")
            elif isinstance(event, MessageReceivedEvent):
                print(f"New email from: {event.message.from_}")
                print(f"Subject: {event.message.subject}")
asyncio.run(main())

The SDK also provides a synchronous client if you prefer:

from agentmail import AgentMail, Subscribe, MessageReceivedEvent
client = AgentMail()
with client.websockets.connect() as socket:
    socket.send_subscribe(Subscribe(
        inbox_ids=["agent@agentmail.to"]
    ))
    for event in socket:
        if isinstance(event, MessageReceivedEvent):
            print(f"New email from: {event.message.from_}")

See the WebSocket Overview for more details.

3. Polling webhooks (Agents That Start and Stop)

A polling webhook queues events for your agent instead of pushing them. Your agent fetches the queued events whenever it runs and commits the ones it has processed, so nothing is lost while it is offline and nothing needs to be reachable from the internet. This fits cron jobs, serverless functions, and agents behind a firewall.

subscription = client.webhooks.polling.create(
    client_id="support-agent",
    event_types=["message.received"],
    inbox_ids=["agent@agentmail.to"],
)
token = subscription.token  # read the queue with the Svix SDK's AutoConfigConsumer

Polling webhooks are available in the US region only for now. See Polling Webhooks for how to read and commit events.

Which should I use?

For most production use cases, webhooks are recommended. They are reliable, event-driven, and integrate well with serverless platforms. If you need real-time events without exposing a public URL, WebSockets are the best option. If your agent cannot stay connected, use a polling webhook so events wait for it.