Skip to main content

Python Integration

Zyphr doesn't have an official Python SDK yet, but the REST API is easy to use with the requests library.

Setup​

pip install requests
import os
import requests

ZYPHR_API_KEY = os.environ["ZYPHR_API_KEY"]
BASE_URL = "https://api.zyphr.dev/v1"

headers = {
    "X-API-Key": ZYPHR_API_KEY,
    "Content-Type": "application/json",
}


def zyphr_request(method, path, json=None, params=None):
    """Make a request to the Zyphr API."""
    response = requests.request(
        method,
        f"{BASE_URL}{path}",
        headers=headers,
        json=json,
        params=params,
    )
    response.raise_for_status()
    return response.json()

Send an Email​

result = zyphr_request("POST", "/emails", json={
    "to": "user@example.com",
    "from": "hello@yourapp.com",
    "subject": "Welcome!",
    "html": "<h1>Hello from Python!</h1>",
    "text": "Hello from Python!",
    "tags": ["welcome", "onboarding"],
})

print(f"Email sent: {result['data']['id']}")

Send with a Template​

result = zyphr_request("POST", "/emails", json={
    "to": "user@example.com",
    "template_id": "welcome-email",
    "template_data": {
        "name": "John",
        "action_url": "https://yourapp.com/activate",
    },
})

Send Batch Emails​

result = zyphr_request("POST", "/emails/batch", json={
    "messages": [
        {"to": "user1@example.com", "subject": "Hello", "html": "<p>Hi!</p>"},
        {"to": "user2@example.com", "subject": "Hello", "html": "<p>Hi!</p>"},
    ]
})

print(f"Succeeded: {result['data']['succeeded']}, Failed: {result['data']['failed']}")

Get Email Status​

result = zyphr_request("GET", "/emails/msg_abc123")
print(f"Status: {result['data']['status']}")

List Emails​

result = zyphr_request("GET", "/emails", params={
    "page": 1,
    "per_page": 25,
    "status": "delivered",
})

for email in result["data"]:
    print(f"{email['id']}: {email['status']}")

print(f"Page {result['meta']['page']} of {result['meta']['total_pages']}")

Send a Push Notification​

result = zyphr_request("POST", "/push", json={
    "user_id": "user_123",
    "title": "New Message",
    "body": "You have a new message",
    "data": {"message_id": "msg_456"},
})

Register a Device​

result = zyphr_request("POST", "/devices", json={
    "user_id": "user_123",
    "token": "fcm_token_here",
    "platform": "android",
})

Send an SMS​

result = zyphr_request("POST", "/sms", json={
    "to": "+14155551234",
    "body": "Your verification code is 123456",
})

print(f"SMS sent: {result['data']['id']}, segments: {result['data']['segments']}")

Send an In-App Notification​

result = zyphr_request("POST", "/inbox/send", json={
    "subscriber_id": "user_123",
    "title": "New Comment",
    "body": "John commented on your post",
    "action_url": "/posts/123#comments",
    "category": "comments",
})

Create a Subscriber​

result = zyphr_request("POST", "/subscribers", json={
    "external_id": "user_123",
    "email": "user@example.com",
    "phone": "+14155551234",
    "name": "John Doe",
    "metadata": {"plan": "pro"},
})

Update Subscriber Preferences​

result = zyphr_request("PUT", "/subscribers/sub_abc/preferences", json={
    "email": {"enabled": True, "categories": {"marketing": False}},
    "push": {"enabled": True},
    "sms": {"enabled": False},
    "in_app": {"enabled": True},
})

Error Handling​

import requests

try:
    result = zyphr_request("POST", "/emails", json={
        "to": "user@example.com",
        "subject": "Hello",
        "html": "<p>Hi!</p>",
    })
except requests.exceptions.HTTPError as e:
    error_data = e.response.json()
    error_code = error_data.get("error", {}).get("code", "unknown")
    error_message = error_data.get("error", {}).get("message", str(e))

    if e.response.status_code == 429:
        retry_after = error_data["error"]["details"]["retry_after"]
        print(f"Rate limited. Retry after {retry_after}s")
    elif e.response.status_code == 422:
        print(f"Validation error: {error_message}")
    elif e.response.status_code == 401:
        print("Invalid API key")
    else:
        print(f"API error ({error_code}): {error_message}")

Async with httpx​

For async applications, use httpx:

pip install httpx
import httpx
import asyncio

async def send_email():
    async with httpx.AsyncClient(
        base_url=BASE_URL,
        headers=headers,
    ) as client:
        response = await client.post("/emails", json={
            "to": "user@example.com",
            "subject": "Hello",
            "html": "<p>Hi from async Python!</p>",
        })
        response.raise_for_status()
        result = response.json()
        print(f"Email sent: {result['data']['id']}")

asyncio.run(send_email())

Django Integration​

# settings.py
ZYPHR_API_KEY = os.environ["ZYPHR_API_KEY"]

# notifications.py
from django.conf import settings
import requests

def send_notification(user, subject, html):
    response = requests.post(
        "https://api.zyphr.dev/v1/emails",
        headers={
            "X-API-Key": settings.ZYPHR_API_KEY,
            "Content-Type": "application/json",
        },
        json={
            "to": user.email,
            "subject": subject,
            "html": html,
        },
    )
    response.raise_for_status()
    return response.json()

Flask Integration​

from flask import Flask, current_app
import requests

app = Flask(__name__)
app.config["ZYPHR_API_KEY"] = os.environ["ZYPHR_API_KEY"]

def send_email(to, subject, html):
    response = requests.post(
        "https://api.zyphr.dev/v1/emails",
        headers={
            "X-API-Key": current_app.config["ZYPHR_API_KEY"],
            "Content-Type": "application/json",
        },
        json={"to": to, "subject": subject, "html": html},
    )
    response.raise_for_status()
    return response.json()

Requirements​

  • Python 3.8 or later
  • requests library (or httpx for async)