e2a 1.3.0

Python SDK for the e2a protocol — email-to-agent authentication

e2a Python SDK

Python SDK for the e2a protocol — email-to-agent authentication.

Install

pip install e2a

For WebSocket real-time delivery:

pip install e2a[ws]

Import paths

The stable, pinned API surface lives under e2a.v1:

from e2a.v1 import E2AClient, AsyncE2AClient, E2AApi

Top-level e2a imports remain available as convenience aliases to the current stable version, but use e2a.v1 in examples, production code, and version-pinned integrations.

Quick start

from e2a.v1 import E2AClient

# Reads E2A_API_KEY from environment automatically
client = E2AClient()

# Or pass explicitly:
# client = E2AClient(api_key="e2a_your_api_key")

Mount the webhook in your web framework:

FastAPI:

from fastapi import FastAPI, Request

app = FastAPI()

@app.post("/webhook")
async def webhook(request: Request):
    email = client.parse(await request.body())
    print(f"From: {email.sender}, Subject: {email.subject}")
    email.reply("Thanks for reaching out!")
    return {"ok": True}

Flask:

from flask import Flask, request

app = Flask(__name__)

@app.post("/webhook")
def webhook():
    email = client.parse(request.get_data())
    email.reply("Thanks for reaching out!")
    return {"ok": True}

Raw vs high-level API

The SDK has two layers:

• E2AApi / AsyncE2AApi — raw typed HTTP client. Returns generated Pydantic models. Uses /api/v1/ paths.
• E2AClient / AsyncE2AClient — high-level wrapper. Returns parsed InboundEmail objects with .reply().

Access the raw layer through client.api:

from e2a.v1 import E2AClient

client = E2AClient(api_key="e2a_...")

# High-level: returns InboundEmail with parsed MIME, .reply(), etc.
email = client.get_message("msg_123")

# Raw: returns generated MessageDetail Pydantic model
detail = client.api.get_message("[e-mail]", "msg_123")

Conversation threading

e2a supports an opaque conversation_id that lets your agent track multi-turn threads across the email boundary. Pass it on any send() or reply(), and e2a will surface it on the recipient's inbound payload when they respond — whether the other side is a human replying from Gmail or another e2a agent.

The basic loop

@app.post("/webhook")
async def webhook(request: Request):
    email = client.parse(await request.body())

    if email.conversation_id:
        # Follow-up — route to the existing conversation
        conversation = get_conversation(email.conversation_id)
    else:
        # First contact — create a new conversation and pick an id for it
        conversation = create_conversation(sender=email.sender)

    response = conversation.generate_reply(email)

    # Tag the reply so future messages in this thread are linked
    email.reply(
        body=response.text,
        html_body=response.html,
        conversation_id=conversation.id,
    )
    return {"ok": True}

Same idea for a new outbound:

result = client.send(
    to="[e-mail]",
    subject="Following up",
    body="Hi Alice, just checking in.",
    conversation_id="conv_abc123",
)
# When Alice replies, the webhook will include conversation_id="conv_abc123"

When is email.conversation_id populated?

Inbound type	Sender passed conversation_id?	What you see
First email from a human (new thread)	n/a — humans don't pass it	None — you must assign one if you want to thread subsequent messages
Human reply to an earlier email from your agent	n/a	The id you passed on your outbound (recovered via In-Reply-To)
Another e2a agent sending you a new message	yes, recommended	The sender's asserted id (carried on a custom header)
Another e2a agent sending you a new message	no	None
Another e2a agent replying to you	either way	Your earlier outbound's id, unless the sender asserted a different one

Rules of thumb:

• Always pass conversation_id when you're tagging an outbound as part of a known thread. It's the only way the recipient's webhook will see it.
• On first contact from a human, assign a new id yourself and stash it before you reply. After that, email.conversation_id will keep threading the conversation.
• Don't look up the id from email.sender alone — the same person can have many parallel threads.

Agent-to-agent conversations

If the recipient is another e2a-managed agent, conversation_id passed on send() arrives on the recipient's inbound on the very first message — no prior exchange needed. e2a carries it across on a custom header (X-E2A-Conversation-Id) for same-platform traffic. External senders (Gmail, Outlook, …) can't forge this header: it's only honored when the message originates from our own relay.

# Agent A initiates a thread with Agent B
await client_a.send(
    to=["[e-mail]"],
    subject="Can you handle this?",
    body="Details in the body.",
    conversation_id="task-2026-04-19-7f3a",
)

# Agent B's webhook immediately sees conversation_id="task-2026-04-19-7f3a"
# on the very first message — no round-trip required.

What conversation_id is not

