e2a 1.0.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 email threads. Pass it when replying, and e2a will include it in the webhook when the human responds.

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

    if email.conversation_id:
        # Follow-up — route to existing conversation
        conversation = get_conversation(email.conversation_id)
    else:
        # First contact — create a new conversation
        conversation = create_conversation(sender=email.sender)

    response = conversation.generate_reply(email)

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

Works the same for outbound emails:

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"

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 local-mode agent
result = client.register_agent("my-bot")
print(result.email)  # [e-mail]

# Custom domain agent
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)