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

Verify and parse incoming webhooks

from e2a import E2AClient, parse_payload

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

# In your webhook handler (Flask, FastAPI, etc.)
def handle_webhook(request):
    body = request.get_data()
    signature = request.headers["X-E2A-Webhook-Signature"]

    if not client.verify_webhook(body, signature):
        return "invalid signature", 401

    payload = parse_payload(request.json, dict(request.headers))

    print(f"From: {payload.sender}")
    print(f"To: {payload.recipient}")
    print(f"Verified: {payload.auth.verified}")
    print(f"Message ID: {payload.message_id}")

Reply to an email

# Use the message_id from the webhook payload
result = client.reply(
    message_id=payload.message_id,
    body="Thanks for your email!",
    html_body="<p>Thanks for your email!</p>",  # optional
)
print(f"Sent via {result.method}, ID: {result.message_id}")

Send a new email

result = client.send(
    to="[e-mail]",
    subject="Hello from my agent",
    body="This is a message from an AI agent.",
)

Conversation threading

e2a supports an opaque conversation_id that lets your agent track multi-turn email threads. This is useful when your agent system has its own concept of conversations and needs to route follow-up emails to the right one.

How it works:

1. When your agent replies or sends an email, pass a conversation_id. e2a stores a mapping from the outgoing email's Message-ID to your conversation ID.

2. When a human replies to that email, e2a matches the In-Reply-To header against stored Message-IDs and includes conversation_id in the webhook payload.

3. For first-contact emails (no prior thread), conversation_id is None.

@app.post("/webhook")
async def webhook(request: Request):
    body = await request.body()
    signature = request.headers.get("X-E2A-Webhook-Signature", "")

    if not e2a.verify_webhook(body, signature):
        raise HTTPException(401, "invalid signature")

    payload = parse_payload(await request.json(), dict(request.headers))

    if payload.conversation_id:
        # This is a follow-up — route to the existing conversation
        conversation = get_conversation(payload.conversation_id)
    else:
        # First contact — create a new conversation
        conversation = create_conversation(sender=payload.sender)

    response = conversation.generate_reply(payload)

    # Tag the reply with your conversation ID so future emails are linked
    e2a.reply(
        payload.message_id,
        body=response.text,
        html_body=response.html,
        conversation_id=conversation.id,
    )

    return {"ok": True}

The same works for client.send() — pass conversation_id when initiating an outbound email, and any reply to it will arrive with that ID in the webhook.

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"

FastAPI example

from fastapi import FastAPI, Request, HTTPException
from e2a import E2AClient, parse_payload

app = FastAPI()
e2a = E2AClient(api_key="e2a_...", signing_key="e2a_...")

@app.post("/webhook")
async def webhook(request: Request):
    body = await request.body()
    signature = request.headers.get("X-E2A-Webhook-Signature", "")

    if not e2a.verify_webhook(body, signature):
        raise HTTPException(401, "invalid signature")

    data = await request.json()
    payload = parse_payload(data, dict(request.headers))

    # Process the email...
    response = process_email(payload)

    # Reply
    if payload.message_id:
        e2a.reply(payload.message_id, body=response)

    return {"ok": True}

API Reference

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

• 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

verify_signature(body, signature, signing_key) → bool

parse_payload(data, headers) → WebhookPayload

Models

• WebhookPayload — message_id, conversation_id, sender, recipient, raw_message, auth, received_at
• AuthHeaders — verified, sender, entity_type, domain_check, agent_id, human_id
• SendResult — status, message_id, method
• E2AError — raised on API errors, has status_code and message