meter-sdk 0.7.2

Python SDK for Meter Scraper API

Meter Scraper API SDK

Python SDK for the Meter Scraper API - a powerful web scraping service with LLM-powered strategy generation, job execution, and scheduling.

Features

• Simple API: Clean, Pythonic interface for all API operations
• LLM-Powered Strategies: Generate extraction strategies using natural language descriptions
• API-Based Scraping: Capture underlying APIs with force_api for dynamic sites
• Strategy Refinement: Iteratively improve strategies with feedback
• Job Execution: Run scrapes with saved strategies (no LLM costs on execution)
• API Parameters: Override parameters at runtime for API-based strategies
• Batch Jobs: Scrape multiple URLs in a single request
• Content Analysis: Track changes with content hashing, structural signatures, and semantic similarity
• Scheduling: Set up recurring scrapes with interval or cron expressions
• Keyword Filtering: Filter change results with Lucene-style syntax
• Error Handling: Comprehensive error handling with custom exceptions
• Type Hints: Full type annotations for better IDE support

Installation

pip install meter-sdk

Or install from source:

git clone https://github.com/reverse/meter-sdk
cd meter-sdk
pip install -e .

Quick Start

from meter_sdk import MeterClient

# Initialize client with your API key
client = MeterClient(api_key="sk_live_")

# Generate a strategy using LLM
result = client.generate_strategy(
    url="https://example.com/products",
    description="Extract product names and prices",
    name="Product Scraper"
)

strategy_id = result["strategy_id"]
print(f"Generated strategy: {strategy_id}")
print(f"Preview data: {result['preview_data']}")

# Create and run a scrape job
job = client.create_job(
    strategy_id=strategy_id,
    url="https://example.com/products"
)

# Wait for job to complete (automatically polls)
completed_job = client.wait_for_job(job["job_id"])
results = completed_job["results"]

print(f"Scraped {len(results)} items")
for item in results:
    print(item)

Authentication

Getting an API Key

The SDK uses API key authentication. API keys are created on the frontend using Supabase Auth. Once you have an API key (starts with sk_live_), use it to initialize the client:

from meter_sdk import MeterClient
import os

# Load from environment variable (recommended)
api_key = os.getenv("METER_API_KEY")
client = MeterClient(api_key=api_key)

# Or use directly
client = MeterClient(api_key="sk_live_")

from meter_sdk import MeterClient
import os

# Load from environment variable (recommended)
api_key = os.getenv("METER_API_KEY")
client = MeterClient(api_key=api_key)

# Or use directly
client = MeterClient(api_key="sk_live_")

Core Concepts

Strategies

A strategy is an extraction plan generated by the LLM that tells the scraper how to extract data from a webpage. Strategies are reusable - once created, you can run multiple jobs with the same strategy without incurring LLM costs.

Jobs

A job is a single execution of a scrape using a strategy. Jobs run asynchronously and can be polled for status and results.

Schedules

A schedule automatically runs jobs at specified intervals or cron times, making it easy to monitor websites for changes.

Usage Guide

Strategy Management

Generate a Strategy

Generate a new extraction strategy using natural language:

result = client.generate_strategy(
    url="https://example.com/products",
    description="Extract product names, prices, and descriptions",
    name="E-commerce Product Scraper"
)

# Response includes:
# - strategy_id: UUID of the created strategy
# - strategy: The extraction strategy (JSON)
# - preview_data: Sample extracted data
# - attempts: Number of LLM attempts (usually 1)

strategy_id = result["strategy_id"]
print(f"Strategy created: {strategy_id}")
print(f"Preview: {result['preview_data']}")

The LLM uses a two-stage approach:

1. Haiku analysis: Quick analysis of the page structure
2. Sonnet generation: Detailed strategy creation

Refine a Strategy

If the initial strategy doesn't capture everything you need, refine it with feedback:

# First, check the preview data
result = client.generate_strategy(...)

# If something is missing, refine it
refined = client.refine_strategy(
    strategy_id=result["strategy_id"],
    feedback="The strategy is missing the product images. Also, extract the SKU field."
)

# The refined strategy uses cached HTML (no re-fetching)
# You can refine multiple times
refined_again = client.refine_strategy(
    strategy_id=result["strategy_id"],
    feedback="The price should include the currency symbol"
)

Refinement is fast and cost-effective because it uses cached HTML from the initial generation.

List Strategies

# Get all strategies
strategies = client.list_strategies(limit=20, offset=0)

for strategy in strategies:
    print(f"{strategy['name']}: {strategy['strategy_id']}")
    print(f"  URL: {strategy['url']}")
    print(f"  Created: {strategy['created_at']}")

Get Strategy Details

strategy = client.get_strategy(strategy_id)

