Metadata-Version: 2.4
Name: e2a
Version: 1.4.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: Apache-2.0
Keywords: agent,authentication,e2a,email,webhook
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: Apache Software 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
Requires-Dist: pydantic<3,>=2.12
Provides-Extra: dev
Requires-Dist: anyio[trio]; extra == 'dev'
Requires-Dist: build; extra == 'dev'
Requires-Dist: pytest-httpx; extra == 'dev'
Requires-Dist: pytest>=7; extra == 'dev'
Requires-Dist: pyyaml>=6; extra == 'dev'
Requires-Dist: twine; extra == 'dev'
Provides-Extra: ws
Requires-Dist: websockets>=12; extra == 'ws'
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
```

For WebSocket real-time delivery:

```bash
pip install e2a[ws]
```

## Import paths

The stable, pinned API surface lives under `e2a.v1`:

```python
from e2a.v1 import E2AClient, AsyncE2AClient, E2AApi
```

Top-level `e2a` imports remain available as convenience aliases to the current stable version, but use `e2a.v1` in examples, production code, and version-pinned integrations.

## Quick start

```python
from e2a.v1 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:**
```python
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:**
```python
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}
```

## Raw vs high-level API

The SDK has two layers:

- **`E2AApi`** / **`AsyncE2AApi`** — raw typed HTTP client. Returns generated Pydantic models. Uses `/api/v1/` paths.
- **`E2AClient`** / **`AsyncE2AClient`** — high-level wrapper. Returns parsed `InboundEmail` objects with `.reply()`.

Access the raw layer through `client.api`:

```python
from e2a.v1 import E2AClient

client = E2AClient(api_key="e2a_...")

# High-level: returns InboundEmail with parsed MIME, .reply(), etc.
email = client.get_message("msg_123")

# Raw: returns generated MessageDetail Pydantic model
detail = client.api.get_message("bot@agents.e2a.dev", "msg_123")
```

## Conversation threading

e2a supports an opaque `conversation_id` that lets your agent track multi-turn
threads across the email boundary. Pass it on any `send()` or `reply()`, and
e2a will surface it on the recipient's inbound payload when they respond —
whether the other side is a human replying from Gmail or another e2a agent.

### The basic loop

```python
@app.post("/webhook")
async def webhook(request: Request):
    email = client.parse(await request.body())

    if email.conversation_id:
        # Follow-up — route to the existing conversation
        conversation = get_conversation(email.conversation_id)
    else:
        # First contact — create a new conversation and pick an id for it
        conversation = create_conversation(sender=email.sender)

    response = conversation.generate_reply(email)

    # Tag the reply so future messages in this thread are linked
    email.reply(
        body=response.text,
        html_body=response.html,
        conversation_id=conversation.id,
    )
    return {"ok": True}
```

Same idea for a new outbound:

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

### When is `email.conversation_id` populated?

| Inbound type | Sender passed `conversation_id`? | What you see |
|---|---|---|
| First email from a human (new thread) | n/a — humans don't pass it | `None` — **you must assign one** if you want to thread subsequent messages |
| Human reply to an earlier email from your agent | n/a | The id you passed on your outbound (recovered via `In-Reply-To`) |
| Another e2a agent sending you a new message | **yes, recommended** | The sender's asserted id (carried on a custom header) |
| Another e2a agent sending you a new message | no | `None` |
| Another e2a agent replying to you | either way | Your earlier outbound's id, unless the sender asserted a different one |

Rules of thumb:

- **Always pass `conversation_id`** when you're tagging an outbound as part of a known thread. It's the only way the *recipient's* webhook will see it.
- On first contact from a human, **assign a new id yourself** and stash it before you reply. After that, `email.conversation_id` will keep threading the conversation.
- Don't look up the id from `email.sender` alone — the same person can have many parallel threads.

### Agent-to-agent conversations

If the recipient is another e2a-managed agent, `conversation_id` passed on
`send()` arrives on the recipient's inbound on the very first message — no
prior exchange needed. e2a carries it across on a custom header
(`X-E2A-Conversation-Id`) for same-platform traffic. External senders
(Gmail, Outlook, …) can't forge this header: it's only honored when the
message originates from our own relay.

```python
# Agent A initiates a thread with Agent B
await client_a.send(
    to=["bob@agent.acme.com"],
    subject="Can you handle this?",
    body="Details in the body.",
    conversation_id="task-2026-04-19-7f3a",
)

