Python SDK

pip install requests

import os
import requests
from typing import Optional, List, Dict, Any
 
# Configuration
LORN_BASE_URL = os.getenv("LORN_BASE_URL", "https://{{YOUR_STORE_URL}}")
LORN_API_KEY = os.getenv("LORN_API_KEY", "{{YOUR_API_KEY}}")
 
class LornClient:
    """Lorn AI API client."""
    
    def __init__(self, base_url: str = None, api_key: str = None):
        self.base_url = base_url or LORN_BASE_URL
        self.api_key = api_key or LORN_API_KEY
        self.session = requests.Session()
        self.session.headers.update({
            "X-ACP-API-Key": self.api_key,
            "Accept": "application/json",
            "Content-Type": "application/json"
        })
    
    def _request(self, method: str, endpoint: str, **kwargs) -> Dict[str, Any]:
        """Make an API request."""
        url = f"{self.base_url}{endpoint}"
        response = self.session.request(method, url, **kwargs)
        response.raise_for_status()
        return response.json()
    
    # Health
    def health(self) -> Dict[str, str]:
        """Check API health."""
        return self._request("GET", "/health")
    
    # Products
    def search_products(
        self,
        q: str = None,
        category: str = None,
        tag: str = None,
        min_price: float = None,
        max_price: float = None,
        page: int = 1,
        page_size: int = 10
    ) -> Dict[str, Any]:
        """Search for products."""
        params = {"page": page, "page_size": page_size}
        if q:
            params["q"] = q
        if category:
            params["category"] = category
        if tag:
            params["tag"] = tag
        if min_price is not None:
            params["min_price"] = min_price
        if max_price is not None:
            params["max_price"] = max_price
        
        return self._request("GET", "/acp/products", params=params)
    
    def get_product(self, product_id: str) -> Dict[str, Any]:
        """Get product by ID."""
        return self._request("GET", f"/acp/products/{product_id}")
    
    # Checkout
    def create_checkout(
        self,
        items: List[Dict[str, Any]],
        shipping_address: Dict[str, Any] = None,
        customer: Dict[str, Any] = None
    ) -> Dict[str, Any]:
        """Create a checkout session."""
        payload = {"items": items}
        if shipping_address:
            payload["shipping_address"] = shipping_address
        if customer:
            payload["customer"] = customer
        
        return self._request("POST", "/checkout_sessions", json=payload)
    
    def get_checkout(self, session_id: str) -> Dict[str, Any]:
        """Get checkout session."""
        return self._request("GET", f"/checkout_sessions/{session_id}")
    
    def update_checkout(
        self,
        session_id: str,
        items: List[Dict[str, Any]] = None,
        shipping_address: Dict[str, Any] = None,
        customer: Dict[str, Any] = None
    ) -> Dict[str, Any]:
        """Update checkout session."""
        payload = {}
        if items is not None:
            payload["items"] = items
        if shipping_address:
            payload["shipping_address"] = shipping_address
        if customer:
            payload["customer"] = customer
        
        return self._request("PATCH", f"/checkout_sessions/{session_id}", json=payload)
    
    def complete_checkout(self, session_id: str) -> Dict[str, Any]:
        """Complete checkout."""
        return self._request("POST", f"/checkout_sessions/{session_id}/complete")
    
    def cancel_checkout(self, session_id: str) -> Dict[str, Any]:
        """Cancel checkout."""
        return self._request("POST", f"/checkout_sessions/{session_id}/cancel")
    
    # Webhooks
    def emit_webhook(
        self,
        event_type: str,
        checkout_session_id: str = None,
        target_url: str = None,
        payload: Dict[str, Any] = None
    ) -> Dict[str, Any]:
        """Emit a webhook event."""
        body = {"event_type": event_type}
        if checkout_session_id:
            body["checkout_session_id"] = checkout_session_id
        if target_url:
            body["target_url"] = target_url
        if payload:
            body["payload"] = payload
        
        return self._request("POST", "/webhooks/emit", json=body)
 
 
# Create default client
client = LornClient()

# Basic search
results = client.search_products(q="comfortable running shoes")
for product in results["items"]:
    print(f"{product['title']} - ${product['price']}")
    if product.get("similarity_score"):
        print(f"  Relevance: {product['similarity_score']:.2f}")
 
# With filters
results = client.search_products(
    q="wireless headphones",
    min_price=50,
    max_price=200,
    page_size=5
)
 
# Category browse
results = client.search_products(category="Electronics", page_size=20)

product = client.get_product("prod_abc123")
 
print(f"Title: {product['title']}")
print(f"Price: ${product['price']}")
print(f"Description: {product['description']}")
 
# Show variants
for variant in product["variants"]:
    attrs = ", ".join(f"{k}: {v}" for k, v in variant["attributes"].items())
    print(f"  - {variant['sku']}: {attrs} - ${variant['price']}")

# Simple checkout
session = client.create_checkout(
    items=[
        {"product_id": "prod_abc123", "quantity": 1}
    ]
)
print(f"Session ID: {session['checkout_session']['id']}")
print(f"Total: ${session['checkout_session']['amounts']['total']}")
 
# With variant and full details
session = client.create_checkout(
    items=[
        {
            "product_id": "prod_abc123",
            "variant_sku": "sku_black_m",
            "quantity": 2
        }
    ],
    shipping_address={
        "name": "Jane Doe",
        "line1": "123 Main Street",
        "city": "San Francisco",
        "state": "CA",
        "postal_code": "94102",
        "country": "US"
    },
    customer={
        "email": "jane@example.com",
        "name": "Jane Doe"
    }
)

