Loop subscription helpers

Plain helpers for the common Loop subscription tasks. They read the business's Loop connection with get_integration_credentials, call the Loop Subscriptions admin API (2023-10) and return subscriptions as Loop returns them; see the Loop API reference for every field. They raise ValueError if Loop isn't connected and requests.HTTPError if Loop rejects the request.

Helper	What it does
`loop_get_subscription(context, subscription_id)`	The subscription, or `None`.
`loop_find_subscriptions(context, email=None, phone=None)`	The customer's subscriptions, by email.
`loop_cancel_subscription(context, subscription_id)`	Cancels the subscription without notifying the customer.
`loop_next_charge_date(subscription)`	The next billing date as a UTC `datetime`, or `None`.
`loop_set_next_charge_date(context, subscription_id, when)`	Reschedules the next billing to `when`.
`loop_apply_next_charge_discount(context, subscription_id, percentage)`	Discounts the next order by `percentage`.

Connect Loop

Create an API token in your Loop admin with access to subscriptions. See Loop's authentication guide for where to generate it.
In Octocom, open Settings -> Integrations -> Loop, paste the token and save. Installing Loop also makes it the business's subscription provider.
Ask Copilot to check the connection, for example "Look up the Loop subscriptions for [email protected]". Copilot sees Loop in list_integrations and tests the token with a read-only lookup.

The token is stored with the Loop integration, not as an organization secret. Python code reads it as get_integration_credentials(context)[business_slug]["loop"]["apiToken"] and sends it in the X-Loop-Token header, as the helpers below do. Never log it or return it from an action.

Businesses moved off the built-in subscription actions get default actions with the built-in names (getSubscriptionById, getSubscriptionByEmail, cancelSubscription, skipSubscriptionMonths, ...) that call these helpers. They are ordinary Python actions you can read and change.

The full source is below. Copy it into your own action when you need something it doesn't do.

loop_subscription_widget(context, cancel_action="cancel_subscription") returns a ready-to-render subscription panel with confirmation buttons. Pair it with loop_cancel_widget_subscription(context) in a widget-owned action. Setup, business scoping and shared source.

Source

from datetime import datetime, timezone
from urllib.parse import quote

import requests

LOOP_API = "https://api.loopsubscriptions.com/admin/2023-10"


def _loop_request(context, method, path, **kwargs):
    business = context.get("business") or {}
    slug = business.get("slug") or (context.get("conversation") or {}).get("businessSlug")
    if not slug:
        raise ValueError("The business is not available in the context")
    loop = (get_integration_credentials(context).get(slug) or {}).get("loop")
    if not loop:
        raise ValueError("Loop is not connected for this business")
    return requests.request(
        method,
        LOOP_API + path,
        headers={"X-Loop-Token": loop["apiToken"], "Content-Type": "application/json", "Accept": "application/json"},
        timeout=20,
        **kwargs,
    )


def loop_get_subscription(context, subscription_id):
    """The Loop subscription with this ID, or None."""
    subscription_id = str(subscription_id).strip().lstrip("#")
    if not subscription_id:
        return None
    response = _loop_request(context, "GET", f"/subscription/{quote(subscription_id, safe='')}")
    if response.status_code in (400, 404):
        return None
    response.raise_for_status()
    return response.json().get("data")


def loop_find_subscriptions(context, email=None, phone=None):
    """The customer's Loop subscriptions, found by email."""
    if not email:
        return []
    response = _loop_request(context, "GET", f"/customer/{quote(email.strip(), safe='')}/subscription")
    if response.status_code == 404:
        return []
    response.raise_for_status()
    return response.json().get("data") or []


def loop_cancel_subscription(context, subscription_id):
    response = _loop_request(
        context, "POST", f"/subscription/{subscription_id}/cancel", json={"notifyCustomer": False}
    )
    response.raise_for_status()
    if response.content and response.json().get("success") is False:
        raise ValueError("Loop did not accept the cancellation. Refresh the subscription before trying again.")


def loop_subscription_widget(context, cancel_action="cancel_subscription"):
    """Sidebar data for the current customer's Loop subscriptions, with cancel buttons."""
    _, email = _subscription_widget_identity(context)
    items = []
    for sub in loop_find_subscriptions(context, email=email) if email else []:
        lines = sub.get("lines") or []
        billing = sub.get("billingPolicy") or {}
        delivery = sub.get("deliveryPolicy") or {}
        date = loop_next_charge_date(sub)
        currency = sub.get("currencyCode") or ""
        items.append(_subscription_widget_item(
            context, sub["id"], (lines[0].get("productTitle") if lines else None) or "Subscription",
            sub.get("status"), {
                "Next charge": date.strftime("%Y-%m-%d UTC") if date else None,
                "Price": _subscription_widget_value(sub.get("totalLineItemDiscountedPrice"), currency),
                "Billing": _subscription_widget_value(billing.get("intervalCount"), billing.get("interval")),
                "Delivery": _subscription_widget_value(delivery.get("intervalCount"), delivery.get("interval")),
            }, [{"title": str(line.get("productTitle") or "Product"), "fields": {
                "Variant": line.get("variantTitle"), "Quantity": line.get("quantity"),
                "Price": _subscription_widget_value(line.get("price"), currency), "SKU": line.get("sku"),
            }} for line in lines], cancel_action,
        ))
    return {"title": "Subscriptions", "items": items}


def loop_cancel_widget_subscription(context):
    """Action for loop_subscription_widget; validates the current customer before cancelling."""
    return _subscription_widget_cancel(context, loop_find_subscriptions, loop_cancel_subscription)


def loop_next_charge_date(subscription):
    """When the subscription next bills (UTC), or None."""
    epoch = subscription.get("nextBillingDateEpoch")
    return datetime.fromtimestamp(epoch, tz=timezone.utc) if epoch else None


def loop_set_next_charge_date(context, subscription_id, when):
    response = _loop_request(
        context,
        "POST",
        f"/subscription/{subscription_id}/reschedule",
        json={"newBillingDateEpoch": int(when.timestamp())},
    )
    response.raise_for_status()


def loop_apply_next_charge_discount(context, subscription_id, percentage):
    """Discount the subscription's next order by a percentage."""
    response = _loop_request(
        context,
        "POST",
        f"/subscription/{subscription_id}/discount",
        json={"manualDiscount": {"type": "PERCENTAGE", "value": percentage, "orderLimit": 1, "title": "octocom-discount"}},
    )
    response.raise_for_status()

Loop subscription helpers

Connect Loop

Sidebar widget and cancellation

Source

On this page