# Agent B's webhook immediately sees conversation_id="task-2026-04-19-7f3a"
# on the very first message — no round-trip required.
```

### What `conversation_id` is *not*

- Not globally unique; not a primary key in e2a's DB. e2a treats it as an
  opaque string tagged on each message.
- Not a security boundary. Don't rely on it for authentication — check
  `email.auth_headers` for verified sender identity.
- Not guaranteed on every message. Design your code to handle `None`
  (typically: first contact from a human, or an external sender you've
  never interacted with before).

## Attachments

### Receiving attachments

Inbound email attachments are automatically parsed and available on
`email.attachments`:

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

```python
from e2a.v1 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:

```python
from e2a.v1 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.

```bash
pip install e2a[ws]
```

```python
import asyncio
from e2a.v1 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).

The WebSocket protocol is notification-only (server-to-client). The client
never sends application frames.

**Parameters:**

| Parameter | Type | Default | Description |
|-----------|------|---------|-------------|
| `agent_email` | `str` | `client.agent_email` | Agent email to listen for |
| `reconnect` | `bool` | `True` | Auto-reconnect on disconnect |
| `max_backoff` | `float` | `30.0` | Maximum reconnect delay (seconds) |

## Agent and domain management

```python
from e2a.v1 import E2AClient

client = E2AClient(api_key="e2a_...")

# Register a shared-domain agent using a slug (just the local part, not a full email).
# The server appends @agents.e2a.dev automatically.
result = client.register_agent("my-bot")        # slug only, e.g. "my-bot"
print(result.email)  # my-bot@agents.e2a.dev

# Custom domain agent — use the `email` parameter with a full email address.
# The domain must be registered and DNS-verified first.
result = client.register_agent(email="support@mycompany.com", agent_mode="cloud", webhook_url="https://mycompany.com/webhook")

# List agents
agents = client.list_agents()

# Domain management
client.register_domain("mycompany.com")
client.verify_domain("mycompany.com")
client.list_domains()
client.delete_domain("mycompany.com")
```

## Sending emails

Send outbound emails directly:

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

| 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` | Per-delivery target — your agent's address |
| `to` | `list[str]` | Parsed `To:` header — every address from the original message |
| `cc` | `list[str]` | Parsed `Cc:` header (empty when no CCs) |
| `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) |
| `received_at` | `str \| None` | Timestamp when the message was received |
| `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, agent_email=None, base_url="https://e2a.dev")`

High-level sync client. `api_key` falls back to `E2A_API_KEY` env var.

- `client.parse(body)` → `InboundEmail` — accepts bytes, str, dict, or `MessageDetail`
- `client.get_message(message_id)` → `InboundEmail`
- `client.get_messages(status="unread", page_size=50)` → `MessageList`
- `client.reply(message_id, body, ...)` → `SendResult`
- `client.send(to, subject, body, ...)` → `SendResult`
- `client.api` → `E2AApi` (raw typed access)

### `AsyncE2AClient(api_key=None, agent_email=None, base_url="https://e2a.dev")`

Same as `E2AClient` — all I/O methods are `async`. `parse()` is sync (no I/O needed).

- `client.listen(agent_email=None, reconnect=True, max_backoff=30.0)` → `AsyncIterator[AsyncInboundEmail]` (requires `e2a[ws]`)
- `client.api` → `AsyncE2AApi` (raw typed async access)

### 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`, `delegation`, `signature`, `timestamp`

### Exceptions

- `E2AApiError` — API error (has `status_code` and `message`)

## License

Apache-2.0 — see [LICENSE](https://github.com/Mnexa-AI/e2a/blob/main/LICENSE) and [NOTICE](https://github.com/Mnexa-AI/e2a/blob/main/NOTICE) in the upstream repo.
