emailingagents

Python SDK

The Python client lives in python/ in the repository and is not published to PyPI yet. It needs Python 3.9 or newer and depends only on httpx. It covers the whole /v0 API, including pods, lists, suppressions, domains, search, metrics and billing reads.

pip install ./python                  # core
pip install "./python[websocket]"     # adds websockets for client.websockets.connect()

Quickstart#

from emailingagents import Client

client = Client(api_key="am_...", base_url="http://localhost:3000")
inbox = client.inboxes.create(username="support", display_name="Support Agent", client_id="support-v1")
sent = client.inboxes.messages.send(inbox.inbox_id, to="user@example.com", subject="Hello", text="Hi there", labels=["outreach"])
print(sent.message_id, sent.thread_id)

for message in client.inboxes.messages.list_all(inbox.inbox_id):
    print(message.subject)   # list responses carry a preview; call messages.get(...) for extracted_text

api_key and base_url fall back to the EMAILINGAGENTS_API_KEY and EMAILINGAGENTS_BASE_URL environment variables. Always set the base URL; the default is http://localhost:3000. Responses are dictionaries that also allow attribute access (inbox.inbox_id), with the snake_case wire names; a key that is a Python keyword gets a trailing underscore (message.from_).

Resources#

inboxes (with .messages, .threads, .drafts, .webhooks, .lists), threads, drafts, webhooks, pods (with .inboxes, .webhooks, .domains, .api_keys, .lists), api_keys, domains, lists, suppressions, billing, metrics (usage, events, rates), events and agent. Methods that return a page also have a list_all variant that follows next_page_token for you. Send, reply, reply-all, forward and draft calls accept track_opens=True when the backend supports open tracking.

draft = client.inboxes.drafts.create(inbox.inbox_id, to="a@example.com", subject="Later", text="...")
client.inboxes.drafts.send(inbox.inbox_id, draft.draft_id)
client.inboxes.messages.update(inbox.inbox_id, sent.message_id, add_labels=["resolved"], remove_labels=["unread"])
hits = client.inboxes.messages.search(inbox.inbox_id, "demo next week")
usage = client.metrics.usage(usage_types=["message_count"], period=3600)

Async#

AsyncClient has the same surface. Every call is awaited and list_all is an async iterator.

import asyncio
from emailingagents import AsyncClient

async def main():
    async with AsyncClient(api_key="am_...", base_url="http://localhost:3000") as client:
        async for thread in client.threads.list_all(limit=50):
            print(thread.subject)

asyncio.run(main())

Errors#

Non-2xx responses raise a subclass of ApiError carrying status, code, message, fix and body: ValidationError (400), AuthenticationError (401), PermissionError_ (403), NotFoundError (404), ConflictError (409, or code already_exists, which the API returns as 403), UnprocessableError (422), RateLimitError (429, with retry_after seconds) and ServerError (5xx). Connection failures raise TransportError. See errors for the codes.

from emailingagents import NotFoundError

try:
    client.inboxes.get("missing")
except NotFoundError as e:
    print(e.status, e.code, e.fix)

Retries and idempotency#

The client retries up to max_retries times (default 2) with exponential backoff. A 429 is retried for every call and honours Retry-After. Network errors and 500, 502, 503 and 504 are retried only when replaying is safe: GET and DELETE, or a call that carries idempotency_key= (sent as the Idempotency-Key header) or a client_id.

client.inboxes.messages.send(inbox.inbox_id, to="a@example.com", text="Hi", idempotency_key="order-4821-confirmation")

Other options: timeout (seconds, default 30), default_headers, http_client and transport for a custom httpx setup.

Verify webhooks#

Pass the raw request body, the headers and the whsec_ secret that webhooks.create returned once. See webhooks for the scheme.

from emailingagents import verify_webhook, WebhookVerificationError

try:
    event = verify_webhook(request.get_data(), request.headers, secret)
except WebhookVerificationError:
    return "bad signature", 400
print(event.event_type)

A valid signature can be replayed inside the tolerance window, so dedupe on svix-id (retries of one event reuse it). Keep seen ids for at least the tolerance:

event = verify_webhook(body, headers, secret)
if not seen.add_if_new(headers["svix-id"], ttl=600):   # e.g. redis SET NX EX 600, or a unique DB column
    return "", 200                                      # already handled
handle(event)

Websockets#

Install the websocket extra, then:

with client.websockets.connect(inbox_ids=[inbox.inbox_id], event_types=["message.received"]) as socket:
    for event in socket:
        print(event.event_type)

The async client uses async with ... as socket and async for event in socket. The key is sent in an Authorization: Bearer header so it stays out of URLs and proxy logs; pass api_key_in_query=True to connect() only if you cannot set headers. See websockets.

To give an AI agent these capabilities as tools, see the LangChain integration.

← TypeScript SDKLangChain integration →