# Update shipping address
session = client.update_checkout(
    session_id="cs_demo_abc123",
    shipping_address={
        "name": "Jane Doe",
        "line1": "456 Oak Avenue",
        "city": "Los Angeles",
        "state": "CA",
        "postal_code": "90001",
        "country": "US"
    }
)
 
# Update cart
session = client.update_checkout(
    session_id="cs_demo_abc123",
    items=[
        {"product_id": "prod_abc123", "quantity": 3}
    ]
)

order = client.complete_checkout("cs_demo_abc123")
print(f"Status: {order['checkout_session']['status']}")
print(f"Total: ${order['checkout_session']['amounts']['total']}")

try:
    result = client.cancel_checkout("cs_demo_abc123")
    print(f"Canceled: {result['checkout_session']['status']}")
except requests.exceptions.HTTPError as e:
    if e.response.status_code == 405:
        print("Cannot cancel: already completed or canceled")
    else:
        raise

def complete_purchase_flow():
    """Complete shopping flow example."""
    
    # 1. Search for products
    print("Searching for running shoes...")
    results = client.search_products(q="running shoes", max_price=150)
    
    if not results["items"]:
        print("No products found")
        return
    
    product = results["items"][0]
    print(f"Found: {product['title']} - ${product['price']}")
    
    # 2. Get product details
    product_details = client.get_product(product["id"])
    variant = product_details["variants"][0] if product_details["variants"] else None
    
    # 3. Create checkout
    print("\nCreating checkout...")
    items = [{"product_id": product["id"], "quantity": 1}]
    if variant:
        items[0]["variant_sku"] = variant["sku"]
    
    session = client.create_checkout(
        items=items,
        shipping_address={
            "name": "Jane Doe",
            "line1": "123 Main St",
            "city": "San Francisco",
            "state": "CA",
            "postal_code": "94102",
            "country": "US"
        },
        customer={"email": "jane@example.com"}
    )
    
    session_id = session["checkout_session"]["id"]
    amounts = session["checkout_session"]["amounts"]
    
    print(f"Session: {session_id}")
    print(f"Subtotal: ${amounts['subtotal']}")
    print(f"Tax: ${amounts['tax']}")
    print(f"Shipping: ${amounts['shipping']}")
    print(f"Total: ${amounts['total']}")
    
    # 4. Complete checkout
    print("\nCompleting purchase...")
    order = client.complete_checkout(session_id)
    
    print(f"Order completed!")
    print(f"Status: {order['checkout_session']['status']}")
    
    return order
 
 
if __name__ == "__main__":
    complete_purchase_flow()

import requests
 
def safe_api_call():
    """Example with error handling."""
    try:
        results = client.search_products(q="shoes")
        return results
    except requests.exceptions.HTTPError as e:
        if e.response.status_code == 401:
            print("Invalid API key")
        elif e.response.status_code == 404:
            print("Not found")
        elif e.response.status_code == 429:
            print("Rate limited - retry later")
        elif e.response.status_code >= 500:
            print("Server error - retry later")
        else:
            print(f"HTTP error: {e.response.status_code}")
        return None
    except requests.exceptions.ConnectionError:
        print("Connection error")
        return None
    except requests.exceptions.Timeout:
        print("Request timed out")
        return None

import hmac
import hashlib
import time
 
def verify_webhook_signature(
    body: bytes,
    signature_header: str,
    secret: str,
    tolerance_seconds: int = 300
) -> bool:
    """Verify webhook signature."""
    # Parse signature header: "t=<timestamp>,v1=<signature>"
    parts = dict(p.split('=', 1) for p in signature_header.split(','))
    timestamp = int(parts.get('t', 0))
    received_sig = parts.get('v1', '')
    
    # Check timestamp freshness
    if abs(time.time() - timestamp) > tolerance_seconds:
        return False
    
    # Compute expected signature
    expected_sig = hmac.new(
        secret.encode('utf-8'),
        body,
        hashlib.sha256
    ).hexdigest()
    
    # Timing-safe comparison
    return hmac.compare_digest(expected_sig, received_sig)
 
 
# Flask webhook handler example
from flask import Flask, request, abort
 
app = Flask(__name__)
WEBHOOK_SECRET = os.getenv("LORN_WEBHOOK_SECRET")
 
@app.route("/webhooks/lorn", methods=["POST"])
def handle_webhook():
    signature = request.headers.get("X-Webhook-Signature", "")
    
    if not verify_webhook_signature(request.data, signature, WEBHOOK_SECRET):
        abort(401)
    
    event = request.json
    event_type = event.get("type")
    
    if event_type == "order.created":
        handle_order_created(event["data"]["object"])
    elif event_type == "order.fulfilled":
        handle_order_fulfilled(event["data"]["object"])
    
    return {"received": True}

import aiohttp
import asyncio
 
class AsyncLornClient:
    """Async Lorn AI client."""
    
    def __init__(self, base_url: str, api_key: str):
        self.base_url = base_url
        self.headers = {
            "X-ACP-API-Key": api_key,
            "Content-Type": "application/json"
        }
    
    async def search_products(self, q: str = None, **params) -> dict:
        """Search products asynchronously."""
        if q:
            params["q"] = q
        
        async with aiohttp.ClientSession() as session:
            async with session.get(
                f"{self.base_url}/acp/products",
                headers=self.headers,
                params=params
            ) as response:
                response.raise_for_status()
                return await response.json()
 
 
async def main():
    client = AsyncLornClient(LORN_BASE_URL, LORN_API_KEY)
    results = await client.search_products(q="running shoes")
    print(results)
 
 
if __name__ == "__main__":
    asyncio.run(main())