aithreads 1.0.0
Official Python SDK for aithreads.io - Email infrastructure for AI agents
Requires Python: >=3.9
Keywords: agents, ai, aithreads, api, email, sdk

# aithreads

Official Python SDK for [aithreads.io](https://aithreads.io) - Email infrastructure for AI agents.

## Installation

```bash
pip install aithreads
```

## Quick Start

```python
from aithreads import AIThreadsClient

# Initialize the client with your API key
client = AIThreadsClient(api_key="ait_xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx")

# List all inboxes
inboxes = client.inboxes.list()
print("Inboxes:", inboxes)

# Create a new inbox
inbox = client.inboxes.create(
    username="support",
    name="Support Agent",
)

# Send an email
email = client.emails.send(
    inbox_id=inbox.id,
    to=[{"email": "user@example.com", "name": "User"}],
    subject="Hello from AI",
    text="This is a test email from my AI agent.",
)

print(f"Email sent: {email.message_id}")
```

## Async Usage

```python
from aithreads import AsyncAIThreadsClient
import asyncio

async def main():
    client = AsyncAIThreadsClient(api_key="ait_xxx...")

    # List inboxes
    inboxes = await client.inboxes.list()

    # Send an email
    email = await client.emails.send(
        inbox_id=inboxes[0].id,
        to=[{"email": "user@example.com"}],
        subject="Hello",
        text="Hi there!",
    )

    await client.close()

asyncio.run(main())
```

## Features

- 📬 **Inbox Management**: Create and manage email inboxes for your AI agents
- 📧 **Email Operations**: Send and receive emails with full threading support
- 🧵 **Thread Management**: Organize conversations with automatic threading
- 🏷️ **Labels**: Categorize threads with custom labels
- 📚 **Knowledge Base**: Upload documents for AI context
- 🔍 **Search**: Find threads by participant email address
- ⚡ **Async Support**: Full async/await support with `AsyncAIThreadsClient`
- 🔒 **Type Safe**: Full type hints with Pydantic models

## API Reference

### Client Initialization

```python
client = AIThreadsClient(
    api_key="ait_...",              # Required: Your API key
    base_url="https://...",         # Optional: API base URL
    timeout=30.0,                   # Optional: Request timeout in seconds
    retries=3,                      # Optional: Number of retries
)
```

### Inboxes

```python
# List all inboxes
inboxes = client.inboxes.list()

# Create an inbox
inbox = client.inboxes.create(
    username="support",
    name="Support Agent",
    webhook_url="https://example.com/webhook",
)

# Get inbox by ID
inbox = client.inboxes.get("inbox-id")

# Update inbox
client.inboxes.update(
    "inbox-id",
    display_name="Updated Name",
    agent_prompt="You are a helpful support agent...",
)

# Delete inbox
client.inboxes.delete("inbox-id")
```

### Threads

```python
# List threads in an inbox
threads = client.threads.list("inbox-id", limit=20, offset=0)

# Find threads by participant email
user_threads = client.threads.find_by_email("inbox-id", email="user@example.com")

# Get thread details
thread = client.threads.get("inbox-id", "thread-id")

# Update thread labels
client.threads.update("inbox-id", "thread-id", labels=["important", "pending"])

# Mark as read/unread
client.threads.mark_as_read("inbox-id", "thread-id")
client.threads.mark_as_unread("inbox-id", "thread-id")

# Archive/unarchive
client.threads.archive("inbox-id", "thread-id")
client.threads.unarchive("inbox-id", "thread-id")
```

### Emails

```python
# Send an email
result = client.emails.send(
    inbox_id="inbox-id",
    to=[{"email": "user@example.com", "name": "User"}],
    cc=[{"email": "cc@example.com"}],
    subject="Hello",
    text="Plain text content",
    html="<p>HTML content</p>",
)

# Reply to a thread
client.emails.send(
    inbox_id="inbox-id",
    to=[{"email": "user@example.com"}],
    subject="Re: Original Subject",
    text="This is a reply",
    reply_to_message_id="original-message-id",
)

# Get emails in a thread
emails = client.emails.list_by_thread("inbox-id", "thread-id")

# Get single email
email = client.emails.get("email-id")
```

### Labels

```python
# Organization-level labels
labels = client.labels.list()
label = client.labels.create(
    name="Important",
    color="#FF5733",
    description="High priority items",
)
client.labels.update("label-id", color="#00FF00")
client.labels.delete("label-id")

# Inbox-specific labels
inbox_labels = client.inbox_labels.list("inbox-id")
client.inbox_labels.create(
    "inbox-id",
    name="Resolved",
    color="#00FF00",
)
```

### Documents (Knowledge Base)

```python
# List documents
docs = client.documents.list("inbox-id")

# Upload a document
with open("document.pdf", "rb") as f:
    doc = client.documents.upload(
        inbox_id="inbox-id",
        file=f.read(),
        filename="document.pdf",
        name="My Document",
    )

# Search documents
results = client.documents.search("inbox-id", query="refund policy")

# Get AI context from documents
context = client.documents.get_context(
    inbox_id="inbox-id",
    query="How do I process refunds?",
    limit=5,
)

# Delete document
client.documents.delete("inbox-id", "doc-id")
```

## Error Handling

The SDK raises typed errors for different scenarios:

```python
from aithreads import (
    AIThreadsError,
    ValidationError,
    AuthenticationError,
    NotFoundError,
    RateLimitError,
)

try:
    client.inboxes.get("invalid-id")
except NotFoundError:
    print("Inbox not found")
except AuthenticationError:
    print("Invalid API key")
except RateLimitError:
    print("Rate limited, try again later")
except AIThreadsError as e:
    print(f"Error: {e.message} ({e.code})")
```

## Context Manager

The client can be used as a context manager to ensure proper cleanup:

```python
with AIThreadsClient(api_key="ait_...") as client:
    inboxes = client.inboxes.list()
# Client is automatically closed
```

For async:

```python
async with AsyncAIThreadsClient(api_key="ait_...") as client:
    inboxes = await client.inboxes.list()
# Client is automatically closed
```

## Type Hints

All types are exported for full type checking support:

```python
from aithreads import (
    Inbox,
    InboxResponse,
    Thread,
    ThreadResponse,
    Email,
    SendEmailRequest,
    Label,
    PaginatedResponse,
)
```

## Requirements

- Python 3.9+
- httpx
- pydantic

## License

MIT