Metadata-Version: 2.4
Name: e2a
Version: 0.1.0
Summary: Python SDK for the e2a protocol — email-to-agent authentication
Project-URL: Homepage, https://e2a.dev
Project-URL: Repository, https://github.com/Mnexa-AI/e2a
Project-URL: Documentation, https://e2a.dev
Author-email: Mnexa AI <josh@mnexa.ai>
License-Expression: MIT
Keywords: agent,authentication,e2a,email,webhook
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Communications :: Email
Requires-Python: >=3.9
Requires-Dist: httpx>=0.24
Provides-Extra: dev
Requires-Dist: build; extra == 'dev'
Requires-Dist: pytest-httpx; extra == 'dev'
Requires-Dist: pytest>=7; extra == 'dev'
Requires-Dist: twine; extra == 'dev'
Description-Content-Type: text/markdown

# e2a Python SDK

Python SDK for the [e2a protocol](https://e2a.dev) — email-to-agent authentication.

## Install

```bash
pip install e2a
```

## Quick start

### Verify and parse incoming webhooks

```python
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

```python
# 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

```python
result = client.send(
    to="alice@example.com",
    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`.

```python
@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.

```python
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"
```

### FastAPI example

```python
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`
