Python SDK — LoopNow developer docs

Install

pip install loopnow

Requires Python 3.9 or later. The SDK is pure Python; no C extensions. Dependencies are httpx (for async) and pydantic v2.

Verify the install:

python -c "import loopnow; print(loopnow.__version__)"
# 1.4.2

Quickstart (5 lines)

from loopnow import LoopNow

client = LoopNow(api_key="ln_live_xxxxxxxxxxxxxxxxxxxx")
client.campaigns.send(
    campaign_id="cmp_xyz789"
)
print("Sent!")

That's the whole "send a campaign" flow. The SDK handles auth, retries, pagination, and rate limiting.

Client

The LoopNow class is the entry point. Construct it once at the top of your application and pass it around.

from loopnow import LoopNow

client = LoopNow(
    api_key="ln_live_xxxxxxxxxxxxxxxxxxxx",   # required
    timeout=30,                                # default 30s, in seconds
    max_retries=3,                             # default 3, with exponential backoff
    workspace_id="ws_xyz789",                  # optional, inferred from key if not set
    api_base="https://api.loopnow.in/v1"      # default, override for testing
)

For test environments, use the ln_test_ prefixed key. The sandbox is fully separate from production.

The client exposes six resource namespaces:

• client.contacts — contact CRUD and queries
• client.campaigns — campaign CRUD, send, schedule
• client.automations — automation listing and triggering
• client.transactional — one-off email sends
• client.analytics — delivery, open, click, complaint stats
• client.webhooks — webhook endpoint registration

Contacts

Create a contact

contact = client.contacts.create(
    email="riya@example.com",
    first_name="Riya",
    last_name="Shah",
    list_ids=["aud_newsletter"],
    tags=["vip"],
    consent={
        "source": "homepage_signup",
        "ip": "203.0.113.42",
        "text": "I agree to receive marketing emails."
    }
)
print(contact.id)  # "ct_def456"

The consent argument is required. It writes a tamper-evident consent receipt with the exact text shown, the source URL, the IP, and the user agent. The SDK will refuse to create a contact without it.

List contacts

for contact in client.contacts.list(tags=["vip"], limit=100):
    print(contact.email, contact.first_name)

The list method returns an iterator that handles pagination automatically.

Update a contact

client.contacts.update(
    "ct_def456",
    tags=["vip", "early-access"],
    custom_fields={"plan": "scale"}
)

Delete a contact (right-to-erasure)

client.contacts.delete("ct_def456")

Soft-delete by default. The contact is suppressed from all future sends and excluded from analytics. Hard delete (with full data purge) is available via the dashboard or by setting hard_delete=True.

Campaigns

Create and send

campaign = client.campaigns.create(
    name="January newsletter",
    from_email="hello@yourdomain.in",
    from_name="Your Brand",
    subject="What shipped in January",
    preview_text="3 product updates, 2 case studies, 1 customer story.",
    audience_id="aud_newsletter",
    template_id="tpl_newsletter_v3"
)
client.campaigns.send(campaign.id)

Schedule for later

from datetime import datetime, timezone

client.campaigns.send(
    campaign.id,
    send_at=datetime(2026, 1, 20, 9, 0, tzinfo=timezone.utc)
)

Automations

List automations

for automation in client.automations.list():
    print(automation.id, automation.name, automation.status)

Trigger an automation

client.automations.trigger(
    automation_id="aut_welcome_series",
    contact_email="newuser@example.com",
    context={
        "plan": "scale",
        "trial_days": 14
    }
)

The context dict is exposed to the automation's email templates as merge variables. Reference them in templates as {{context.plan}} and {{context.trial_days}}.

Transactional

Send a transactional email

client.transactional.send(
    to="user@example.com",
    from_email="no-reply@yourdomain.in",
    from_name="Your Brand",
    subject="Your order #1234 is confirmed",
    html="<h1>Thanks for your order</h1>",
    metadata={"order_id": "1234"},
    idempotency_key="order-1234-confirmation"
)

The idempotency_key is the most important argument for transactional sends. It makes the call safe to retry on network failures without risking two emails going out.

Analytics

stats = client.analytics.campaign("cmp_xyz789")
print(f"Sent: {stats.sent}")
print(f"Open rate: {stats.open_rate:.1%}")
print(f"Complaint rate: {stats.complaint_rate:.2%}")

Returns a CampaignStats object with typed accessors for every metric. The data is updated every 60 seconds; for real-time, use the webhook campaign.opened event instead.

Async usage

The SDK ships a fully async variant. Use AsyncLoopNow from the same package.

import asyncio
from loopnow import AsyncLoopNow

