Integration guide

LeadMerlin + Python

A single script using the requests library: resolve a city, start a search, poll, page through results and write a CSV.

Before you start

• Python 3.9+ with requests installed (pip install requests).
• LEADMERLIN_API_KEY set in your environment.

Setup

1. 1

   Run the script

   Save as leads.py and run python leads.py.

   import os, sys, time, csv, requests
   
   API = "https://api.leadmerlin.com/api"
   H = {"x-api-key": os.environ["LEADMERLIN_API_KEY"]}
   
   def call(method, path, **kw):
       r = requests.request(method, f"{API}{path}", headers=H, timeout=30, **kw)
       if r.status_code == 402:
           sys.exit(f"Out of lead credits: {r.json()}")
       r.raise_for_status()  # 401 bad key, 400 bad request, 5xx server error
       return r.json()
   
   # 1. Resolve a city to a location code
   loc = call("GET", "/v2/locations", params={"q": "austin"})
   if not loc["locations"]:
       sys.exit("No matching location")
   code = loc["locations"][0]["location_code"]
   
   # 2. Start a search job ONCE (don't retry a POST blindly: it starts a second paid job)
   job = call("POST", "/v2/searches", json={
       "keyword": "roofing contractor",
       "location_codes": [code],
       "strictness": "moderate",
       "max_leads": 50,  # hard budget for this run
   })
   job_id = job["job_id"]
   
   # 3. Poll until the job finishes; completed and failed are both final
   deadline = time.time() + 15 * 60
   while True:
       status = call("GET", f"/v2/jobs/{job_id}")
       if status["status"] == "failed":
           sys.exit(f"Job failed: {status['error']}")
       if status["status"] == "completed":
           break
       if time.time() > deadline:
           sys.exit(f"Still running after 15 min; check job {job_id} later")
       print(f"{status['status']} {status['progress']}%")
       time.sleep(5)
   
   # 4. Page through results (requests URL-encodes the cursor)
   businesses, cursor = [], None
   while True:
       params = {"limit": 200, **({"cursor": cursor} if cursor else {})}
       page = call("GET", f"/v2/jobs/{job_id}/results", params=params)
       businesses += page["businesses"]
       if not page["has_more"]:
           break
       cursor = page["next_cursor"]
   
   # 5. Write a simple CSV (guard against spreadsheet formulas in scraped text)
   def cell(v):
       v = "" if v is None else str(v)
       return "'" + v if v[:1] in ("=", "+", "-", "@") and not v.replace("+", "").replace(" ", "").isdigit() else v
   
   with open("leads.csv", "w", newline="") as f:
       w = csv.writer(f)
       w.writerow(["name", "owner", "email", "phone", "phone_type", "website", "city"])
       for b in businesses:
           w.writerow([cell(x) for x in (b["name"], b["owner"], b["email"], b["phone"],
                                         b["phone_type"], b["website"], b["contact"]["city"])])
   print(f"Saved {len(businesses)} leads ({status['over_budget_count']} more matched beyond the budget)")

Location codes in these examples are placeholders — always look yours up with GET /v2/locations first.

Tips

• Treat HTTP 402 (insufficient_credits) as a stop, not a retry. Retry 5xx on GET requests only; never blindly re-send a POST, or you start a second paid job.
• For domains you already have, POST them to /v2/enrichments (up to 100 per request) and reuse the same poll + results loop.

API at a glance

All calls use the x-api-key header except the demo endpoints. Base URL: https://api.leadmerlin.com/api

Related