print(f"Name: {strategy['name']}")
print(f"Description: {strategy['description']}")
print(f"Preview data: {strategy['preview_data']}")
print(f"Attempts: {strategy['attempts']}")

Delete a Strategy

client.delete_strategy(strategy_id)

Job Execution

Create a Job

Create a scrape job using an existing strategy:

job = client.create_job(
    strategy_id="your-strategy-uuid",
    url="https://example.com/products"
)

job_id = job["job_id"]
status = job["status"]  # "pending"

Jobs run asynchronously in the background. No LLM costs are incurred during job execution - the strategy is reused.

Check Job Status

job = client.get_job(job_id)

print(f"Status: {job['status']}")  # pending, running, completed, failed

if job["status"] == "completed":
    results = job["results"]
    print(f"Scraped {job['item_count']} items")
    print(f"Content hash: {job['content_hash']}")
elif job["status"] == "failed":
    print(f"Error: {job['error']}")

Wait for Job Completion

The SDK provides a convenient method to poll a job until it completes:

# Wait indefinitely (default: polls every 1 second)
completed_job = client.wait_for_job(job_id)

# With timeout (raises MeterError if timeout exceeded)
try:
    completed_job = client.wait_for_job(
        job_id,
        poll_interval=2.0,  # Check every 2 seconds
        timeout=300.0  # 5 minute timeout
    )
    results = completed_job["results"]
except MeterError as e:
    print(f"Job failed or timed out: {e}")

List Jobs

# Get all jobs
all_jobs = client.list_jobs(limit=50, offset=0)

# Filter by strategy
strategy_jobs = client.list_jobs(strategy_id="your-strategy-uuid")

# Filter by status
completed_jobs = client.list_jobs(status="completed")

# Combined filters
recent_completed = client.list_jobs(
    strategy_id="your-strategy-uuid",
    status="completed",
    limit=10
)

Compare Jobs

Compare two jobs to detect changes:

comparison = client.compare_jobs(job_id_1, job_id_2)

print(f"Content hash match: {comparison['content_hash_match']}")
print(f"Structural match: {comparison['structural_match']}")
print(f"Semantic similarity: {comparison['semantic_similarity']}")  # 0.0-1.0
print(f"Item count difference: {comparison['item_count_diff']}")

if comparison['structural_changes']:
    print("Structural changes detected:")
    for change in comparison['structural_changes']:
        print(f"  - {change}")

Get Strategy History

Get a timeline of all jobs for a strategy:

history = client.get_strategy_history(strategy_id)

for entry in history:
    print(f"Job {entry['job_id']}: {entry['status']}")
    print(f"  Items: {entry['item_count']}")
    print(f"  Has changes: {entry['has_changes']}")
    print(f"  Created: {entry['created_at']}")

The has_changes field indicates if content changed compared to the previous job.

Schedule Management

Create a Schedule (Interval)

Run a scrape at regular intervals:

# Run every hour (3600 seconds)
schedule = client.create_schedule(
    strategy_id="your-strategy-uuid",
    url="https://example.com/products",
    interval_seconds=3600
)

print(f"Schedule ID: {schedule['schedule_id']}")
print(f"Next run: {schedule['next_run_at']}")

Create a Schedule (Cron)

Use cron expressions for more complex schedules:

# Run daily at 9 AM
schedule = client.create_schedule(
    strategy_id="your-strategy-uuid",
    url="https://example.com/products",
    cron_expression="0 9 * * *"
)

# Run every weekday at 8 AM
schedule = client.create_schedule(
    strategy_id="your-strategy-uuid",
    url="https://example.com/products",
    cron_expression="0 8 * * 1-5"
)

Create a Schedule with Webhook

You can optionally provide a webhook URL to receive scrape results:

# Create schedule with webhook for receiving results
schedule = client.create_schedule(
    strategy_id="your-strategy-uuid",
    url="https://example.com/products",
    interval_seconds=3600,
    webhook_url="https://your-app.com/webhooks/scrape-results"
)

List Schedules

schedules = client.list_schedules()

for schedule in schedules:
    print(f"{schedule['schedule_id']}: {schedule['schedule_type']}")
    print(f"  Enabled: {schedule['enabled']}")
    print(f"  Next run: {schedule['next_run_at']}")

Update a Schedule

# Disable a schedule
client.update_schedule(schedule_id, enabled=False)

# Change the interval
client.update_schedule(
    schedule_id,
    interval_seconds=7200  # Every 2 hours
)

# Change to cron expression
client.update_schedule(
    schedule_id,
    cron_expression="0 10 * * *"  # Daily at 10 AM
)

# Update webhook URL
client.update_schedule(
    schedule_id,
    webhook_url="https://your-new-webhook-url.com/results"
)

