Skip to main content

Complete Automation Workflow

This example shows a complete end-to-end automation:
Search prospects → Save to CRM → Create list → Launch campaign

Python Client Library

First, create a reusable client:
leadlex_client.py
import requests
from typing import List, Dict, Optional

class LeadLexClient:
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.base_url = "https://nbkxaqxwvkgbddekwsma.supabase.co/functions/v1/api-gateway"
        self.session = requests.Session()
        self.session.headers.update({"Authorization": f"Bearer {api_key}"})
    
    def _request(self, method: str, path: str, **kwargs) -> Dict:
        url = f"{self.base_url}{path}"
        response = self.session.request(method, url, **kwargs)
        
        # Check rate limits
        if 'X-RateLimit-Remaining' in response.headers:
            remaining = int(response.headers['X-RateLimit-Remaining'])
            if remaining < 10:
                print(f"⚠️  Only {remaining} API calls remaining today")
        
        response.raise_for_status()
        return response.json()
    
    def search_prospects(self, query: Dict, per_page: int = 25) -> List[Dict]:
        result = self._request("POST", "/v1/prospects/search", 
                             json={"query": query, "per_page": per_page})
        return result["data"]["prospects"]
    
    def save_prospects(self, prospects: List[Dict], list_id: str = None) -> Dict:
        result = self._request("POST", "/v1/prospects/save", 
                             json={"prospects": prospects, "list_id": list_id})
        return result["data"]
    
    def create_list(self, name: str, description: str = "") -> Dict:
        result = self._request("POST", "/v1/lists", 
                             json={"name": name, "description": description})
        return result["data"]
    
    def create_campaign(self, name: str, list_id: str, template: str) -> Dict:
        result = self._request("POST", "/v1/campaigns",
                             json={"name": name, "list_id": list_id, "template": template})
        return result["data"]
    
    def chat_with_lexi(self, message: str, context: Dict = None) -> Dict:
        result = self._request("POST", "/v1/lexi/chat",
                             json={"message": message, "context": context or {}})
        return result["data"]

Complete Workflow

automation.py
from leadlex_client import LeadLexClient
import os

client = LeadLexClient(api_key=os.getenv("LEADLEX_API_KEY"))

# Step 1: Search for prospects
print("🔍 Searching for CEOs at law firms...")
prospects = client.search_prospects(query={
    "person_titles": ["CEO", "Chief Executive Officer"],
    "organization_industries": ["Legal"],
    "person_locations": ["United States"],
    "organization_num_employees_ranges": ["11-50", "51-200"]
}, per_page=50)

print(f"✅ Found {len(prospects)} prospects\n")

# Step 2: Create a list
print("📋 Creating list...")
new_list = client.create_list(
    name="Law Firm CEOs Q1 2026",
    description="Target prospects for legal tech campaign"
)
list_id = new_list["id"]
print(f"✅ List created: {list_id}\n")

# Step 3: Save prospects to CRM
print("💾 Saving prospects to CRM...")
saved = client.save_prospects(prospects, list_id=list_id)
print(f"✅ Saved {saved['created']} contacts\n")

# Step 4: Ask Lexi to draft campaign
print("🤖 Asking Lexi to draft campaign...")
lexi_response = client.chat_with_lexi(
    message=f"Create an outreach campaign for list {list_id}. "
            "Focus on legal tech automation benefits.",
    context={"list_id": list_id}
)
print(f"✅ Lexi: {lexi_response['response']}")

Common Patterns

Error Handling

import requests

try:
    response = requests.get(
        f"{BASE_URL}/v1/contacts",
        headers={"Authorization": f"Bearer {API_KEY}"}
    )
    response.raise_for_status()
    data = response.json()
    
except requests.exceptions.HTTPError as e:
    error = e.response.json()
    if error["error"]["code"] == "rate_limited":
        print("Rate limit exceeded! Try again later.")
    elif error["error"]["code"] == "invalid_key":
        print("Invalid API key")
    else:
        print(f"Error: {error['error']['message']}")

Pagination

def get_all_contacts(client):
    """Fetch all contacts across multiple pages"""
    all_contacts = []
    page = 1
    
    while True:
        response = client._request("GET", "/v1/contacts", params={
            "page": page,
            "per_page": 100
        })
        
        contacts = response["data"]["contacts"]
        all_contacts.extend(contacts)
        
        # Check if we've reached the end
        total = response["meta"]["total"]
        if len(all_contacts) >= total:
            break
        
        page += 1
    
    return all_contacts

Rate Limit Monitoring

def monitor_rate_limits(response: requests.Response):
    """Extract and log rate limit headers"""
    limit = response.headers.get('X-RateLimit-Limit')
    remaining = response.headers.get('X-RateLimit-Remaining')
    reset = response.headers.get('X-RateLimit-Reset')
    
    if remaining:
        percent = (int(remaining) / int(limit)) * 100
        print(f"Rate limit: {remaining}/{limit} ({percent:.1f}% remaining)")
        
        if int(remaining) < 10:
            print("⚠️  Warning: Running low on API calls!")

Batch Operations

Bulk Contact Creation

def bulk_create_contacts(client, contacts_data):
    """Create multiple contacts efficiently"""
    created = []
    
    for contact in contacts_data:
        try:
            result = client.create_contact(**contact)
            created.append(result)
            print(f"✓ Created: {contact['full_name']}")
        except Exception as e:
            print(f"✗ Failed: {contact['full_name']} - {e}")
    
    return created

# Usage
contacts = [
    {"full_name": "Alice Smith", "email": "alice@example.com"},
    {"full_name": "Bob Jones", "email": "bob@example.com"},
    {"full_name": "Carol White", "email": "carol@example.com"}
]

results = bulk_create_contacts(client, contacts)
print(f"\nCreated {len(results)}/{len(contacts)} contacts")

Next Steps