Python SDK

pip install pigeons-py

from pigeons import Pigeons

pigeons = Pigeons(
    base_url="https://pigeons.yourdomain.com",
    api_key="pgn_your_key_here",
)

email = pigeons.emails.send(
    to="alice@example.com",
    subject="Welcome!",
    html="<h1>Welcome aboard</h1>",
)

print(email.id)      # msg_abc123...
print(email.status)  # "queued"

email = pigeons.emails.send(
    to="alice@example.com",
    template_id="welcome-email",  # slug or tpl_ ID
    data={"name": "Alice", "company": "Acme"},
)

email = pigeons.emails.send(
    to=["alice@example.com", "bob@example.com"],
    cc=["manager@example.com"],
    subject="Team Update",
    html="<p>Here is your update.</p>",
)

email = pigeons.emails.send(
    to="alice@example.com",
    subject="Reminder",
    text="Your trial expires tomorrow.",
    scheduled_at="2024-06-01T09:00:00Z",
)

email = pigeons.emails.get("msg_abc123...")
print(email.status)  # "delivered"

page = pigeons.emails.list(limit=10, status="delivered")

for email in page.data:
    print(email.id, email.subject, email.status)

if page.has_more:
    next_page = pigeons.emails.list(cursor=page.next_cursor)

cancelled = pigeons.emails.cancel("msg_abc123...")

results = pigeons.emails.batch([
    {"to": "alice@example.com", "subject": "Hi Alice", "text": "Hello!"},
    {"to": "bob@example.com", "subject": "Hi Bob", "text": "Hello!"},
])

email = pigeons.emails.send(
    to="alice@example.com",
    subject="Order #123",
    html="...",
    idempotency_key="order-123-confirmation",
)

contact = pigeons.contacts.create(
    email="alice@example.com",
    first_name="Alice",
    last_name="Smith",
    tags=["newsletter", "beta"],
    properties={"plan": "pro", "signup_source": "website"},
)

# By ID
contact = pigeons.contacts.get("ctc_abc123...")

# By email
contact = pigeons.contacts.get("alice@example.com")

updated = pigeons.contacts.update("ctc_abc123...", tags=["newsletter", "vip"])

page = pigeons.contacts.list(tag="newsletter", limit=50)

pigeons.contacts.delete("ctc_abc123...")

result = pigeons.contacts.batch_create([
    {"email": "alice@example.com", "first_name": "Alice", "tags": ["import"]},
    {"email": "bob@example.com", "first_name": "Bob", "tags": ["import"]},
])
print(result.created)  # 2
print(result.skipped)  # 0

# Global unsubscribe
pigeons.contacts.unsubscribe("ctc_abc123...")

# Per-topic unsubscribe
pigeons.contacts.unsubscribe("ctc_abc123...", topic_id="top_xyz...")

# Resubscribe
pigeons.contacts.resubscribe("ctc_abc123...")

template = pigeons.templates.create(
    slug="welcome-email",
    name="Welcome Email",
    subject="Welcome, {{name}}!",
    body="<h1>Welcome to {{company}}</h1><p>Hi {{name}},</p>",
    sample_data={"name": "Alice", "company": "Acme"},
)

tpl = pigeons.templates.get("welcome-email")  # by slug or ID
updated = pigeons.templates.update("welcome-email", subject="New subject")
pigeons.templates.delete("welcome-email")

preview = pigeons.templates.preview("welcome-email", data={"name": "Alice", "company": "Acme"})
print(preview.subject)  # "Welcome, Alice!"
print(preview.html)     # rendered HTML

campaign = pigeons.campaigns.create(
    name="Product Launch",
    template_id="product-launch",
    template_data={"launch_date": "2024-06-01"},
    audience_filter={"tags": {"contains": "newsletter"}},
)

# Send immediately
pigeons.campaigns.send(campaign.id)

# Or schedule
pigeons.campaigns.send(campaign.id, scheduled_at="2024-06-01T09:00:00Z")

pigeons.campaigns.pause(campaign.id)
pigeons.campaigns.resume(campaign.id)
pigeons.campaigns.cancel(campaign.id)

campaign = pigeons.campaigns.get("cmp_abc123...")
print(campaign.stats)
# {"total": 500, "sent": 450, "delivered": 440, "opened": 120, ...}

sequence = pigeons.sequences.create(
    name="Onboarding",
    steps=[
        {"type": "email", "template_id": "welcome-email", "subject": "Welcome!"},
        {"type": "delay", "delay_duration": "1d"},
        {"type": "email", "template_id": "getting-started", "subject": "Getting Started"},
        {"type": "delay", "delay_duration": "3d"},
        {"type": "email", "template_id": "tips", "subject": "Pro Tips"},
    ],
)

pigeons.sequences.activate(sequence.id)

enrollment = pigeons.sequences.enroll(sequence.id, "ctc_abc123...")
print(enrollment.status)  # "active"

page = pigeons.sequences.enrollments(sequence.id, status="active")

inbox = pigeons.inboxes.create(
    name="Support",
    email_address="support@yourdomain.com",
    webhook_url="https://your-app.com/webhooks/email",
)

pigeons.inboxes.pause(inbox.id)
pigeons.inboxes.resume(inbox.id)
pigeons.inboxes.set_catch_all(inbox.id)
pigeons.inboxes.delete(inbox.id)

messages = pigeons.inboxes.messages.list(inbox.id, direction="inbound", limit=10)

message = pigeons.inboxes.messages.get(inbox.id, "imsg_abc123...")

sent = pigeons.inboxes.messages.create(
    inbox.id,
    to="customer@example.com",
    subject="Re: Your ticket",
    html="<p>Thanks for reaching out.</p>",
    in_reply_to="imsg_previous...",  # maintains thread
)

threads = pigeons.inboxes.threads.list(inbox.id, status="open")

thread = pigeons.inboxes.threads.get(inbox.id, "thr_abc123...")

pigeons.inboxes.threads.update(inbox.id, "thr_abc123...", status="closed")

webhook = pigeons.webhooks.create(
    url="https://your-app.com/webhooks/pigeons",
    events=["email.delivered", "email.bounced", "inbox.message.received"],
)

# The secret is only returned on creation
print(webhook.secret)  # "whsec_..."

domain = pigeons.domains.add(domain="example.com")
print(domain.records)  # DNS records to add

# Re-check verification
verified = pigeons.domains.verify(domain.id)
print(verified.verification_status)  # "verified"

from pigeons import PigeonsError, NotFoundError, AuthError, RateLimitError

try:
    pigeons.emails.get("msg_nonexistent")
except NotFoundError as e:
    print(e.message)      # "Email 'msg_nonexistent' not found"
    print(e.status_code)  # 404
    print(e.code)         # "not_found"
except PigeonsError as e:
    print(f"API error: {e.message}")