Automating Lead Searches at Scale

Automate LinkedIn Sales Navigator lead searches via API—no LinkedIn account, cookies, or browser extensions needed.

What You'll Build

A Python script that:

1. Initiates a lead search with your criteria
2. Polls for completion
3. Retrieves all matching leads

Example output:

Search initiated: search-leads-1765185558588-4pu1297rl
Status: processing (45/100 results)
Status: done (100/100 results)
Found 100 leads

Search Criteria Example

We'll search for leads matching:

• Job titles: Owner, Founder, Director
• Location: United States
• Current companies: Google, Microsoft

For all available filters, see Lead Search Filters.

Step-by-Step Implementation

Step 1: Initiate the Search

import requests

API_KEY = "YOUR_API_KEY"
BASE_URL = "https://api.infoplug.io"

def initiate_search():
    response = requests.post(
        f"{BASE_URL}/search-leads",
        headers={
            "X-INFOPLUG-API-KEY": API_KEY,
            "Content-Type": "application/json"
        },
        json={
            "geo_codes": [103644278],  # United States
            "title_keywords": ["Owner", "Founder", "Director"],
            "current_company_names": ["Google", "Microsoft"],
            "limit": 100
        }
    )
    response.raise_for_status()
    return response.json().get("request_id")

Response:

{
  "message": "Lead search job created",
  "request_id": "search-leads-1765185558588-4pu1297rl",
  "status": "pending",
  "search_type": "filters",
  "limit": 100
}

Step 2: Poll for Completion

import time

def wait_for_completion(request_id):
    while True:
        response = requests.get(
            f"{BASE_URL}/get-lead-search-status",
            headers={"X-INFOPLUG-API-KEY": API_KEY},
            params={"request_id": request_id}
        )
        response.raise_for_status()
        data = response.json()
        status = data.get("status")
        results_found = data.get("results_found", 0)
        limit = data.get("limit", 0)
        print(f"Status: {status} ({results_found}/{limit} results)")
        
        if status == "done":
            return True
        if status == "failed":
            return False
        
        time.sleep(30)  # Check every 30 seconds

Response:

{
  "status": "done",
  "message": "Found 100/100 results. Use /get-lead-results to retrieve data.",
  "total_count": 500,
  "results_found": 100,
  "limit": 100
}

Step 3: Fetch Results

def fetch_results(request_id):
    all_results = []
    page = 1
    
    while True:
        response = requests.get(
            f"{BASE_URL}/get-lead-results",
            headers={"X-INFOPLUG-API-KEY": API_KEY},
            params={"request_id": request_id, "page": page}
        )
        response.raise_for_status()
        data = response.json().get("data", [])
        
        if not data:
            break
        
        all_results.extend(data)
        page += 1
    
    return all_results

Full Script

Copy this complete script to get started:

import requests
import time

API_KEY = "YOUR_API_KEY"
BASE_URL = "https://api.infoplug.io"

def initiate_search():
    response = requests.post(
        f"{BASE_URL}/search-leads",
        headers={
            "X-INFOPLUG-API-KEY": API_KEY,
            "Content-Type": "application/json"
        },
        json={
            "geo_codes": [103644278],
            "title_keywords": ["Owner", "Founder", "Director"],
            "current_company_names": ["Google", "Microsoft"],
            "limit": 100
        }
    )
    response.raise_for_status()
    return response.json().get("request_id")

def wait_for_completion(request_id):
    while True:
        response = requests.get(
            f"{BASE_URL}/get-lead-search-status",
            headers={"X-INFOPLUG-API-KEY": API_KEY},
            params={"request_id": request_id}
        )
        response.raise_for_status()
        data = response.json()
        status = data.get("status")
        results_found = data.get("results_found", 0)
        limit = data.get("limit", 0)
        print(f"Status: {status} ({results_found}/{limit} results)")
        
        if status == "done":
            return True
        if status == "failed":
            return False
        
        time.sleep(30)

def fetch_results(request_id):
    all_results = []
    page = 1
    
    while True:
        response = requests.get(
            f"{BASE_URL}/get-lead-results",
            headers={"X-INFOPLUG-API-KEY": API_KEY},
            params={"request_id": request_id, "page": page}
        )
        response.raise_for_status()
        data = response.json().get("data", [])
        
        if not data:
            break
        
        all_results.extend(data)
        page += 1
    
    return all_results

if __name__ == "__main__":
    print("Initiating search...")
    request_id = initiate_search()
    print(f"Request ID: {request_id}")
    
    if wait_for_completion(request_id):
        results = fetch_results(request_id)
        print(f"Found {len(results)} leads")
        for lead in results[:3]:  # Preview first 3
            print(f"  - {lead.get('full_name')} @ {lead.get('company')}")
    else:
        print("Search failed")

Pricing

• Initiation: 50 credits (charged even if no results)
• Status checks: Free
• Results: 0.5 credits per lead

Example: 100 leads = 50 + (0.5 × 100) = 100 credits

Rate Limits

Status checks are free and unlimited. Space them 30 seconds apart to avoid unnecessary polling.