• Not globally unique; not a primary key in e2a's DB. e2a treats it as an opaque string tagged on each message.
• Not a security boundary. Don't rely on it for authentication — check email.auth_headers for verified sender identity.
• Not guaranteed on every message. Design your code to handle None (typically: first contact from a human, or an external sender you've never interacted with before).

Attachments

Receiving attachments

Inbound email attachments are automatically parsed and available on email.attachments:

email = client.parse(body)
for att in email.attachments:
    print(f"{att.filename} ({att.content_type}, {att.size} bytes)")
    save_file(att.filename, att.data)

Sending attachments

Pass Attachment objects when sending or replying:

from e2a.v1 import Attachment

# Read a file
with open("report.pdf", "rb") as f:
    pdf_data = f.read()

# Send with attachment
client.send(
    to="[e-mail]",
    subject="Your report",
    body="See attached.",
    attachments=[
        Attachment(
            filename="report.pdf",
            content_type="application/pdf",
            data=pdf_data,
            size=len(pdf_data),
        )
    ],
)

# Or reply with attachment
email.reply(
    "Here's the file you requested.",
    attachments=[
        Attachment(filename="data.csv", content_type="text/csv", data=csv_bytes, size=len(csv_bytes))
    ],
)

Async support

For async frameworks like FastAPI, use AsyncE2AClient. Same interface, all I/O methods are async:

from e2a.v1 import AsyncE2AClient

client = AsyncE2AClient()  # reads E2A_API_KEY from env

@app.post("/webhook")
async def webhook(request: Request):
    email = client.parse(await request.body())
    await email.reply("Thanks!", conversation_id="conv_123")
    return {"ok": True}

WebSocket (real-time delivery for local agents)

Local-mode agents can receive emails in real time via WebSocket using the async listen() method. No public URL needed.

pip install e2a[ws]

import asyncio
from e2a.v1 import AsyncE2AClient

async def main():
    async with AsyncE2AClient(api_key="e2a_...") as client:
        async for email in client.listen("[e-mail]"):
            print(f"From: {email.sender}, Subject: {email.subject}")
            await email.reply("Got it!")

asyncio.run(main())

listen() connects to e2a's WebSocket endpoint, receives lightweight notifications, fetches the full message via REST, and yields AsyncInboundEmail objects. It reconnects automatically with exponential backoff (1s, 2s, 4s, ... up to 30s).

The WebSocket protocol is notification-only (server-to-client). The client never sends application frames.

Parameters:

Parameter	Type	Default	Description
agent_email	str	client.agent_email	Agent email to listen for
reconnect	bool	True	Auto-reconnect on disconnect
max_backoff	float	30.0	Maximum reconnect delay (seconds)

Agent and domain management

from e2a.v1 import E2AClient

client = E2AClient(api_key="e2a_...")

# Register a shared-domain agent using a slug (just the local part, not a full email).
# The server appends @agents.e2a.dev automatically.
result = client.register_agent("my-bot")        # slug only, e.g. "my-bot"
print(result.email)  # [e-mail]

# Custom domain agent — use the `email` parameter with a full email address.
# The domain must be registered and DNS-verified first.
result = client.register_agent(email="[e-mail]", agent_mode="cloud", webhook_url="https://mycompany.com/webhook")

# List agents
agents = client.list_agents()

# Domain management
client.register_domain("mycompany.com")
client.verify_domain("mycompany.com")
client.list_domains()
client.delete_domain("mycompany.com")

Sending emails

Send outbound emails directly:

result = client.send(
    to="[e-mail]",
    subject="Hello from my agent",
    body="Hi Alice!",
    conversation_id="conv_abc123",  # optional
)
print(result.status, result.message_id)

InboundEmail

Field	Type	Description
message_id	str	Unique e2a message ID
conversation_id	str | None	Your thread ID from a prior reply, or None for first contact
sender	str	Sender email address
recipient	str	Recipient email address (your agent)
subject	str	Email subject line
text_body	str	Plain-text email body
html_body	str | None	HTML email body, if present
attachments	list[Attachment]	File attachments (empty list if none)
received_at	str | None	Timestamp when the message was received
is_verified	bool	Whether the sender's identity is verified
auth	AuthHeaders	Full authentication details
raw_message	bytes	Raw RFC 2822 email bytes

Methods:

• email.reply(body, html_body=None, conversation_id=None, attachments=None) → SendResult

API Reference

E2AClient(api_key=None, agent_email=None, base_url="https://e2a.dev")

High-level sync client. api_key falls back to E2A_API_KEY env var.

• client.parse(body) → InboundEmail — accepts bytes, str, dict, or MessageDetail
• client.get_message(message_id) → InboundEmail
• client.get_messages(status="unread", page_size=50) → MessageList
• client.reply(message_id, body, ...) → SendResult
• client.send(to, subject, body, ...) → SendResult
• client.api → E2AApi (raw typed access)

AsyncE2AClient(api_key=None, agent_email=None, base_url="https://e2a.dev")

Same as E2AClient — all I/O methods are async. parse() is sync (no I/O needed).

• client.listen(agent_email=None, reconnect=True, max_backoff=30.0) → AsyncIterator[AsyncInboundEmail] (requires e2a[ws])
• client.api → AsyncE2AApi (raw typed async access)

Models

• InboundEmail / AsyncInboundEmail — parsed email with .reply()
• Attachment — filename, content_type, data (bytes), size
• SendResult — status, message_id, method
• AuthHeaders — verified, sender, entity_type, domain_check, delegation, signature, timestamp

Exceptions

• E2AApiError — API error (has status_code and message)