commune-mail 0.1.0

Python SDK for Commune – email infrastructure for agents. Threads, inboxes, domains, attachments, and sending.

Commune Python SDK

Python SDK for Commune — email infrastructure for AI agents.

pip install commune-ai

---

Quickstart

From zero to a working email agent in 4 lines — no domain setup, no DNS:

from commune import CommuneClient

client = CommuneClient(api_key="comm_...")

# Create an inbox — domain is auto-assigned
inbox = client.inboxes.create(local_part="support")
print(f"Inbox ready: {inbox.address}")  # → "[e-mail]"

# List email threads
threads = client.threads.list(inbox_id=inbox.id, limit=5)
for t in threads.data:
    print(f"  [{t.message_count} msgs] {t.subject}")

# Send an email
client.messages.send(
    to="[e-mail]",
    subject="Hello from my agent",
    text="Hi there!",
)

That's it. No domain verification, no DNS records. Just create an inbox and start sending/receiving.

---

Concepts

Commune organizes email around four layers:

Domain  →  Inbox  →  Thread  →  Message

• Domain — A custom email domain you own (e.g. example.com). You verify it by adding DNS records.
• Inbox — A mailbox under a domain (e.g. [e-mail]). Each inbox can have webhooks for real-time notifications.
• Thread — A conversation: a group of related messages sharing a subject/reply chain. Called conversation_id internally, exposed as thread_id in the SDK.
• Message — A single email (inbound or outbound) within a thread.

---

Client

from commune import CommuneClient

client = CommuneClient(
    api_key="comm_...",     # Required. Your API key.
    base_url=None,          # Optional. Override API URL.
    timeout=30.0,           # Optional. Request timeout in seconds.
)

Supports context manager:

with CommuneClient(api_key="comm_...") as client:
    domains = client.domains.list()
# Connection closed automatically

---

Domains

Domains are the foundation. You register a domain, add DNS records, verify it, then create inboxes under it.

client.domains.list()

List all domains in your organization.

domains = client.domains.list()
# → [Domain(id="d_abc123", name="example.com", status="verified", ...)]

Returns: list[Domain]

Field	Type	Description
id	str	Domain ID
name	str	Domain name
status	str	"not_started", "pending", "verified", "failed"
region	str	AWS region
records	list	DNS records (MX, TXT, CNAME)
inboxes	list[Inbox]	Inboxes under this domain

client.domains.create(name, region=None)

Register a new domain. After creating, you'll need to verify it.

domain = client.domains.create(name="example.com")
print(domain.id)      # → "d_abc123"
print(domain.status)  # → "not_started"

Parameter	Type	Required	Description
name	str	Yes	Domain name (e.g. "example.com")
region	str	No	AWS region (e.g. "us-east-1")

client.domains.get(domain_id)

Get full details for a single domain.

domain = client.domains.get("d_abc123")

client.domains.records(domain_id)

Get the DNS records you need to add at your registrar.

records = client.domains.records("d_abc123")
for r in records:
    print(f"  {r['type']} {r['name']} → {r['value']}")

Returns: list[dict] — each record has type, name, value, status, ttl.

client.domains.verify(domain_id)

Trigger verification after you've added the DNS records.

result = client.domains.verify("d_abc123")

Typical flow

# 1. Create the domain
domain = client.domains.create(name="example.com")

# 2. Get DNS records to configure
records = client.domains.records(domain.id)
print("Add these DNS records at your registrar:")
for r in records:
    print(f"  {r['type']} {r['name']} → {r['value']}")

# 3. After adding records, verify
result = client.domains.verify(domain.id)

# 4. Check status
domain = client.domains.get(domain.id)
print(f"Status: {domain.status}")  # → "verified"

---

Inboxes

Inboxes are mailboxes that receive and send email. Create one with just a local_part — the domain is auto-assigned.

client.inboxes.create(local_part, *, domain_id=None, name=None, webhook=None)

Create a new inbox. Domain is auto-resolved if not provided — no DNS setup needed.

# Simplest — domain auto-assigned
inbox = client.inboxes.create(local_part="support")
print(inbox.address)  # → "[e-mail]"

# Explicit domain (if you have a custom domain)
inbox = client.inboxes.create(local_part="billing", domain_id="d_abc123")

Parameter	Type	Required	Description
local_part	str	Yes	Part before @ (e.g. "support", "billing")
domain_id	str	No	Domain to create under. Auto-resolved if omitted.
name	str	No	Display name
webhook	dict	No	{"endpoint": "https://...", "events": ["inbound"]}

