Python SDK

pip install lobstrio-sdk

export LOBSTR_TOKEN=your_api_key

import time
from lobstrio import LobstrClient

# Token resolved from LOBSTR_TOKEN env var, then ~/.config/lobstr/config.toml.
# Set LOBSTR_TOKEN explicitly in production — misnamed env vars fail silently.
client = LobstrClient()

# 1. Resolve crawler slug → id (create() requires the hash, not the slug)
crawlers = client.crawlers.list()
crawler = next(c for c in crawlers if c.slug == "google-maps-leads-scraper")

# 2. Create a squid
squid = client.squids.create(crawler=crawler.id, name="Restaurants Paris")

# 3. Configure squid-level params — required before runs.start(), even when all
#    params are optional. Pass None for params you don't need to set.
client.squids.update(squid.id, params={"language": "English (United States)", "max_results": 50})

# 4. Add tasks
client.tasks.add(squid=squid.id, tasks=[
    {"url": "https://www.google.com/maps/search/restaurants/@48.8566,2.3522,14z"}
])

# 5. Start run and stream progress
def on_progress(stats):
    print(f"  {stats.percent_done} — {stats.total_results} results, ETA {stats.eta}")

run = client.runs.start(squid=squid.id)
print(f"Run {run.id} started")
run = client.runs.wait(run.id, callback=on_progress)
# Shortcut: client.runs.call(squid=squid.id, callback=on_progress) does the
# start + wait in one call. Add timeout=<seconds> to either to bound the wait.
print(f"Run complete — {run.total_results} results, {run.credit_used} credits used")

# 6. Wait for export to be ready before fetching results
while not run.export_done:
    time.sleep(3)
    run = client.runs.get(run.id)

# 7. Iterate results (PageIterator yields one dict per result)
for place in client.results.iter(squid=squid.id):
    print(f"{place['name']} — {place.get('score', 'N/A')}★  {place.get('phone', '')}")

# 8. Clean up — delete the squid to free the slot
client.squids.delete(squid.id)