SDK Python

pip install labyrinthe-connect

# With a framework helper
pip install "labyrinthe-connect[fastapi]"   # or [flask], [django]

from labyrinthe_connect import AsyncLabyrinthe, Labyrinthe

# From the environment: LABYRINTHE_API_KEY, LABYRINTHE_BASE_URL
with Labyrinthe() as client:
    print(client.is_test_mode)

# Or explicitly
client = Labyrinthe(
    "lsk_test_xxxxxxxxxxxxxxxx",
    timeout=30.0,       # seconds
    max_retries=2,      # 0 to 10
    default_sender="MABOUTIQUE",
)

# Async: same methods, awaited
async with AsyncLabyrinthe() as async_client:
    balance = await async_client.balance.get()

from labyrinthe_connect import Labyrinthe

client = Labyrinthe("lsk_test_xxxxxxxxxxxxxxxx")  # or Labyrinthe() (LABYRINTHE_API_KEY)

message = client.sms.send(
    to="243812345678",
    body="Votre commande 1042 est prête.",
    reference="CMD-1042",
    idempotency_key="order-1042-ready",  # replaying within 24 h sends nothing again
)

print(message.id, message.status)  # queued

from labyrinthe_connect import Labyrinthe

client = Labyrinthe("lsk_test_xxxxxxxxxxxxxxxx")  # or Labyrinthe() (LABYRINTHE_API_KEY)

batch = client.sms.send_many(
    recipients=["243812345678", "0899999999"],
    body="Rappel : reunion demain 9h.",
    sender="MABOUTIQUE",  # an approved Sender ID of the account
    reference="RAPPEL-2026-11-06",
    idempotency_key="reminder-2026-11-06",
)

for message in batch.messages:
    print(message.recipient, message.status)

from labyrinthe_connect import Labyrinthe

client = Labyrinthe("lsk_test_xxxxxxxxxxxxxxxx")  # or Labyrinthe() (LABYRINTHE_API_KEY)

estimate = client.sms.estimate(
    body="Votre code est 482913",
    recipients=["243812345678", "0812345678", "0899999999"],
)

print(estimate.encoding, estimate.recipients_count, estimate.total_credits)
if not estimate.is_affordable:
    pass  # top up before sending

from labyrinthe_connect import Labyrinthe

client = Labyrinthe("lsk_test_xxxxxxxxxxxxxxxx")  # or Labyrinthe() (LABYRINTHE_API_KEY)

balance = client.balance.get()

print(f"{balance.balance} credits available, {balance.held} held")

from labyrinthe_connect import Labyrinthe

client = Labyrinthe("lsk_test_xxxxxxxxxxxxxxxx")  # or Labyrinthe() (LABYRINTHE_API_KEY)

page = client.sms.list(status="delivered", from_="2026-11-01", per_page=50)
print(page.total, page.has_more)

# Iterating a page walks every page
for message in page:
    print(message.id, message.recipient)

from labyrinthe_connect import Labyrinthe, OtpResendTooSoonError

client = Labyrinthe("lsk_test_xxxxxxxxxxxxxxxx")  # or Labyrinthe() (LABYRINTHE_API_KEY)

try:
    otp = client.otp.send(phone="243812345678", locale="fr")
except OtpResendTooSoonError as error:
    print(f"Wait {error.retry_after} s before asking for a new code.")
else:
    session["otp_id"] = otp.otp_id  # keep it server-side
    print(otp.code_length, otp.resend_after_seconds)  # 6 60
    print(otp.test_code)  # "482913" with a lsk_test_ key, None in live

from labyrinthe_connect import Labyrinthe, OtpFailureReason

client = Labyrinthe("lsk_test_xxxxxxxxxxxxxxxx")  # or Labyrinthe() (LABYRINTHE_API_KEY)

result = client.otp.verify(otp_id=session["otp_id"], code=request.form["code"])

if result.verified:
    del session["otp_id"]  # phone verified: log the user in
elif result.reason is OtpFailureReason.INVALID_CODE:
    print(f"Wrong code, {result.attempts_remaining} attempt(s) left")
else:
    print("Ask for a new code:", result.reason)

from typing import Dict

from fastapi import Depends, FastAPI
from labyrinthe_connect import MessageStatusChanged, WebhookEvent
from labyrinthe_connect.contrib.fastapi import WebhookVerifier

app = FastAPI()
verify_webhook = WebhookVerifier()  # LABYRINTHE_WEBHOOK_SECRET; 400 on a bad signature


@app.post("/webhooks/labyrinthe")
async def labyrinthe_webhook(event: WebhookEvent = Depends(verify_webhook)) -> Dict[str, bool]:
    # Deduplicate on event.id
    if isinstance(event, MessageStatusChanged):
        print(event.data.message_id, event.data.status)
    return {"received": True}

# Without a framework helper:
# event = labyrinthe_connect.construct_event(raw_body, request.headers["X-Signature"], secret)

from labyrinthe_connect import (
    ApiError,
    FeatureDisabledError,
    InsufficientCreditsError,
    Labyrinthe,
    RateLimitError,
    RecipientNotSendableError,
)

client = Labyrinthe("lsk_test_xxxxxxxxxxxxxxxx")

try:
    client.sms.send(to="243812345678", body="Bonjour")
except FeatureDisabledError as error:
    # 503 maintenance: nothing was sent, no credit used
    print(error.feature, error.blocked_by, error.admin_message, error.until)
    schedule_retry(error.retry_after) if error.retry_after else queue_for_later()
except InsufficientCreditsError as error:
    notify_top_up(error.available, error.required)
except RecipientNotSendableError as error:
    for rejected in error.rejected:
        print(rejected.recipient, rejected.reason)
except RateLimitError as error:
    schedule_retry(error.retry_after or 60)
except ApiError as error:
    print(error.code, error.status, error.request_id)

# Flask
from labyrinthe_connect.contrib.flask import create_webhook_blueprint

def on_event(event):
    print(event.type, event.id)

app.register_blueprint(create_webhook_blueprint(on_event))  # POST /webhooks/labyrinthe

# Django (urls.py) — secret from settings.LABYRINTHE_WEBHOOK_SECRET
from django.urls import path
from labyrinthe_connect.contrib.django import WebhookView

urlpatterns = [path("webhooks/labyrinthe", WebhookView.as_view())]