e2a 0.8.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

Quick start

from e2a 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}

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 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 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 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).

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)

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)
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, base_url="https://e2a.dev")

api_key is optional — falls back to the E2A_API_KEY environment variable.

• client.parse(body) → InboundEmail
• client.reply(message_id, body, html_body=None, conversation_id=None, attachments=None) → SendResult
• client.send(to, subject, body, conversation_id=None, attachments=None) → SendResult

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

Same as E2AClient — reply() and send() are async. parse() is sync (no I/O needed). api_key falls back to env var. Use as async context manager:

async with AsyncE2AClient() as client:
    ...

Additional async methods:

• client.listen(agent_email=None, reconnect=True, max_backoff=30.0) → AsyncIterator[AsyncInboundEmail] (requires e2a[ws])

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, agent_id, human_id

Exceptions

• E2AError — API error (has status_code and message)