Python SDK

pip install ark-email

pip install ark-email[aiohttp]

import os
from ark import Ark

# Initialize the client
client = Ark(api_key=os.environ.get("ARK_API_KEY"))

# Send an email
email = client.emails.send(
    from_="Security <security@myapp.com>",
    to=["user@example.com"],
    subject="Reset your password",
    html="<h1>Password Reset</h1><p>Click the link below to reset your password.</p>",
    text="Password Reset\n\nClick the link below to reset your password."
)

print(f"Email ID: {email.data.id}")
print(f"Status: {email.data.status}")

from ark import Ark

client = Ark(
    api_key="ark_...",           # Required (or set ARK_API_KEY env var)
    timeout=60.0,                 # Request timeout in seconds (default: 60)
    max_retries=2,                # Number of retry attempts (default: 2)
    base_url="https://api.arkhq.io/v1",  # API base URL (rarely needed)
)

export ARK_API_KEY="ark_live_..."
export ARK_LOG="debug"  # Enable debug logging

import asyncio
from ark import AsyncArk

async def send_welcome_emails(user_emails: list[str]):
    client = AsyncArk()

    # Send emails concurrently
    tasks = [
        client.emails.send(
            from_="welcome@myapp.com",
            to=[email],
            subject="Welcome to MyApp!",
            html="<h1>Welcome!</h1>"
        )
        for email in user_emails
    ]

    results = await asyncio.gather(*tasks)

    for result in results:
        print(f"Sent to {result.data.to[0]}: {result.data.id}")

# Run it
asyncio.run(send_welcome_emails([
    "user1@example.com",
    "user2@example.com",
    "user3@example.com"
]))

# Sync
with Ark() as client:
    client.emails.send(...)

# Async
async with AsyncArk() as client:
    await client.emails.send(...)

from ark import Ark
import ark

client = Ark()

try:
    email = client.emails.send(
        from_="hello@yourdomain.com",
        to=["invalid"],
        subject="Test",
        html="<p>Test</p>"
    )
except ark.BadRequestError as e:
    print(f"Invalid request: {e.message}")
    print(f"Error code: {e.code}")
except ark.AuthenticationError:
    print("Invalid API key")
except ark.RateLimitError as e:
    print(f"Rate limited. Retry after: {e.response.headers.get('Retry-After')}")
except ark.APIConnectionError:
    print("Network error - check your connection")
except ark.APIStatusError as e:
    print(f"API error {e.status_code}: {e.message}")

ark.APIError
├── ark.APIConnectionError      # Network/connection issues
├── ark.APITimeoutError         # Request timeout
└── ark.APIStatusError          # HTTP error responses
    ├── ark.BadRequestError         # 400
    ├── ark.AuthenticationError     # 401
    ├── ark.PermissionDeniedError   # 403
    ├── ark.NotFoundError           # 404
    ├── ark.UnprocessableEntityError # 422
    ├── ark.RateLimitError          # 429
    └── ark.InternalServerError     # 5xx

response = client.emails.with_raw_response.send(
    from_="hello@yourdomain.com",
    to=["user@example.com"],
    subject="Test",
    html="<p>Test</p>"
)

print(f"Status: {response.status_code}")
print(f"Headers: {response.headers}")
print(f"Request ID: {response.headers.get('x-request-id')}")

# Parse the response body
email = response.parse()
print(f"Email ID: {email.data.id}")

with client.emails.with_streaming_response.list() as response:
    for chunk in response.iter_bytes():
        process(chunk)

import httpx
from ark import Ark

# Configure proxy
http_client = httpx.Client(
    proxies="http://proxy.example.com:8080",
    verify=False,  # Skip SSL verification (not recommended for production)
)

client = Ark(http_client=http_client)

import uuid

idempotency_key = str(uuid.uuid4())

# This request can be safely retried with the same key
email = client.emails.send(
    from_="hello@yourdomain.com",
    to=["user@example.com"],
    subject="Order Confirmation",
    html="<p>Your order is confirmed.</p>",
    idempotency_key=idempotency_key
)

email = client.emails.retrieve("msg_abc123")

# Access known fields
print(email.data.subject)

# Access any additional fields returned by the API
print(email.data.model_extra)  # Dict of extra fields

from ark import Ark
from ark.types import EmailSendParams

def send_email(params: EmailSendParams) -> str:
    client = Ark()
    response = client.emails.send(**params)
    return response.data.id

# settings.py
ARK_API_KEY = os.environ.get("ARK_API_KEY")

# emails.py
from django.conf import settings
from ark import Ark

_client = None

def get_ark_client():
    global _client
    if _client is None:
        _client = Ark(api_key=settings.ARK_API_KEY)
    return _client

def send_welcome_email(user):
    client = get_ark_client()
    return client.emails.send(
        from_="welcome@yourapp.com",
        to=[user.email],
        subject=f"Welcome, {user.first_name}!",
        html=render_to_string("emails/welcome.html", {"user": user})
    )

from fastapi import FastAPI, Depends
from ark import AsyncArk
from contextlib import asynccontextmanager

app = FastAPI()
ark_client: AsyncArk | None = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    global ark_client
    ark_client = AsyncArk()
    yield
    await ark_client.close()

app = FastAPI(lifespan=lifespan)

@app.post("/send-email")
async def send_email(to: str, subject: str, body: str):
    email = await ark_client.emails.send(
        from_="noreply@yourapp.com",
        to=[to],
        subject=subject,
        html=body
    )
    return {"email_id": email.data.id}

from flask import Flask, g
from ark import Ark

app = Flask(__name__)

def get_ark():
    if 'ark' not in g:
        g.ark = Ark()
    return g.ark

@app.teardown_appcontext
def close_ark(error):
    ark = g.pop('ark', None)
    if ark is not None:
        ark.close()

@app.route("/send", methods=["POST"])
def send_email():
    client = get_ark()
    email = client.emails.send(...)
    return {"email_id": email.data.id}