aithreads 1.0.0

Official Python SDK for aithreads.io - Email infrastructure for AI agents

aithreads

Official Python SDK for aithreads.io - Email infrastructure for AI agents.

Installation

pip install aithreads

Quick Start

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": "[e-mail]", "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

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": "[e-mail]"}],
        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

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

# 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

# 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="[e-mail]")

# 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

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

# Reply to a thread
client.emails.send(
    inbox_id="inbox-id",
    to=[{"email": "[e-mail]"}],
    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

# 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)

# 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:

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:

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

For async:

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:

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

Requirements

• Python 3.9+
• httpx
• pydantic

License

MIT