Delete a Schedule

client.delete_schedule(schedule_id)

Complete Workflow Examples

Example 1: API-Based Scraping with Parameters

For sites that load data via JavaScript APIs, use force_api=True to capture the underlying API:

from meter_sdk import MeterClient

client = MeterClient(api_key="sk_live_...")

# Generate strategy with API capture
strategy = client.generate_strategy(
    url="https://jobs.example.com/listings",
    description="Extract job titles, companies, salaries, and locations",
    name="Job Listings API",
    force_api=True  # Force API-based capture
)

# Check the scraper type and available parameters
print(f"Scraper type: {strategy['scraper_type']}")  # 'api' or 'css'
if strategy.get('api_parameters'):
    print(f"Available parameters: {strategy['api_parameters']}")
    # e.g., {'page': 1, 'limit': 20, 'category': 'all', 'location': 'remote'}

# Run job with custom parameters
job = client.create_job(
    strategy_id=strategy["strategy_id"],
    url="https://jobs.example.com/api/listings",
    parameters={
        "category": "engineering",
        "location": "remote",
        "limit": 100
    }
)

results = client.wait_for_job(job["job_id"])
print(f"Found {results['item_count']} matching jobs")

Example 2: E-commerce Product Monitoring

from meter_sdk import MeterClient
import os

client = MeterClient(api_key=os.getenv("METER_API_KEY"))

# Step 1: Generate strategy
strategy = client.generate_strategy(
    url="https://example-store.com/products",
    description="Extract product name, price, availability status, and product URL",
    name="Product Monitor"
)

strategy_id = strategy["strategy_id"]
print(f"Strategy created: {strategy_id}")

# Step 2: Run initial scrape
job = client.create_job(strategy_id, "https://example-store.com/products")
initial_results = client.wait_for_job(job["job_id"])

print(f"Initial scrape: {initial_results['item_count']} products")

# Step 3: Set up daily monitoring
schedule = client.create_schedule(
    strategy_id=strategy_id,
    url="https://example-store.com/products",
    cron_expression="0 9 * * *"  # Daily at 9 AM
)

print(f"Monitoring schedule created: {schedule['schedule_id']}")

# Step 4: Check for changes later
history = client.get_strategy_history(strategy_id)
if len(history) > 1:
    latest = history[0]
    previous = history[1]

    if latest["has_changes"]:
        print("Changes detected!")
        comparison = client.compare_jobs(latest["job_id"], previous["job_id"])
        print(f"Semantic similarity: {comparison['semantic_similarity']}")

Example 2: News Article Scraping

from meter_sdk import MeterClient

client = MeterClient(api_key="sk_live_...")

# Generate strategy for news articles
strategy = client.generate_strategy(
    url="https://news.example.com/latest",
    description="Extract article headlines, authors, publication dates, and article URLs",
    name="News Scraper"
)

# Refine to include article summaries
refined = client.refine_strategy(
    strategy_id=strategy["strategy_id"],
    feedback="Also extract the article summary/excerpt if available"
)

# Run scrape
job = client.create_job(
    strategy_id=strategy["strategy_id"],
    url="https://news.example.com/latest"
)

results = client.wait_for_job(job["job_id"])["results"]

for article in results:
    print(f"{article['headline']} by {article['author']}")
    print(f"  Published: {article['publication_date']}")
    print(f"  URL: {article['url']}")

Example 3: Real Estate Listings

from meter_sdk import MeterClient

client = MeterClient(api_key="sk_live_...")

# Create strategy
strategy = client.generate_strategy(
    url="https://realestate.example.com/listings",
    description="Extract property address, price, bedrooms, bathrooms, square footage, and listing URL",
    name="Real Estate Monitor"
)

# Set up hourly monitoring
schedule = client.create_schedule(
    strategy_id=strategy["strategy_id"],
    url="https://realestate.example.com/listings",
    interval_seconds=3600  # Every hour
)

# Check results periodically
jobs = client.list_jobs(
    strategy_id=strategy["strategy_id"],
    status="completed",
    limit=10
)

for job_data in jobs:
    job = client.get_job(job_data["id"])
    print(f"Scrape at {job['completed_at']}: {job['item_count']} listings")

Error Handling

The SDK raises MeterError for all API errors:

from meter_sdk import MeterClient, MeterError

client = MeterClient(api_key="sk_live_...")

try:
    strategy = client.generate_strategy(
        url="https://example.com",
        description="Extract data",
        name="Test"
    )
except MeterError as e:
    print(f"API error: {e}")
    # Handle error (invalid API key, rate limit, etc.)

try:
    job = client.wait_for_job(job_id, timeout=60.0)
