e2a Python SDK

Python SDK for the e2a protocol — email-to-agent authentication.

Install

pip install e2a

Quick start

from e2a import E2AClient, InboundEmail

client = E2AClient(
    api_key="e2a_your_api_key",
    signing_key="e2a_your_signing_key",
)

@client.on_email
def handle(email: InboundEmail):
    print(f"From: {email.sender}")
    print(f"Subject: {email.subject}")
    print(f"Body: {email.text_body}")
    print(f"Verified: {email.is_verified}")

    # Reply directly from the email object
    email.reply("Thanks for reaching out!")

Mount the handler in your web framework:

FastAPI:

from fastapi import FastAPI, Request

app = FastAPI()

@app.post("/webhook")
async def webhook(request: Request):
    return client.handle_webhook(await request.body(), dict(request.headers))

Flask:

from flask import Flask, request

app = Flask(__name__)

@app.post("/webhook")
def webhook():
    return client.handle_webhook(request.get_data(), dict(request.headers))

That's it — signature verification, payload parsing, and email content extraction are handled automatically.

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.

@client.on_email
def handle(email: InboundEmail):
    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,
    )

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"

InboundEmail

The InboundEmail object passed to your handler has these fields:

Methods:

• email.reply(body, html_body=None, conversation_id=None) → SendResult

Using receive() directly

If you prefer not to use the @client.on_email decorator, you can parse webhooks inline:

@app.post("/webhook")
async def webhook(request: Request):
    email = client.receive(await request.body(), dict(request.headers))

    # Use email.sender, email.subject, email.text_body, etc.
    email.reply("Hello!")

    return {"ok": True}

Low-level API

For full control, the client also exposes the underlying API methods:

# Reply to a message by ID
result = client.reply("msg_123", body="Hello!", html_body="<p>Hello!</p>")

# Send a new email
result = client.send(to="alice@example.com", subject="Hi", body="Hello")

# Verify a webhook signature manually
is_valid = client.verify_webhook(body_bytes, signature_string)

API Reference

E2AClient(api_key, signing_key, base_url="https://e2a.dev")

High-level:

• @client.on_email — register an email handler
• client.handle_webhook(body, headers) → {"ok": True}
• client.receive(body, headers) → InboundEmail

Low-level:

• client.reply(message_id, body, html_body=None, conversation_id=None) → SendResult
• client.send(to, subject, body, content_type=None, conversation_id=None) → SendResult
• client.verify_webhook(body, signature) → bool

Models

• InboundEmail — parsed email with .reply() method
• 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)
• WebhookVerificationError — invalid or missing webhook signature