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="alice@example.com",
    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="alice@example.com",
    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("my-bot@agents.e2a.dev"):
            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:

Sending emails

Send outbound emails directly:

result = client.send(
    to="alice@example.com",
    subject="Hello from my agent",
    body="Hi Alice!",
    conversation_id="conv_abc123",  # optional
)
print(result.status, result.message_id)

InboundEmail

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)