Going to Production

This guide covers the requirements and best practices for launching your Lorn AI integration in production.

Pre-Launch Checklist

API Integration

Using production API keys (not sandbox)
All endpoints tested and working
Error handling implemented
Retry logic with exponential backoff
Idempotency keys for all mutating operations

Security

API keys stored securely (not in code)
HTTPS enforced for all requests
Webhook signatures verified
Request signing implemented (if required)
Input validation on all user data

Reliability

Timeouts configured for API calls
Graceful degradation when API unavailable
Logging for debugging and auditing
Monitoring and alerting set up

Compliance

Privacy policy updated
Terms of service reviewed
Data handling compliant with GDPR/CCPA
PCI compliance for payment data (if applicable)

Security Requirements

API Key Management

Do:

import os
 
# Load from environment
API_KEY = os.getenv("LORN_API_KEY")
if not API_KEY:
    raise ValueError("LORN_API_KEY environment variable required")

Don’t:

# Never hardcode keys!
API_KEY = "sk_live_abc123..."  # ❌ WRONG

Secure Storage Options

Method	Use Case
Environment variables	Simple deployments
Secret manager (AWS, GCP, Azure)	Cloud deployments
HashiCorp Vault	Enterprise environments
Kubernetes secrets	Container orchestration

Key Rotation

Implement key rotation without downtime:

import os
 
# Support multiple keys during rotation
PRIMARY_KEY = os.getenv("LORN_API_KEY")
FALLBACK_KEY = os.getenv("LORN_API_KEY_FALLBACK")
 
def get_api_key():
    return PRIMARY_KEY or FALLBACK_KEY

HTTPS Requirements

Enforce TLS

All production traffic must use HTTPS:

BASE_URL = "https://{{YOUR_STORE_URL}}"  # ✅ Always HTTPS
 
# Verify SSL certificates
response = requests.get(url, verify=True)  # Default, but be explicit

Certificate Validation

Never disable certificate verification:

# ❌ NEVER do this in production
response = requests.get(url, verify=False)

Webhook Security

Signature Verification

Always verify webhook signatures:

import hmac
import hashlib
import time
 
def verify_webhook(body: bytes, signature_header: str, secret: str) -> bool:
    """Verify webhook signature with timestamp validation."""
    # Parse signature
    parts = dict(p.split('=', 1) for p in signature_header.split(','))
    timestamp = int(parts.get('t', 0))
    signature = parts.get('v1', '')
    
    # Reject old webhooks (replay attack protection)
    if abs(time.time() - timestamp) > 300:  # 5 minute window
        return False
    
    # Verify signature
    expected = hmac.new(
        secret.encode(),
        body,
        hashlib.sha256
    ).hexdigest()
    
    return hmac.compare_digest(expected, signature)

Webhook Endpoint Security

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():
    # Verify signature
    signature = request.headers.get("X-Webhook-Signature")
    if not verify_webhook(request.data, signature, WEBHOOK_SECRET):
        abort(401)
    
    # Process webhook
    event = request.json
    process_event(event)
    
    return {"received": True}

Request Signing

For enhanced security, sign all requests:

import hmac
import hashlib
import base64
from datetime import datetime, timezone
 
def sign_request(body: bytes, secret: str) -> tuple[str, str]:
    """Generate request signature and timestamp."""
    timestamp = datetime.now(timezone.utc).isoformat()
    
    # Create signature
    signature = hmac.new(
        secret.encode(),
        body,
        hashlib.sha256
    ).digest()
    
    return base64.b64encode(signature).decode(), timestamp
 
def make_signed_request(method, url, body=None):
    headers = {"X-ACP-API-Key": API_KEY}
    
    if body:
        body_bytes = json.dumps(body).encode()
        signature, timestamp = sign_request(body_bytes, SIGNING_SECRET)
        headers["Signature"] = signature
        headers["Timestamp"] = timestamp
        headers["Content-Type"] = "application/json"
    
    return requests.request(method, url, headers=headers, data=body_bytes)

Error Handling

Implement Retries

import time
from functools import wraps
 
def with_retry(max_attempts=3, backoff_factor=2):
    """Retry decorator with exponential backoff."""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            last_exception = None
            
            for attempt in range(max_attempts):
                try:
                    return func(*args, **kwargs)
                except requests.exceptions.RequestException as e:
                    last_exception = e
                    
                    # Don't retry client errors (4xx)
                    if hasattr(e, 'response') and 400 <= e.response.status_code < 500:
                        raise
                    
                    # Exponential backoff
                    if attempt < max_attempts - 1:
                        sleep_time = backoff_factor ** attempt
                        time.sleep(sleep_time)
            
            raise last_exception
        return wrapper
    return decorator
 
@with_retry(max_attempts=3)
def search_products(query: str):
    return requests.get(f"{BASE_URL}/acp/products", params={"q": query})

Handle Rate Limits

def handle_rate_limit(response):
    """Handle 429 responses with Retry-After header."""
    if response.status_code == 429:
        retry_after = int(response.headers.get("Retry-After", 60))
        time.sleep(retry_after)
        return True
    return False

