e2a 1.6.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 notif in client.listen("[e-mail]"):
            # The notification is lightweight metadata only — no body, no REST call.
            print(f"From: {notif.from_}, Subject: {notif.subject}")

            # Fetch the full email when you actually want it.
            email = await client.get_message(notif.message_id)
            await email.reply("Got it!")

asyncio.run(main())

listen() yields WSNotification objects — the lightweight metadata the server pushes (message_id, from_, recipient, subject, received_at, conversation_id). It does not auto-fetch the body: that's the caller's call. This matches the server design (small WS frames, explicit REST fetch) and lets callers skip messages without a network round-trip.

Reconnects automatically with exponential backoff (1s, 2s, 4s, ... up to 30s by default). Protocol is server-to-client only — the client never sends application frames.

WSNotification fields:

Field	Type	Notes
message_id	str	Pass to client.get_message(...) to fetch the body
from_	str	Sender. Trailing underscore: from is a Python keyword
recipient	str	Per-delivery target (your agent's address)
subject	str
received_at	str	RFC 3339 timestamp
conversation_id	str | None	Threading; None for first contact

listen() 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	Per-delivery target — your agent's address
to	list[str]	Parsed To: header — every address from the original message
cc	list[str]	Parsed Cc: header (empty when no CCs)
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[WSNotification] (requires e2a[ws]). Yields lightweight notifications; call await client.get_message(notif.message_id) to fetch the body.
• 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)

License

Apache-2.0 — see LICENSE and NOTICE in the upstream repo.