Returns: Inbox

Field	Type	Description
id	str	Inbox ID
local_part	str	Part before @
address	str	Full email address
webhook	InboxWebhook | str | None	Webhook configuration
status	str | None	Inbox status
created_at	str | None	ISO timestamp

client.inboxes.list(domain_id=None)

List inboxes. Without domain_id, lists all inboxes across all domains.

# All inboxes
inboxes = client.inboxes.list()

# Inboxes for a specific domain
inboxes = client.inboxes.list(domain_id="d_abc123")

client.inboxes.get(domain_id, inbox_id)

inbox = client.inboxes.get("d_abc123", "i_xyz")

client.inboxes.update(domain_id, inbox_id, **fields)

Update one or more fields. Only provided fields are changed.

inbox = client.inboxes.update("d_abc123", "i_xyz", local_part="help")

client.inboxes.set_webhook(domain_id, inbox_id, *, endpoint, events=None)

Shortcut to set a webhook. You'll receive a POST when emails arrive.

client.inboxes.set_webhook(
    "d_abc123", "i_xyz",
    endpoint="https://your-app.com/webhook",
    events=["inbound"],
)

client.inboxes.remove(domain_id, inbox_id)

Delete an inbox permanently.

client.inboxes.remove("d_abc123", "i_xyz")  # → True

---

Threads

A thread is a conversation — a group of related email messages. Threads are listed with cursor-based pagination for efficient browsing of large mailboxes.

client.threads.list(*, inbox_id=None, domain_id=None, limit=20, cursor=None, order="desc")

List threads for an inbox or domain. Returns newest first by default.

result = client.threads.list(inbox_id="i_xyz", limit=10)

for thread in result.data:
    print(f"[{thread.message_count} msgs] {thread.subject}")
    print(f"  Last activity: {thread.last_message_at}")
    print(f"  Preview: {thread.snippet}")

# Paginate
if result.has_more:
    page2 = client.threads.list(inbox_id="i_xyz", cursor=result.next_cursor)

Parameter	Type	Required	Description
inbox_id	str	One of these	Filter by inbox
domain_id	str	required	Filter by domain
limit	int	No	1–100, default 20
cursor	str	No	Cursor from previous next_cursor
order	str	No	"desc" (newest first) or "asc"

Returns: ThreadList

ThreadList(
    data=[Thread(...)],   # List of thread summaries
    next_cursor="abc...", # Pass to next call for next page (None if no more)
    has_more=True,        # Whether more pages exist
)

Thread object:

Field	Type	Description
thread_id	str	Thread identifier
subject	str | None	Email subject
last_message_at	str	ISO timestamp of last message
first_message_at	str | None	ISO timestamp of first message
message_count	int	Total messages in thread
snippet	str | None	Preview of last message (up to 200 chars)
last_direction	str | None	"inbound" or "outbound"
inbox_id	str | None	Inbox this thread belongs to
domain_id	str | None	Domain this thread belongs to
has_attachments	bool	Whether any message has attachments

client.threads.messages(thread_id, *, limit=50, order="asc")

Get all messages in a thread. Returns oldest first by default (chronological reading order).

messages = client.threads.messages("conv_abc123")

for msg in messages:
    sender = next((p.identity for p in msg.participants if p.role == "sender"), "unknown")
    print(f"  [{msg.direction}] From: {sender}")
    print(f"  Subject: {msg.metadata.subject}")
    print(f"  {msg.content[:200]}")
    print()

Parameter	Type	Required	Description
thread_id	str	Yes	Thread ID
limit	int	No	1–1000, default 50
order	str	No	"asc" (chronological) or "desc"

Returns: list[Message]

Message object:

Field	Type	Description
message_id	str	Unique message identifier
conversation_id	str	Thread ID this message belongs to
direction	str	"inbound" or "outbound"
participants	list[Participant]	[{role: "sender", identity: "user@..."}, ...]
content	str	Plain text body
content_html	str | None	HTML body
attachments	list[str]	Attachment IDs
created_at	str	ISO timestamp
metadata.subject	str	Subject line
metadata.inbox_id	str	Inbox ID

---

Messages

client.messages.send(**kwargs)

Send an email. Returns the sent message data.

result = client.messages.send(
    to="[e-mail]",
    subject="Order Confirmation",
    html="<h1>Thanks for your order!</h1><p>Your order #1234 is confirmed.</p>",
)