async def main():
    client = AsyncLoopNow(api_key="ln_live_xxxxxxxxxxxxxxxxxxxx")

    # Concurrent sends
    results = await asyncio.gather(
        client.transactional.send(to="a@example.com", subject="A", html="..."),
        client.transactional.send(to="b@example.com", subject="B", html="..."),
        client.transactional.send(to="c@example.com", subject="C", html="...")
    )
    print(results)

asyncio.run(main())

The async client uses httpx with HTTP/2 and connection pooling. For 10,000 concurrent transactional sends, the async client will complete in a fraction of the time of the sync client.

Error handling

The SDK raises typed exceptions that you can catch individually:

from loopnow import LoopNow
from loopnow.errors import (
    LoopNowError,         # base
    InvalidRequestError,  # 400
    AuthenticationError,  # 401
    ForbiddenError,       # 403
    NotFoundError,        # 404
    RateLimitError,       # 429
    ServerError,          # 500
)

client = LoopNow(api_key="ln_live_xxx")

try:
    client.campaigns.send("cmp_does_not_exist")
except NotFoundError as e:
    print(f"Campaign {e.request_id} not found")
except RateLimitError as e:
    print(f"Rate limited, retry after {e.retry_after}s")
except LoopNowError as e:
    print(f"Generic error: {e.code} — {e.message}")

For transient errors (RateLimitError, ServerError, network timeouts), the SDK retries automatically with exponential backoff up to max_retries times. Set max_retries=0 to disable.

Webhooks

The SDK includes a webhook handler utility that you can drop into FastAPI, Flask, or Django:

from fastapi import FastAPI, Request
from loopnow.webhooks import WebhookHandler

app = FastAPI()
wh = WebhookHandler(webhook_secret="whsec_xxx")

@app.post("/loopnow/webhooks")
async def handle(request: Request):
    event = await wh.verify_and_parse(request)
    if event.type == "contact.created":
        # sync to your CRM
        sync_to_crm(event.data)
    return {"received": True}

The handler does the signature verification, the timestamp check, and the JSON parsing. The event object is a typed WebhookEvent with a type discriminator and a typed data attribute that depends on the type. Your IDE will autocomplete event.data for each event type.

Analytics deeper dive

For real-time event-level data, the analytics namespace exposes a streaming API that yields events as they happen:

async for event in client.analytics.stream_events(campaign_id="cmp_xyz789"):
    print(f"{event.email} {event.type} at {event.timestamp}")

The stream stays open until you break out of the loop or the campaign completes. This is the recommended pattern for live dashboards.

For historical analytics, you can pull aggregate stats by time range:

stats = client.analytics.campaign(
    "cmp_xyz789",
    start=datetime(2026, 1, 1, tzinfo=timezone.utc),
    end=datetime(2026, 1, 31, tzinfo=timezone.utc),
    granularity="day"
)
for day in stats.timeseries:
    print(f"{day.date}: sent={day.sent}, opens={day.opens}, clicks={day.clicks}")

Type safety with Pydantic

Every return value from the SDK is a Pydantic model. This means you can:

• Access fields with full IDE autocomplete.
• Validate inputs with Pydantic’s validators (e.g. field_validator on a custom field).
• Serialise to JSON with .model_dump() for storage or API responses.
• Deserialise from JSON with Contact.model_validate(data) when reading from a queue or a file.

Testing with the sandbox

The SDK has a built-in test mode for use in unit tests:

import loopnow
from unittest.mock import patch

with patch.object(loopnow, "api_base", "https://api-sandbox.loopnow.in/v1"):
    client = LoopNow(api_key="ln_test_xxx")
    # All calls go to the sandbox, no production data touched
    contact = client.contacts.create(email="test@example.com", ...)

The sandbox is a fully separate workspace with its own data, rate limits, and webhook endpoints. It is safe to use in CI/CD pipelines.

Performance tips for high-volume callers

If you are calling the API at high volume (10,000+ requests / minute), three things will materially improve throughput.

1. Use the async client. The sync client is one request at a time. The async client uses connection pooling and HTTP/2. The throughput difference is roughly 10x on a single host.
2. Batch operations where possible. contacts.create_many() accepts up to 500 contacts in one call, vs 500 individual contacts.create() calls. The throughput difference is 50x.
3. Use streaming for large list responses. contacts.list() returns an iterator that fetches pages lazily. For a 100,000-contact list, this keeps memory flat.

Benchmark: on a 4-core m6i.xlarge AWS instance, the async client can sustain 8,000 contacts / second for bulk create. The sync client tops out at around 80 / second.

Open source

The SDK is open source under the MIT licence. Source: github.com/loopnow/loopnow-python. Issues, feature requests, and PRs welcome.