Graceful Degradation

def search_with_fallback(query: str):
    """Search with fallback to cached results."""
    try:
        return lorn_search(query)
    except requests.exceptions.RequestException:
        # Fall back to cached popular products
        return get_cached_products()

Idempotency

Use Idempotency Keys

Prevent duplicate operations:

import uuid
 
def create_checkout(items: list, idempotency_key: str = None) -> dict:
    """Create checkout with idempotency."""
    if not idempotency_key:
        idempotency_key = str(uuid.uuid4())
    
    response = requests.post(
        f"{BASE_URL}/checkout_sessions",
        headers={
            "X-ACP-API-Key": API_KEY,
            "Idempotency-Key": idempotency_key,
            "Content-Type": "application/json"
        },
        json={"items": items}
    )
    
    return response.json()

Generate Deterministic Keys

For retries, use deterministic keys:

def generate_checkout_key(user_id: str, cart_hash: str) -> str:
    """Generate deterministic idempotency key."""
    return f"checkout_{user_id}_{cart_hash}_{date.today().isoformat()}"

Monitoring & Logging

Structured Logging

import logging
import json
 
class JSONFormatter(logging.Formatter):
    def format(self, record):
        log_data = {
            "timestamp": self.formatTime(record),
            "level": record.levelname,
            "message": record.getMessage(),
            "logger": record.name
        }
        if hasattr(record, 'extra'):
            log_data.update(record.extra)
        return json.dumps(log_data)
 
# Usage
logger = logging.getLogger("lorn")
logger.info("API call", extra={
    "endpoint": "/acp/products",
    "method": "GET",
    "duration_ms": 150,
    "status_code": 200
})

Key Metrics to Track

Metric	Description
API latency	Response time for each endpoint
Error rate	Percentage of failed requests
Checkout conversion	Ratio of started to completed checkouts
Search queries/min	Volume of search traffic

Alerting

Set up alerts for:

Error rate > 5%
Latency p95 > 2 seconds
Any 503 (service unavailable) responses
Webhook delivery failures

Rate Limit Compliance

Monitor Usage

def track_rate_limits(response):
    """Track rate limit headers."""
    limit = response.headers.get("X-RateLimit-Limit")
    remaining = response.headers.get("X-RateLimit-Remaining")
    reset = response.headers.get("X-RateLimit-Reset")
    
    metrics.gauge("lorn.ratelimit.remaining", int(remaining))
    
    if int(remaining) < 100:
        logger.warning(f"Rate limit low: {remaining} remaining")

Implement Throttling

from ratelimit import limits, sleep_and_retry
 
# 600 requests per minute
@sleep_and_retry
@limits(calls=600, period=60)
def lorn_api_call(endpoint: str):
    return requests.get(f"{BASE_URL}{endpoint}")

Data Handling

PII Protection

def sanitize_for_logging(data: dict) -> dict:
    """Remove PII before logging."""
    sensitive_fields = ["email", "phone", "name", "line1", "postal_code"]
    sanitized = data.copy()
    
    for field in sensitive_fields:
        if field in sanitized:
            sanitized[field] = "***REDACTED***"
    
    return sanitized
 
logger.info("Checkout created", extra={
    "session": sanitize_for_logging(session)
})

Data Retention

Don’t store raw API responses longer than needed
Implement automatic cleanup of old session data
Follow GDPR right-to-erasure requirements

Health Checks

Implement Liveness Check

@app.route("/health/live")
def liveness():
    """Basic liveness check."""
    return {"status": "ok"}
 
@app.route("/health/ready")
def readiness():
    """Readiness check including Lorn AI connectivity."""
    try:
        response = requests.get(
            f"{LORN_BASE_URL}/health",
            timeout=5
        )
        lorn_status = response.json().get("status") == "ok"
    except:
        lorn_status = False
    
    if not lorn_status:
        return {"status": "degraded", "lorn": False}, 503
    
    return {"status": "ok", "lorn": True}

Deployment Checklist

Before Deploy

Environment Variables

LORN_API_KEY=sk_live_...
LORN_WEBHOOK_SECRET=whsec_...
LORN_BASE_URL=https://your-store.lorn.ai

Test Production Endpoints

curl "https://your-store.lorn.ai/health"

Verify Webhook Endpoint

curl -X POST "https://your-app.com/webhooks/lorn" \
  -H "X-Webhook-Signature: test" \
  -d '{}'
# Should return 401 (signature invalid)

After Deploy

Monitor Error Rates — Watch for increased failures
Check Latency — Verify response times are acceptable
Test Full Flow — Complete a test purchase
Verify Webhooks — Confirm webhook delivery

Support

Getting Help

Issue Type	Contact
API errors	mayank@lornai.com.com
Integration questions	mayank@lornai.com
Security concerns	mayank@lornai.com
Enterprise support	mayank@lornai.com

Reporting Issues

Include:

Request ID (from response headers)
Timestamp of issue
Endpoint and parameters
Error message/response