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

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}

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, 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:
    ...

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)