Python SDK

pip install zend-sh

from zend_sh import ZendClient

client = ZendClient(api_key="zk_live_...")

from zend_sh import AsyncZendClient

client = AsyncZendClient(api_key="zk_live_...")

# Use with async/await
accounts = await client.accounts.list()

# List all accounts
result = client.accounts.list()
for account in result.accounts:
    print(account.email, account.health_score)

# Connect a new mailbox
account = client.accounts.create(
    email="outreach@yourcompany.com",
    provider="google",
    smtp_host="smtp.gmail.com",
    smtp_port=465,
    smtp_username="outreach@yourcompany.com",
    smtp_password="your-app-password",
    imap_host="imap.gmail.com",
    imap_port=993,
    imap_username="outreach@yourcompany.com",
    imap_password="your-app-password",
)

# Get, update, delete
account = client.accounts.get("account-id")
account = client.accounts.update("account-id", daily_send_limit=50)
client.accounts.delete("account-id")

# Send a test email
result = client.accounts.test("account-id")

# List campaigns
result = client.campaigns.list()

# Create a campaign
campaign = client.campaigns.create(
    name="Q1 Outreach",
    stop_on_reply=True,
    send_window={
        "timezone": "America/New_York",
        "days": ["monday", "tuesday", "wednesday", "thursday", "friday"],
        "start_hour": 9,
        "end_hour": 17,
    },
)

# Get, update, delete
campaign = client.campaigns.get(campaign.id)
campaign = client.campaigns.update(campaign.id, name="Updated Name")
client.campaigns.delete(campaign.id)  # only draft/archived

# Start and pause
client.campaigns.start(campaign.id)
client.campaigns.pause(campaign.id)

# List steps
result = client.campaigns.steps.list(campaign.id)

# Add an email step
client.campaigns.steps.create(
    campaign.id,
    type="email",
    position=1,
    subject="Hi {{first_name|there}}",
    body="Your email body with {{company}} variables...",
)

# Add a delay step
client.campaigns.steps.create(
    campaign.id,
    type="delay",
    position=2,
    delay_days=3,
)

# Add leads
result = client.campaigns.leads.add(
    campaign.id,
    leads=[
        {"email": "jane@acme.com", "first_name": "Jane", "company": "Acme"},
        {"email": "bob@globex.com", "first_name": "Bob", "company": "Globex"},
    ],
)
print(f"Enrolled: {result.enrolled}")

# List enrolled leads
result = client.campaigns.leads.list(campaign.id)

# Import with duplicate handling
result = client.campaigns.leads.import_leads(
    campaign.id,
    leads=[{"email": "jane@acme.com", "first_name": "Jane"}],
)

# Team-wide overview
overview = client.analytics.overview()
print(f"Sent: {overview.sent}, Replied: {overview.replied}")

# Per-campaign analytics
stats = client.analytics.campaign(campaign.id)

# Per-account health
health = client.analytics.account("account-id")
print(f"Health score: {health.health_score}")

# Create a webhook
webhook = client.webhooks.create(
    url="https://yourapp.com/webhooks/zend",
    events=["email.sent", "email.bounced", "email.replied"],
)

# List, update, delete
result = client.webhooks.list()
client.webhooks.update(webhook.id, events=["email.replied"])
client.webhooks.delete(webhook.id)

result = client.events.list(type="email.replied", limit=50)
for event in result.events:
    print(event.type, event.created_at)

# Add a suppression
client.suppressions.create(email="unsubscribed@example.com")

# Check if suppressed
result = client.suppressions.check("user@example.com")
print(result.suppressed)  # True or False

# List and delete
result = client.suppressions.list()
client.suppressions.delete("unsubscribed@example.com")

# Create a new key
result = client.keys.create(name="Production Key")
print(result.key)  # 'zk_live_...' (shown only once)

# List keys (masked)
result = client.keys.list()

# Revoke a key
client.keys.delete("key-id")

import asyncio
from zend_sh import AsyncZendClient

async def main():
    client = AsyncZendClient(api_key="zk_live_...")

    # All the same methods, just with await
    result = await client.accounts.list()
    campaign = await client.campaigns.create(
        name="Async Campaign",
        stop_on_reply=True,
    )
    await client.campaigns.start(campaign.id)

asyncio.run(main())

from zend_sh.types import Account, Campaign, Lead, AnalyticsOverview

# Response objects have typed attributes
account: Account = client.accounts.get("account-id")
print(account.email)        # str
print(account.health_score) # float
print(account.provider)     # Literal["google", "microsoft", "smtp"]

from zend_sh import ZendClient, ZendError

client = ZendClient(api_key="zk_live_...")

try:
    client.campaigns.start("nonexistent-id")
except ZendError as e:
    print(e.type)    # 'not_found'
    print(e.message) # 'Campaign not found'
    print(e.status)  # 404

client = ZendClient(
    api_key="zk_live_...",
    base_url="https://app.zend.sh",  # optional, defaults to production
)