Parameter	Type	Required	Description
to	str | list[str]	Yes	Recipient(s)
subject	str	Yes	Subject line
html	str	No*	HTML body
text	str	No*	Plain text body
from_address	str	No	Sender (uses domain default)
cc	list[str]	No	CC recipients
bcc	list[str]	No	BCC recipients
reply_to	str	No	Reply-to address
thread_id	str	No	Reply in existing thread
domain_id	str	No	Send from specific domain
inbox_id	str	No	Send from specific inbox
attachments	list[str]	No	Attachment IDs
headers	dict[str, str]	No	Custom headers

*At least one of html or text is required.

Reply to a thread:

client.messages.send(
    to="[e-mail]",
    subject="Re: Order Issue",
    html="<p>We're looking into this for you.</p>",
    thread_id="conv_abc123",  # continues the thread
    inbox_id="i_xyz",
)

client.messages.list(**kwargs)

List messages with filters. Provide at least one of inbox_id, domain_id, or sender.

messages = client.messages.list(
    inbox_id="i_xyz",
    limit=20,
    order="desc",
    after="2025-01-01T00:00:00Z",
)

Parameter	Type	Required	Description
inbox_id	str	One of	Filter by inbox
domain_id	str	these	Filter by domain
sender	str	required	Filter by sender email
limit	int	No	1–1000, default 50
order	str	No	"asc" or "desc" (default)
before	str	No	ISO date — messages before this time
after	str	No	ISO date — messages after this time

---

Attachments

Upload files, then reference them when sending emails.

client.attachments.upload(content, filename, mime_type)

Upload a file. Returns an attachment_id you pass to messages.send().

import base64

with open("invoice.pdf", "rb") as f:
    content = base64.b64encode(f.read()).decode()

upload = client.attachments.upload(
    content=content,
    filename="invoice.pdf",
    mime_type="application/pdf",
)
print(upload.attachment_id)  # → "att_abc123"
print(upload.size)           # → 45230

Parameter	Type	Required	Description
content	str	Yes	Base64-encoded file data
filename	str	Yes	Original filename
mime_type	str	Yes	MIME type

Returns: AttachmentUpload

Field	Type	Description
attachment_id	str	ID to use in messages.send()
filename	str	Filename
mime_type	str	MIME type
size	int	Size in bytes

client.attachments.get(attachment_id)

Get metadata for an uploaded attachment.

att = client.attachments.get("att_abc123")
print(att.filename, att.mime_type, att.size)

client.attachments.url(attachment_id, *, expires_in=3600)

Get a temporary download URL.

url_info = client.attachments.url("att_abc123", expires_in=7200)
print(url_info.url)         # → "https://..."
print(url_info.expires_in)  # → 7200

Returns: AttachmentUrl

Field	Type	Description
url	str	Temporary download URL
expires_in	int	Seconds until URL expires
filename	str	Filename
mime_type	str	MIME type
size	int	Size in bytes

Full attachment flow

import base64

# 1. Upload the file
with open("report.pdf", "rb") as f:
    content = base64.b64encode(f.read()).decode()

upload = client.attachments.upload(content, "report.pdf", "application/pdf")

# 2. Send email with attachment
client.messages.send(
    to="[e-mail]",
    subject="Monthly Report",
    html="<p>Please find the report attached.</p>",
    attachments=[upload.attachment_id],
)

# 3. Later, get a download URL for that attachment
url_info = client.attachments.url(upload.attachment_id)
print(f"Download: {url_info.url}")

---

Error Handling

All errors inherit from CommuneError. Catch specific types or the base class.

from commune import (
    CommuneClient,
    CommuneError,
    AuthenticationError,
    NotFoundError,
    ValidationError,
    RateLimitError,
)

try:
    client = CommuneClient(api_key="comm_...")
    domain = client.domains.get("nonexistent")
except AuthenticationError:
    # 401 — invalid or expired API key
    print("Check your API key")
except NotFoundError:
    # 404 — resource doesn't exist
    print("Domain not found")
except ValidationError as e:
    # 400 — bad request parameters
    print(f"Invalid request: {e.message}")
except RateLimitError:
    # 429 — too many requests
    print("Slow down, try again in a moment")
except CommuneError as e:
    # Catch-all for any API error
    print(f"Error ({e.status_code}): {e.message}")

Exception	HTTP Status	When
AuthenticationError	401	Invalid/expired API key
ValidationError	400	Bad request parameters
NotFoundError	404	Resource doesn't exist
RateLimitError	429	Too many requests
CommuneError	Any	Base class for all errors

---

License

MIT