except MeterError as e:
    print(f"Job error: {e}")
    # Handle timeout or job failure

Common error scenarios:

• 401 Unauthorized: Invalid or missing API key
• 400 Bad Request: Invalid request parameters
• 404 Not Found: Resource doesn't exist
• 500 Internal Server Error: Server-side error

Advanced Usage

Context Manager

The client can be used as a context manager for automatic cleanup:

with MeterClient(api_key="sk_live_...") as client:
    strategies = client.list_strategies()
    # Client automatically closes HTTP connections

Custom Base URL

For development or custom deployments:

client = MeterClient(
    api_key="sk_live_...",
    base_url="http://localhost:8000"  # Local development
)

Pagination

For endpoints that support pagination:

# List strategies with pagination
offset = 0
limit = 20
all_strategies = []

while True:
    strategies = client.list_strategies(limit=limit, offset=offset)
    if not strategies:
        break
    all_strategies.extend(strategies)
    offset += limit

API Reference

MeterClient

Main client class for interacting with the API.

Constructor

MeterClient(api_key: str, base_url: str = "https://api.meter.sh")

Strategy Methods

• generate_strategy(url: str, description: str, name: str, force_api: bool = False) -> Dict
• refine_strategy(strategy_id: str, feedback: str) -> Dict
• list_strategies(limit: int = 20, offset: int = 0) -> List[Dict]
• get_strategy(strategy_id: str) -> Dict
• delete_strategy(strategy_id: str) -> Dict

Job Methods

• create_job(strategy_id: str, url: Optional[str] = None, urls: Optional[List[str]] = None, parameters: Optional[Dict] = None) -> Dict
• execute_job(strategy_id: str, url: str, parameters: Optional[Dict] = None) -> Dict
• get_job(job_id: str) -> Dict
• list_jobs(strategy_id: Optional[str] = None, status: Optional[str] = None, limit: int = 20, offset: int = 0) -> List[Dict]
• wait_for_job(job_id: str, poll_interval: float = 1.0, timeout: Optional[float] = None) -> Dict
• compare_jobs(job_id: str, other_job_id: str) -> Dict
• get_strategy_history(strategy_id: str) -> List[Dict]

Schedule Methods

• create_schedule(strategy_id: str, url: Optional[str] = None, urls: Optional[List[str]] = None, interval_seconds: Optional[int] = None, cron_expression: Optional[str] = None, webhook_url: Optional[str] = None, parameters: Optional[Dict] = None) -> Dict
• list_schedules() -> List[Dict]
• update_schedule(schedule_id: str, enabled: Optional[bool] = None, url: Optional[str] = None, urls: Optional[List[str]] = None, interval_seconds: Optional[int] = None, cron_expression: Optional[str] = None, webhook_url: Optional[str] = None, parameters: Optional[Dict] = None) -> Dict
• delete_schedule(schedule_id: str) -> Dict
• get_schedule_changes(schedule_id: str, mark_seen: bool = True, filter: Optional[str] = None) -> Dict

MeterError

Exception raised for all API errors.

class MeterError(Exception):
    """Base exception for Meter SDK errors"""
    pass

Response Formats

All methods return dictionaries matching the API response format. See the API documentation for detailed response schemas.

Key response fields:

• Strategy responses: strategy_id, strategy, preview_data, attempts, scraper_type ('css' or 'api'), api_parameters (for API strategies)
• Job responses: job_id, status, results, item_count, content_hash, structural_signature, parameters (if API strategy)
• Schedule responses: id, strategy_id, url, urls, schedule_type, interval_seconds, cron_expression, enabled, webhook_url, parameters, next_run_at, last_run_at, created_at, updated_at

Best Practices

1. Store API keys securely: Use environment variables or secure storage, never hardcode
2. Handle errors gracefully: Always wrap API calls in try/except blocks
3. Use timeouts: Set appropriate timeouts for wait_for_job() to avoid hanging
4. Reuse strategies: Generate once, use many times to avoid LLM costs
5. Monitor schedules: Regularly check schedule status and job history
6. Use context managers: Use with statement for automatic resource cleanup
7. Poll efficiently: Use appropriate poll_interval values for wait_for_job()

Troubleshooting

Connection Errors

If you see connection errors, check:

• API key is valid and not expired
• Base URL is correct (default: https://api.meter.sh)
• Network connectivity

Job Timeouts

If jobs frequently timeout:

• Check if the target URL is accessible
• Verify the strategy is correct
• Check API logs for errors

Strategy Generation Fails

If strategy generation fails:

• Ensure the URL is accessible
• Provide clear, specific descriptions
• Check API logs for LLM errors

License

MIT

Support

For API documentation and interactive testing, visit https://docs.meter.sh/