lobstrio-sdk
Python SDK for the Lobstr.io API — web scraping automation platform
- Sync + async clients with the same API surface
- Typed dataclass models for all responses
- Lazy auto-pagination
- Automatic token resolution from CLI config or environment
pip install lobstrio-sdkRequires Python 3.10+. The only runtime dependency is httpx.
The client resolves your API token in this order:
- Explicit —
LobstrClient(token="your-token") - Environment variable —
LOBSTR_TOKEN - CLI config file —
~/.config/lobstr/config.toml(same file used bylobstrCLI)
If you already have the CLI set up, the SDK works with no configuration:
from lobstrio import LobstrClient
client = LobstrClient() # token auto-resolved
user = client.me()
print(user.email)from lobstrio import LobstrClient
with LobstrClient() as client:
# Account info
user = client.me()
balance = client.balance()
print(f"{user.email} — {balance.credits} credits")
# List crawlers
for crawler in client.crawlers.list():
print(f"{crawler.name} ({crawler.id})")
# Create a squid, add tasks, run it
squid = client.squids.create("google-maps-scraper", name="My Scrape")
client.squids.update(squid.id, params={"language": "English (United States)"})
client.tasks.add(squid=squid.id, tasks=[{"url": "https://maps.google.com/..."}])
run = client.runs.start(squid=squid.id)
# Wait for completion with progress callback
# (or use client.runs.call(squid=squid.id) to start + wait in one call)
final = client.runs.wait(run.id, callback=lambda s: print(f"{s.percent_done}%"))
# Download results
client.runs.download(run.id, "results.csv")All API operations are organized under resource namespaces on the client.
User
user = client.me() # User profile
balance = client.balance() # Account balance (credits, subscription)Crawlers — browse scraper templates
crawlers = client.crawlers.list() # All crawlers
crawler = client.crawlers.get("crawler-id") # Single crawler
params = client.crawlers.params("crawler-id") # Parameter schema
attrs = client.crawlers.attributes("crawler-id") # Result columnsModels: Crawler, CrawlerAttribute, CrawlerParams
Squids — manage scraper instances
# List & iterate
squids = client.squids.list(limit=50, page=1)
for squid in client.squids.iter(): # auto-paginate all squids
print(squid.name)
# CRUD
squid = client.squids.create("crawler-id", name="My Project")
squid = client.squids.get("squid-id")
squid = client.squids.update("squid-id", name="Renamed", concurrency=2,
params={"language": "English"})
client.squids.empty("squid-id") # remove all tasks
client.squids.delete("squid-id")Model: Squid (id, name, crawler, is_active, concurrency, params, created_at, ...)
Tasks — manage input URLs and keywords
# List & iterate
tasks = client.tasks.list(squid="squid-id")
for task in client.tasks.iter(squid="squid-id"):
print(task.id)
# Add tasks
result = client.tasks.add(
squid="squid-id",
tasks=[
{"url": "https://maps.google.com/maps?cid=123"},
{"url": "https://maps.google.com/maps?cid=456"},
],
)
print(f"Added {len(result.tasks)}, {result.duplicated_count} duplicates")
# Upload from CSV/TSV
resp = client.tasks.upload(squid="squid-id", file="tasks.csv")
status = client.tasks.upload_status(resp["id"])
# Get & delete
task = client.tasks.get("task-hash")
client.tasks.delete("task-hash")Models: Task, TaskStatus, AddTasksResult, UploadStatus, UploadMeta
Runs — start, monitor, and download
# Start a run
run = client.runs.start(squid="squid-id")
# List runs
runs = client.runs.list(squid="squid-id")
for run in client.runs.iter(squid="squid-id"):
print(run.id, run.status)
# Monitor
run = client.runs.get("run-id")
stats = client.runs.stats("run-id")
print(f"{stats.percent_done}% done, {stats.total_results} results")
# Wait for completion (blocking, with optional progress callback)
final = client.runs.wait("run-id", poll_interval=5.0,
callback=lambda s: print(f"{s.percent_done}%"))
# Start and wait in one call (mirrors start() + wait())
final = client.runs.call(squid="squid-id", poll_interval=5.0,
callback=lambda s: print(f"{s.percent_done}%"))
# Bound the wait: RunTimeout (a TimeoutError subclass) if it overruns.
# The run keeps going server-side — re-attach later with wait()/get().
from lobstrio import RunTimeout
try:
final = client.runs.wait("run-id", timeout=600) # also on call()
except RunTimeout:
...
# Download results
url = client.runs.download_url("run-id") # signed S3 URL
client.runs.download("run-id", "output.csv") # download to file
# Abort
client.runs.abort("run-id")
# Tasks within a run
tasks = client.runs.tasks("run-id")Models: Run, RunStats
Results — fetch scraped data
results = client.results.list(squid="squid-id", page_size=100)
# Auto-paginate all results
for row in client.results.iter(squid="squid-id"):
print(row) # dictResults are returned as plain dict objects (the schema depends on the crawler).
Accounts — manage connected platform accounts
accounts = client.accounts.list()
account = client.accounts.get("account-id")
types = client.accounts.types() # available account types
# Sync account with cookies
resp = client.accounts.sync(type="google", cookies={"SID": "...", "HSID": "..."})
status = client.accounts.sync_status(resp["id"])
# Update limits
client.accounts.update("account-id", type="google", params={"daily_limit": 100})
# Delete
client.accounts.delete("account-id")Models: Account, AccountType, SyncStatus
Delivery — configure result delivery
# Email
client.delivery.email("squid-id", email="you@example.com")
client.delivery.test_email(email="you@example.com")
# Google Sheets
client.delivery.google_sheet("squid-id", url="https://docs.google.com/spreadsheets/d/...", append=True)
client.delivery.test_google_sheet(url="https://docs.google.com/spreadsheets/d/...")
# Webhook
client.delivery.webhook("squid-id", url="https://your-server.com/hook",
on_done=True, on_error=True)
client.delivery.test_webhook(url="https://your-server.com/hook")
# S3
client.delivery.s3("squid-id", bucket="my-bucket", target_path="scrapes/",
aws_access_key="...", aws_secret_key="...")
client.delivery.test_s3(bucket="my-bucket")
# SFTP
client.delivery.sftp("squid-id", host="ftp.example.com", username="user",
password="pass", directory="/uploads")
client.delivery.test_sftp(host="ftp.example.com", username="user",
password="pass", directory="/uploads")Models: EmailDelivery, GoogleSheetDelivery, S3Delivery, WebhookDelivery, SFTPDelivery
The async client mirrors the sync API exactly, using async/await:
from lobstrio import AsyncLobstrClient
async def main():
async with AsyncLobstrClient() as client:
user = await client.me()
print(user.email)
crawlers = await client.crawlers.list()
for c in crawlers:
print(c.name)
squid = await client.squids.create("crawler-id", name="Async Scrape")
await client.tasks.add(squid=squid.id, tasks=[{"url": "..."}])
run = await client.runs.start(squid=squid.id)
final = await client.runs.wait(run.id)
await client.runs.download(run.id, "results.csv")All resource methods (client.crawlers.*, client.squids.*, etc.) work identically — just add await.
Resources that return lists support two patterns:
Single page (.list()) — returns one page of results:
page1 = client.squids.list(limit=10, page=1)
page2 = client.squids.list(limit=10, page=2)Auto-pagination (.iter()) — lazy iterator that fetches pages on demand:
for squid in client.squids.iter(limit=50):
print(squid.name) # automatically fetches next pagesThe async client provides AsyncPageIterator for use with async for.
All API errors raise typed exceptions with status_code, message, and body:
from lobstrio import LobstrClient, AuthError, NotFoundError, RateLimitError, APIError
try:
client.squids.get("nonexistent")
except NotFoundError as e:
print(f"Not found: {e.message}")
except AuthError:
print("Invalid or expired token")
except RateLimitError as e:
print(f"Rate limited, retry after {e.retry_after}s")
except APIError as e:
print(f"API error [{e.status_code}]: {e.message}")| Exception | HTTP Status | When |
|---|---|---|
AuthError |
401 | Invalid or missing token |
NotFoundError |
404 | Resource doesn't exist |
RateLimitError |
429 | Too many requests (has retry_after) |
APIError |
4xx/5xx | All other API errors |
CLI (pip install lobstrio) |
SDK (pip install lobstrio-sdk) |
|
|---|---|---|
| Use case | Terminal workflows, quick scrapes, cron jobs | Scripts, pipelines, applications |
| Interface | Shell commands | Python API |
| Output | Rich tables, progress bars, CSV files | Typed dataclass models |
| Async | No | Yes (AsyncLobstrClient) |
| Pagination | Manual (--page, --limit) |
Auto (client.squids.iter()) |
For terminal workflows, see lobstrio — the companion CLI tool.
Where do I get an API token?
Go to Dashboard → API to find your token. It's always available there, pre-generated.
Do I need the CLI installed for the SDK to work?
No. The SDK is standalone. However, if you have the CLI configured (lobstr config set-token), the SDK will automatically pick up the token from ~/.config/lobstr/config.toml — no code changes needed.
How do I handle rate limiting?
Catch RateLimitError and use its retry_after attribute:
from lobstrio import RateLimitError
import time
try:
results = client.results.list(squid="squid-id")
except RateLimitError as e:
time.sleep(float(e.retry_after or 5))
results = client.results.list(squid="squid-id")Can I use the async client with Django/FastAPI?
Yes. Use AsyncLobstrClient in any async context:
from lobstrio import AsyncLobstrClient
async def scrape_view(request):
async with AsyncLobstrClient() as client:
results = await client.results.list(squid="squid-id")
return results# Clone and install
git clone https://github.com/lobstrio/lobstrio-sdk.git
cd lobstrio-sdk
pip install -e ".[dev]"
# Run unit tests
pytest
# Run live tests (requires API token)
pytest tests/test_live.py -v
# Lint & type check
ruff check src/ tests/
mypy src/lobstrio/Publishing to PyPI is triggered by a GitHub Release, not by merging to master.
Merging alone does not publish. To cut a release:
- Bump the version in
src/lobstrio/_version.py(e.g.0.4.1). - Add a matching entry to
CHANGELOG.md. - Merge to
master(via PR; CI must be green). - Create a GitHub Release tagged
vX.Y.Zonmaster:gh release create v0.4.1 --target master --title "v0.4.1" --notes "…"
Creating the release fires .github/workflows/publish.yml (on: release: published),
which builds with uv and publishes to PyPI using the PYPI_TOKEN repo secret.
Confirm it went out: pip index versions lobstrio-sdk (or check the PyPI page).
Note: a PyPI version cannot be overwritten — bump the version for every release.
Contributions are welcome! See CONTRIBUTING.md for development setup, code style, and versioning guidelines.
See CHANGELOG.md for release history.