DEV Community: Roberto Francisco junior

Recovering batch and crawl jobs after a worker restart

Roberto Francisco junior — Sun, 13 Sep 2026 16:45:52 +0000

Asynchronous scraping changes the failure model. POST /batch, /crawl, and /search return a job_id before the work finishes. If the process that submitted the request exits, the server-side job can keep running, but a client that discarded the identifier has no reliable way to collect the result.

Persist the job identifier before starting the polling loop. Treat the identifier as a durable handle, not as a temporary variable in a web request.

Store the enqueue record

import os
import time
import requests

API = "https://api.messora.dev"
HEADERS = {"X-API-Key": os.environ["MESSORA_API_KEY"]}


def enqueue_batch(urls: list[str], save_job) -> str:
    response = requests.post(
        f"{API}/batch",
        headers=HEADERS,
        json={"urls": urls, "formats": ["markdown"]},
        timeout=30,
    )
    response.raise_for_status()       # 202 Accepted
    job_id = response.json()["job_id"]

    # Save before returning control to the rest of the pipeline.
    save_job({
        "job_id": job_id,
        "kind": "batch",
        "state": "submitted",
        "submitted_urls": len(urls),
    })
    return job_id

The persisted record should include the operation type, the input reference, enqueue time, and a local processing state. Store the URLs in object storage or a database table when the list is large; do not make a 50-item request payload the only copy of your input.

Resume with GET /jobs/{job_id}

The same status endpoint serves jobs created by batch, crawl, and search. Polling is recommended every 2 seconds. The response contains a status and, when the job succeeds, the results. A restarted worker can read all records in submitted or polling state and resume without submitting duplicates.

TERMINAL = {"SUCCESS", "FAILURE", "REVOKED"}


def collect_job(job_id: str, timeout_s: int = 600) -> dict:
    deadline = time.monotonic() + timeout_s
    last_status = "UNKNOWN"

    while time.monotonic() < deadline:
        response = requests.get(
            f"{API}/jobs/{job_id}",
            headers=HEADERS,
            timeout=30,
        )
        response.raise_for_status()
        job = response.json()
        last_status = job["status"]

        if last_status in TERMINAL:
            return job
        time.sleep(2)

    raise TimeoutError(f"{job_id} remained {last_status} for {timeout_s}s")

A 404 from the status endpoint is not the same as PENDING. The API checks job ownership by tenant and returns 404 when the identifier does not exist or belongs to another account. Mark the local record as an operator-visible lookup failure and do not keep polling an identifier that the API cannot resolve.

Commit results idempotently

A worker can crash after receiving SUCCESS but before marking the local record complete. The next worker may see the same terminal job again. Write results with a uniqueness key such as (job_id, item_index) or (job_id, url), then mark the job complete in the same database transaction where possible.

def persist_results(job: dict, insert_if_missing, mark_complete) -> None:
    if job["status"] != "SUCCESS":
        raise RuntimeError(job.get("error") or job["status"])

    for item in job.get("results", []):
        insert_if_missing(
            job_id=job["job_id"],
            url=item["url"],
            scrape_status=item["scrape_status"],
            markdown=item.get("markdown"),
        )

    mark_complete(job["job_id"], credits_used=job.get("credits_used", 0))

insert_if_missing must be safe to call twice. Do not use the number of rows inserted as proof that the remote job ran only once. The remote job is identified by job_id; local delivery is a separate state machine.

Interpret partial success correctly

A batch or crawl can reach job-level SUCCESS while individual result entries carry blocked_antibot, timeout, or extraction_failed. Reconcile per-item status and the job's credits_used. Do not retry the entire job because one page failed; select only statuses your policy considers retriable, and submit a new job with a new local attempt record.

For search jobs, inspect stopped_reason as well. A token_cap result means the search stopped before processing every source, not that the status endpoint failed. Preserve that reason next to the results so downstream consumers know whether the response was complete.

Make restart recovery boring

The recovery loop is simple when three records are durable: the original input, the remote job_id, and the local delivery state. On startup, query unfinished records, poll each identifier with a bounded deadline, and commit terminal results idempotently. Never infer completion from a process exit, and never enqueue a replacement merely because the submitting process disappeared.

Asynchronous APIs are easier to operate when the client treats job submission and result delivery as two separate steps. Persisting the handle closes the gap between them and turns a worker restart into a normal retry of local observation, not a duplicate scrape.

Avoiding duplicate fetches with max_age_ms

Roberto Francisco junior — Sat, 12 Sep 2026 15:56:33 +0000

Scheduled ingestion often revisits the same pages: a product catalog every hour, documentation every morning, or a set of policy pages after a deployment. Fetching every URL from scratch wastes time when the source has not changed. The MESSORA /scrape endpoint accepts max_age_ms so a caller can opt into a recent cached result for the same request.

The default is no cache lookup. That is the safe choice for a fresh scrape. Set a positive max_age_ms only when stale content is acceptable for the job's purpose. The accepted range is 0 through 2,592,000,000 milliseconds, or 30 days.

Opt in for a bounded freshness window

import os
import requests

API = "https://api.messora.dev"
HEADERS = {"X-API-Key": os.environ["MESSORA_API_KEY"]}

response = requests.post(
    f"{API}/scrape",
    headers=HEADERS,
    json={
        "url": "https://example.com/pricing",
        "formats": ["markdown"],
        "only_main_content": True,
        "max_age_ms": 3_600_000,  # accept a result up to one hour old
    },
    timeout=90,
)
response.raise_for_status()
payload = response.json()

print(payload["markdown"])
print(payload.get("diagnostics", {}).get("from_cache", False))

A one-hour window is appropriate for a page where hourly changes are not important. A legal notice or incident page may need max_age_ms: 0, which explicitly bypasses the cache. Do not use a long freshness window just because a scrape is expensive; choose it from the data's acceptable staleness.

Know what identifies a cached result

The cache key includes the scrape request fingerprint, not only the URL. Changes to output format or extraction parameters should not be treated as the same result. Keep the request stable between runs if you want cache hits: changing only_main_content, render_js, wait_for, or the requested format can represent a different extraction.

The response exposes cache provenance in diagnostics.from_cache. Use it in metrics and in a persisted ingestion record. A cache hit returns credits_used: 0 for that request and restores the credit that was provisioned for the attempt. That lets a frequent refresh avoid paying for a duplicate fetch while still making the freshness decision explicit.

def scrape_and_record(url: str, max_age_ms: int | None) -> dict:
    payload = requests.post(
        f"{API}/scrape",
        headers=HEADERS,
        json={
            "url": url,
            "formats": ["markdown"],
            "max_age_ms": max_age_ms,
        },
        timeout=90,
    )
    payload.raise_for_status()
    data = payload.json()

    if data["scrape_status"] != "success":
        raise RuntimeError(data.get("error") or data["scrape_status"])

    return {
        "url": url,
        "markdown": data["markdown"],
        "from_cache": data.get("diagnostics", {}).get("from_cache", False),
        "credits_used": data["credits_used"],
    }

Persisting from_cache prevents an operator from mistaking a repeated cached read for a fresh observation. It also gives you a way to compare freshness policy with actual behavior when a site changes.

Do not confuse cache with a source change check

A cache hit says that the API has a completed result within the requested age. It does not prove that the origin currently serves identical bytes. If the workflow needs to detect every source update, use a fresh scrape and compare a normalized hash of the returned content. If the workflow only needs a boundedly fresh copy, the cache is cheaper and simpler.

A cache miss follows the normal scrape path. The result is stored only after a successful extraction, using the requested age as the cache TTL. Failed or incomplete fetches should remain visible as failures rather than poisoning future runs with an empty body.

Choose policy per dataset

Keep cache settings in the dataset configuration, not hidden inside a generic HTTP helper. A news feed, a product page, and a versioned API reference have different freshness requirements. Review the policy when consumers change: a value that was harmless for a search index may be wrong for an alerting system.

max_age_ms is a small parameter with a clear tradeoff. Zero favors freshness and predictable observation. A positive value favors lower latency and fewer duplicate fetches. Recording cache provenance and actual credits makes that tradeoff measurable instead of leaving it as an undocumented optimization.

Writing a scraping client that respects API rate limits

Roberto Francisco junior — Fri, 11 Sep 2026 16:50:06 +0000

A rate limit is part of an API contract. Treating 429 as an exception to hide or as a signal to retry immediately turns a temporary throttle into a longer outage. A client should know which endpoint it is calling, pace requests before the limit, and use bounded backoff when the server still rejects a request.

The public MESSORA limits differ by operation. /scrape allows 10 requests per minute on the free plan. /crawl also allows 10 enqueues per minute, /search allows 2 enqueues per minute per tenant, and /account/usage allows 30 requests per minute. /batch scales its enqueue limit with the plan: 10, 60, 120, or 300 requests per minute for free, starter, growth, and scale.

Back off instead of looping

import os
import random
import time

import requests


API = "https://api.messora.dev"
HEADERS = {"X-API-Key": os.environ["MESSORA_API_KEY"]}
RETRYABLE = {429, 500, 502, 503, 504}


def post_with_backoff(
    url: str,
    *,
    headers: dict,
    payload: dict,
    attempts: int = 5,
) -> requests.Response:
    delay = 1.0
    for attempt in range(attempts):
        response = requests.post(
            url,
            headers=headers,
            json=payload,
            timeout=30,
        )
        if response.status_code not in RETRYABLE:
            response.raise_for_status()
            return response

        if attempt == attempts - 1:
            response.raise_for_status()

        # Jitter prevents several workers from retrying on the same tick.
        time.sleep(delay + random.uniform(0, 0.25))
        delay = min(delay * 2, 30.0)

    raise AssertionError("unreachable")

The retry set in this example is intentionally narrow. A 401 or 403 is a credential problem, and a 422 is invalid input; repeating either request does not repair it. A 429 means the client should slow down. A 5xx may be transient, but the caller still needs a maximum attempt count and a visible final error.

Pace before the server answers

Backoff only helps after a rejection. For a known limit, pace enqueues in the client as well. A simple worker can keep one timestamp per endpoint and sleep until the next permitted slot. A shared scheduler is more accurate when several processes use the same API key; independent workers otherwise each believe they are below the limit while their combined traffic exceeds it.

Do not apply the /search limit to /scrape or treat all operations as interchangeable. Search starts a premium multi-source job and is limited to two enqueues per minute. If an agent asks three search questions in a short loop, queue the third instead of launching a burst and depending on retries to serialize it.

Poll jobs at the documented cadence

/batch, /crawl, and /search return a job_id and run asynchronously. The recommended polling interval for GET /jobs/{job_id} is 2 seconds. Keep the polling loop separate from enqueue throttling: the limit on creating work and the protection against excessive status reads solve different problems.

TERMINAL = {"SUCCESS", "FAILURE", "REVOKED"}


def wait_for_job(job_id: str, headers: dict, timeout_s: int = 600) -> dict:
    deadline = time.monotonic() + timeout_s
    while time.monotonic() < deadline:
        response = requests.get(
            f"https://api.messora.dev/jobs/{job_id}",
            headers=headers,
            timeout=30,
        )
        if response.status_code == 429:
            time.sleep(5)
            continue
        response.raise_for_status()
        job = response.json()
        if job["status"] in TERMINAL:
            return job
        time.sleep(2)

    raise TimeoutError(f"job {job_id} exceeded {timeout_s}s")

A 429 during polling should not be interpreted as a failed scrape. Preserve the job_id, slow the next read, and continue until your overall deadline. If the client process restarts, persist the identifier so it can resume polling instead of submitting a duplicate job.

Make throttling observable

Count attempts, 429 responses, time spent sleeping, and jobs that hit the overall deadline. Include the endpoint and plan in metrics, never the API key. A rising 429 count on /batch may mean the plan limit is too low for the workload; the same count on /search usually means the caller needs a queue or a cache.

The goal is not to make retries invisible. It is to make them bounded, attributable, and rare. A client that paces work, honors the operation-specific limits, and polls at a controlled interval spends its request budget on extraction instead of fighting the service that performs it.

Keeping a crawl inside the documentation tree

Roberto Francisco junior — Thu, 10 Sep 2026 16:47:47 +0000

A crawl that starts at /docs/ can reach a login page, a changelog, a language switcher, and every marketing link in the site header. The crawler is doing what the HTML tells it to do. The boundary has to be part of the request.

MESSORA's /crawl endpoint exposes three controls for that boundary: url_regex filters discovered absolute URLs, max_pages caps the total work, and follow_subdomains decides whether links on subdomains are eligible. The endpoint runs a breadth-first crawl and returns a job_id immediately.

Start with a small page budget

import os
import requests

API = "https://api.messora.dev"
HEADERS = {"X-API-Key": os.environ["MESSORA_API_KEY"]}

response = requests.post(
    f"{API}/crawl",
    headers=HEADERS,
    json={
        "url": "https://docs.example.com/",
        "max_pages": 10,
        "max_depth": 2,
        "url_regex": r"^https://docs\.example\.com/(guide|reference)(/.*)?$",
        "follow_subdomains": False,
        "only_main_content": True,
        "include_links": True,
    },
    timeout=30,
)
response.raise_for_status()       # 202 Accepted
job_id = response.json()["job_id"]
print(job_id)

max_pages is required and accepts values from 1 to 50. A value of 10 is a useful first run: it exposes whether the regex matches the site's real link shape without committing the account to a full tree. Increase the limit only after inspecting the result set and credit use.

Match the final URL shape

url_regex is evaluated against absolute URLs discovered by the crawler. Anchor the expression when the crawl must stay under one host. The expression above accepts /guide/ and /reference/, but not /blog/, /login, or a different hostname.

Python's raw string notation keeps the backslashes readable. The dot in docs.example.com is escaped because an unescaped dot matches any character. Test the expression with representative URLs before sending it:

import re

pattern = re.compile(r"^https://docs\.example\.com/(guide|reference)(/.*)?$")

candidates = [
    "https://docs.example.com/guide/install",
    "https://docs.example.com/reference/api",
    "https://docs.example.com/blog/release",
]

for url in candidates:
    print(pattern.fullmatch(url) is not None, url)

The seed itself must still be a valid URL and pass the API's network safety policy. A regex is a discovery filter, not a permission to fetch an otherwise forbidden destination.

Choose subdomain behavior explicitly

follow_subdomains defaults to false. Keep it disabled when the documentation is isolated on one host. Enable it only when the site intentionally spreads its public documentation across subdomains, and widen the regex to describe those hosts. Otherwise a link to status.example.com or community.example.com can become part of the crawl without adding useful source material.

max_depth is optional. It limits traversal depth, while max_pages limits total pages and therefore provides the hard cost ceiling. Use both: depth controls shape, page count controls spend. The crawl returns Markdown in the current public contract; use /scrape or /batch when you need other output formats.

Inspect why it stopped

After the job reaches SUCCESS, read pages_crawled, stopped_reason, and each result's scrape_status. Fewer pages than max_pages is not automatically an error. completed can mean the frontier ran out of matching links. A non-success item can be a timeout, an extraction failure, or an anti-bot block.

job = wait_for_job(job_id)

print("pages:", job["pages_crawled"])
print("stop:", job["stopped_reason"])

for item in job["results"]:
    if item["scrape_status"] == "success":
        print("kept:", item["url"])
    else:
        print("not extracted:", item["url"], item["scrape_status"])

Do not solve an unexpectedly broad crawl by lowering max_pages alone. That hides the symptom and produces a partial corpus. Tighten the regex, verify subdomain behavior, then raise the budget only when the boundary is intentional. A bounded crawl is easier to review, cheaper to rerun, and safer to put in a scheduled ingestion pipeline.

Why a scraping API rejects private and metadata URLs

Roberto Francisco junior — Wed, 09 Sep 2026 16:57:21 +0000

A scraper accepts a URL and makes a server-side network request. That makes URL validation a security boundary, not a convenience check. If an API worker can fetch 127.0.0.1, a private network address, or a cloud metadata service, an untrusted caller may use the worker to inspect infrastructure that was never meant to be public.

MESSORA applies an SSRF guard to /scrape, /crawl, and discovered crawl links. URLs in RFC1918 ranges, loopback, link-local ranges, and cloud metadata targets are rejected. The request fails with 422; it does not create a job or return a partial page.

The API is the final authority

A client-side check is useful for fast feedback, but it cannot replace the server guard. DNS can resolve differently from one network to another, redirects can point to a different host, and a public hostname may resolve to an internal address from the worker's network.

import ipaddress
import requests
from urllib.parse import urlparse


def obvious_private_target(url: str) -> bool:
    parsed = urlparse(url)
    if parsed.scheme not in {"http", "https"} or not parsed.hostname:
        return True

    try:
        address = ipaddress.ip_address(parsed.hostname)
    except ValueError:
        # A hostname still needs server-side validation and DNS-aware policy.
        return False

    return any(
        (
            address.is_private,
            address.is_loopback,
            address.is_link_local,
            address.is_reserved,
        )
    )


for candidate in [
    "http://127.0.0.1:8080/health",
    "http://192.168.1.20/admin",
    "https://public.example.com/docs",
]:
    print(candidate, obvious_private_target(candidate))

This helper deliberately does not claim that a hostname is safe when it is not a literal IP. It only avoids sending obvious bad input. Submit accepted URLs to the API and handle its 422 response as the authoritative decision.

Crawl validation happens twice

A crawl starts with one seed, then follows links discovered in the returned pages. Validating only the seed would leave a gap: a public page could contain a link to an internal address. The API applies the guard to the seed and each discovered link before that link enters the breadth-first frontier.

This is why a crawl can complete with fewer pages than max_pages. A link may be outside the allowed URL pattern, point to another host when follow_subdomains is false, or be rejected by the SSRF policy. The job's results and stopped_reason tell you whether the frontier ended normally; the client should not infer a security failure from page count alone.

Do not turn failures into retries

A 422 for a private target is deterministic. Retrying the same URL only repeats validation and adds noise to your logs. Separate input rejection from transient fetch outcomes:

response = requests.post(
    f"{API}/scrape",
    headers={"X-API-Key": os.environ["MESSORA_API_KEY"]},
    json={"url": url, "formats": ["markdown"]},
    timeout=90,
)

if response.status_code == 422:
    detail = response.json().get("detail", "invalid request")
    raise ValueError(f"URL rejected by API policy: {detail}")

response.raise_for_status()

Do not log full URLs if they may contain credentials or sensitive query strings. Normalize and redact those values before putting the rejection into an operator-facing log. The API key also belongs only in the request header and environment, never in the URL.

Safe URL intake is a system property

If users can submit arbitrary URLs, validate the scheme, enforce a maximum length, and keep redirects under the same network policy. If a crawler accepts a user-supplied regex, bound its length and reject malformed patterns before enqueueing. These controls reduce accidental work, but the worker-side SSRF guard remains necessary because only the worker sees the final network destination.

A 422 is the correct result here: the request was understood, but its target violates the API's input policy. Treating it as a retriable outage obscures the real boundary. Treating it as a successful empty scrape risks silently dropping a security decision. Keep the distinction explicit in the client and in your metrics.

Getting stable JSON from a page with a schema

Roberto Francisco junior — Tue, 08 Sep 2026 16:57:34 +0000

Markdown is useful for reading and indexing, but downstream code usually needs a typed record: a product name, a price, a release date, or a list of features. The MESSORA /scrape endpoint can return structured JSON when the request includes "json" in formats and supplies a json_schema.

The schema is the boundary between page text and application data. Keep it narrow. A small contract is easier to validate, compare across pages, and repair when a site changes its layout.

Send a schema with the request

import os
import requests

API = "https://api.messora.dev"
HEADERS = {"X-API-Key": os.environ["MESSORA_API_KEY"]}

product_schema = {
    "type": "object",
    "properties": {
        "name": {"type": "string"},
        "price": {"type": ["number", "null"]},
        "currency": {"type": ["string", "null"]},
        "availability": {"type": "string"},
    },
    "required": ["name", "price", "currency", "availability"],
    "additionalProperties": False,
}

response = requests.post(
    f"{API}/scrape",
    headers=HEADERS,
    json={
        "url": "https://example.com/products/widget",
        "formats": ["json"],
        "json_schema": product_schema,
        "only_main_content": True,
    },
    timeout=90,
)
response.raise_for_status()
payload = response.json()

if payload["scrape_status"] != "success":
    raise RuntimeError(payload.get("error") or payload["scrape_status"])

record = payload["json"]
print(record["name"], record["price"], record["currency"])

json_schema is required whenever formats includes json. Omitting it is a request validation error, not a reason to accept an unstructured response. The response places the extracted object under json; the usual fields such as scrape_status, credits_used, and remaining_credits remain outside that object.

Keep extraction failures visible

A successful HTTP response does not guarantee that structured extraction produced useful values. When extraction fails completely, the API sets json_extraction_error_code to extraction_failed and keeps the schema properties in json with null values. Check that field before writing the record to a database.

def extract_record(payload: dict) -> dict:
    if payload["scrape_status"] != "success":
        raise RuntimeError(payload.get("error") or payload["scrape_status"])

    if payload.get("json_extraction_error_code") == "extraction_failed":
        raise ValueError("page fetched, but no structured record was extracted")

    record = payload.get("json")
    if not isinstance(record, dict):
        raise ValueError("response did not contain a JSON object")
    return record

Do not use a missing json value as the only failure signal. A page can be fetched successfully while a model cannot locate the requested fields. The explicit error code preserves that distinction for metrics and retry policy.

Make null a deliberate value

If a source sometimes omits a price, define price as number or null instead of forcing a fake zero. Zero means a free product; null means the page did not provide a usable price. The same rule applies to dates, ratings, and availability. Your schema should represent what the source can actually say.

For fields that need a controlled vocabulary, describe the values in the schema or in json_prompt. Do not put parser instructions only in a comment in your application. The request should carry the extraction contract that explains how the result is meant to be interpreted.

Validate before persistence

The API returns a JSON object, but the receiving service still owns its database invariants. Validate types, required fields, and domain constraints locally. Record the source URL and the extraction status alongside the normalized row, so a later review can distinguish a changed product from a failed extraction.

For a mixed pipeline, request both markdown and json during development. Markdown lets an operator inspect the source interpretation; JSON feeds the application. Once the schema is stable, request only the format you consume to keep payloads smaller. The cost remains one successful scrape, while the response contract becomes explicit instead of relying on positional text parsing.

Monitoring API credit usage before an automation runs out

Roberto Francisco junior — Mon, 07 Sep 2026 18:00:57 +0000

A scraper can be healthy and still stop producing data because the account balance reached zero. HTTP availability does not tell you whether a job can be paid for. The MESSORA API exposes the account's current-month usage through GET /account/usage, so an automation can check its budget before submitting more work.

The endpoint returns the plan, monthly allowance, credits already used, remaining credits, and a per-endpoint breakdown. It is a small response, but it gives a better control signal than counting requests in the client.

Read the usage endpoint

import os
import requests

API = "https://api.messora.dev"
HEADERS = {"X-API-Key": os.environ["MESSORA_API_KEY"]}


def get_usage() -> dict:
    response = requests.get(
        f"{API}/account/usage",
        headers=HEADERS,
        timeout=30,
    )
    response.raise_for_status()
    return response.json()


usage = get_usage()
print(f"plan: {usage['plan']}")
print(f"used: {usage['credits_used']}")
print(f"remaining: {usage['remaining_credits']}")

for row in usage["breakdown"]:
    print(row["endpoint"], row["credits_used"])

remaining_credits is the value to use for admission control. credits_used describes the current month, while monthly_credits is the plan allowance. The remaining value also includes top-up credits, so it is not always equal to monthly_credits - credits_used.

Reserve a local budget

Usage is a snapshot. Two workers can read the same remaining value and both decide to submit a job. The API remains authoritative, but a local reservation prevents your own workers from oversubscribing the same account.

class CreditBudget:
    def __init__(self, available: int, safety_margin: int = 0):
        self.available = available
        self.safety_margin = safety_margin

    def can_start(self, estimated_cost: int) -> bool:
        return estimated_cost <= self.available - self.safety_margin

    def reserve(self, estimated_cost: int) -> None:
        if not self.can_start(estimated_cost):
            raise RuntimeError(
                f"need {estimated_cost} credits, "
                f"only {max(self.available - self.safety_margin, 0)} are available"
            )
        self.available -= estimated_cost

This is not a replacement for an atomic server-side balance check. It is a coordination layer inside one process or queue. If several machines submit jobs, put the reservation in shared storage or let each worker refresh usage immediately before enqueueing.

Estimate by endpoint

The usage breakdown also tells you which operation is consuming the account. A successful /scrape costs one credit. A batch or crawl charges successful pages, not merely the number of jobs submitted. A premium /search charges one credit per result returned and performs a balance pre-check using the requested num_results.

That difference matters for a scheduler. A queue containing one crawl with max_pages: 50 can consume up to 50 credits, while a queue containing ten empty or blocked pages may consume fewer. Use the endpoint's worst-case cost when admitting work, then reconcile against the actual job result.

Treat errors differently

A 401 from /account/usage means the X-API-Key header is absent. A 403 means the key is invalid or revoked. Neither should be retried with a shorter delay; fix the credential or configuration. A 429 is the usage endpoint's own rate limit, documented as 30 requests per minute per tenant. Cache the last successful snapshot for a short interval instead of calling it before every page.

A practical policy is to refresh usage when a worker starts, before a large batch, and after a job completes. Log plan, credits_used, remaining_credits, and the endpoint breakdown, but never log the API key. When the local reservation cannot cover the worst-case cost, leave the job queued and surface a clear budget warning rather than allowing a predictable mid-run failure.

Budget checks turn "the scraper stopped" into an observable state: no balance, invalid credential, rate limit, or real extraction failure. Each case has a different operator action, and /account/usage supplies the evidence needed to choose it.

When a scraped page comes back empty and render_js is the answer

Roberto Francisco junior — Sun, 06 Sep 2026 15:55:03 +0000

A scrape returns scrape_status: "success", statusCode: 200, and 300 characters of Markdown consisting of a nav bar and a footer. The page in your browser shows a full article. Nothing failed — the HTML the server sent genuinely contains no content, because the content is assembled by JavaScript after load.

This is the single most common false-success in web extraction, and the diagnosis takes one request.

Confirming the diagnosis

Fetch the raw HTML and look for the content, not for errors:

import os
import requests

API = "https://api.messora.dev"
HEADERS = {"X-API-Key": os.environ["MESSORA_API_KEY"]}


def diagnose(url: str, needle: str) -> None:
    resp = requests.post(
        f"{API}/scrape",
        headers=HEADERS,
        json={"url": url, "formats": ["raw", "markdown"]},
        timeout=90,
    )
    resp.raise_for_status()
    data = resp.json()

    html = data.get("rawHtml") or ""
    md = data.get("markdown") or ""

    print(f"status      : {data['scrape_status']} / {data['metadata']['statusCode']}")
    print(f"raw html    : {len(html)} chars")
    print(f"markdown    : {len(md)} chars")
    print(f"needle in raw: {needle.lower() in html.lower()}")


diagnose("https://example.com/app/article/42", needle="conclusion")

If the needle is absent from rawHtml, no extraction setting will find it. The text does not exist in the response — it arrives later, from an XHR call the static fetch never makes.

Enabling rendering

json={
    "url": url,
    "formats": ["markdown"],
    "render_js": True,
    "wait_for": 2500,
    "only_main_content": True,
    "timeout": 120,
}

render_js runs a real browser engine. wait_for adds a fixed pause in milliseconds after load, which is the blunt instrument that works when you cannot know which request populates the DOM.

Raise timeout alongside render_js. Rendering a heavy single-page application routinely takes 8 to 15 seconds, and the default 60-second budget covers that, but a client-side timeout of 30 will not.

Calibrating wait_for

Guessing wait_for wastes either time or content. Measure it once:

def calibrate(url: str, needle: str, candidates=(0, 1000, 2500, 5000, 8000)) -> int | None:
    for wait in candidates:
        resp = requests.post(
            f"{API}/scrape",
            headers=HEADERS,
            json={
                "url": url,
                "formats": ["markdown"],
                "render_js": True,
                "wait_for": wait,
                "timeout": 120,
            },
            timeout=180,
        )
        resp.raise_for_status()
        md = resp.json().get("markdown") or ""
        found = needle.lower() in md.lower()
        print(f"wait_for={wait:>5}ms  chars={len(md):>7}  found={found}")
        if found:
            return wait
    return None

Run it against three representative URLs from the same site and take the highest value that worked, plus a margin. A site with a consistent framework has a consistent hydration time; per-URL tuning is over-fitting.

Every calibration run costs 1 credit per attempt, so five candidates across three URLs is 15 credits to permanently settle a parameter you would otherwise guess at forever.

Do not enable it by default

Rendering costs wall-clock time on every page. On a /batch of 50 URLs where 45 are static, blanket render_js turns a two-minute job into something much longer for no additional content.

The cheaper pattern is static first, render on failure:

def scrape_adaptive(url: str, min_chars: int = 500) -> dict:
    """Static fetch first; re-fetch with rendering only when the result looks empty."""
    base = {"url": url, "formats": ["markdown"], "only_main_content": True}

    resp = requests.post(f"{API}/scrape", headers=HEADERS, json=base, timeout=90)
    resp.raise_for_status()
    data = resp.json()

    thin = len((data.get("markdown") or "").strip()) < min_chars
    if data["scrape_status"] == "success" and not thin:
        return data

    resp = requests.post(
        f"{API}/scrape",
        headers=HEADERS,
        json={**base, "render_js": True, "wait_for": 2500, "timeout": 120},
        timeout=180,
    )
    resp.raise_for_status()
    return resp.json()

This costs 2 credits on pages that need rendering and 1 on pages that do not. Against a corpus that is mostly static, that beats paying the rendering latency everywhere.

When rendering is not the problem

Two cases look identical from the outside and rendering does not fix either:

scrape_status: "blocked_antibot" — the fetch was refused. Rendering does not change that verdict; the request never got content to render.
Content behind authentication. A page that requires a session returns the login form, rendered perfectly. The Markdown will be a complete, correct rendering of a page you did not want.

Check scrape_status before reaching for render_js. It is only the right lever when the status is success and the body is thin — that specific combination is what "the HTML is empty on purpose" looks like from the client side.

Reading PDFs found during a crawl, including scanned ones

Roberto Francisco junior — Sat, 05 Sep 2026 15:43:29 +0000

Half the useful content on government, academic, and enterprise sites is behind a .pdf link. A crawler that only understands HTML walks past it, and a RAG index built from that crawl has a hole exactly where the technical specifications live.

The MESSORA API handles PDFs in the same call as HTML. parse_pdf defaults to true, so a /scrape against a PDF URL returns Markdown without any special casing.

A PDF is just another URL

import os
import requests

API = "https://api.messora.dev"
HEADERS = {"X-API-Key": os.environ["MESSORA_API_KEY"]}

resp = requests.post(
    f"{API}/scrape",
    headers=HEADERS,
    json={"url": "https://example.gov.br/edital-2026.pdf", "formats": ["markdown"]},
    timeout=180,
)
resp.raise_for_status()
data = resp.json()

print(data["metadata"]["title"])
print(data["markdown"][:500])

Raise the client timeout for PDFs. A 200-page document takes materially longer than an HTML page, and a 90-second client timeout on a job the server is still working through wastes the credit you already committed.

Turning it off on purpose

parse_pdf: false makes PDF URLs fail fast instead of consuming time and credits:

json={"url": url, "formats": ["markdown"], "parse_pdf": False}

That is the right setting for a crawl of a marketing site where the only PDFs are brochures you do not want in the index. You are trading coverage for predictable cost.

Scanned documents need OCR

A PDF produced by a scanner has no text layer — it is images wrapped in PDF structure. Text extraction returns empty, which surfaces as extraction_failed rather than an error, because the fetch worked fine.

On /crawl, pdf_ocr enables optical recognition for exactly this case:

resp = requests.post(
    f"{API}/crawl",
    headers=HEADERS,
    json={
        "url": "https://example.gov.br/transparencia/",
        "max_pages": 30,
        "url_regex": r".*\.pdf$",
        "parse_pdf": True,
        "pdf_ocr": True,
        "timeout_ms": 180_000,
    },
    timeout=30,
)
job_id = resp.json()["job_id"]

Two notes on this configuration:

url_regex restricted to .pdf turns the crawl into a document harvester. The frontier still discovers HTML links to traverse, but only PDFs are extracted.
timeout_ms raised to 180000. The default of 60000 is tuned for HTML. OCR over a multi-page scan will exceed it, and a timeout mid-crawl shows up as per-item timeout entries rather than a job failure.

OCR is slower and less accurate than a real text layer. Enable it when you know the corpus is scanned, not as a blanket default — on a digital-native PDF it adds latency for a worse result than plain extraction.

Detecting the empty-text case

def needs_ocr(item: dict) -> bool:
    """A PDF that fetched fine but yielded almost nothing is probably scanned."""
    if item["scrape_status"] == "extraction_failed":
        return True
    text = item.get("markdown") or ""
    return item["url"].lower().endswith(".pdf") and len(text.strip()) < 200


job = wait_for_job(job_id)
rescan = [item["url"] for item in job["results"] if needs_ocr(item)]
print(f"{len(rescan)} documents to reprocess with OCR")

The length threshold catches the worst case: a scanned PDF where extraction returned only the header text a text layer happened to contain, so scrape_status reads success while the body is missing. That silently poisons a RAG index with documents that look present and answer nothing.

Chunking extracted PDFs

PDF-derived Markdown has different structure from HTML-derived Markdown. Page breaks become paragraph breaks, and heading levels are inferred from font size rather than declared. Splitting on Markdown headers works less reliably than it does for web pages.

For PDFs, size-based splitting with overlap is more predictable:

from langchain_text_splitters import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=1200,
    chunk_overlap=150,
    separators=["\n\n", "\n", ". ", " "],
)
chunks = splitter.split_text(data["markdown"])

The overlap matters more here than for HTML, because a table or a numbered clause split across a page boundary loses its subject line otherwise.

Cost

PDF extraction is billed as a page scrape: 1 credit, same as HTML, regardless of page count inside the document. A 300-page technical standard costs the same single credit as a landing page. For document-heavy corpora that pricing shape is the reason to prefer a PDF-aware crawl over building a separate download-and-parse pipeline.

Giving an agent a web search tool that returns analyzed results

Roberto Francisco junior — Fri, 04 Sep 2026 16:40:37 +0000

An agent with a search tool that returns ten titles and ten snippets has to spend a second inference pass deciding which links are worth fetching, then a third fetching them. Each hop adds latency and a chance to pick wrong.

POST /search collapses that: multi-source retrieval plus content analysis in one asynchronous job. It returns 202 with a job identifier, so the tool wrapper needs to poll before it can answer.

The enqueue call

import os
import requests

API = "https://api.messora.dev"
HEADERS = {"X-API-Key": os.environ["MESSORA_API_KEY"]}


def start_search(query: str, num_results: int = 10) -> tuple[str, int]:
    resp = requests.post(
        f"{API}/search",
        headers=HEADERS,
        json={
            "query": query,
            "num_results": num_results,
            "query_fanout": True,
        },
        timeout=30,
    )
    resp.raise_for_status()          # 202
    body = resp.json()
    return body["job_id"], body["credits_estimated"]

credits_estimated comes back with the enqueue, before any work happens. Billing is 1 credit per result, so a num_results of 10 reserves 10. Reading that number and logging it is how you keep an autonomous agent from quietly spending a month of balance in an afternoon.

Narrowing the search

Four filters matter more than the query wording:

json={
    "query": "postgres logical replication lag monitoring",
    "num_results": 8,
    "include_domains": ["postgresql.org", "aws.amazon.com"],
    "exclude_domains": ["pinterest.com", "quora.com"],
    "freshness": "month",
    "country": "BR",
}

include_domains is the strongest signal available. For technical queries, restricting to official documentation domains does more for answer quality than any amount of prompt engineering downstream, because it removes the aggregator pages that repeat each other.

query_fanout defaults to True and expands one query into several related ones before retrieval. Turn it off when you need the literal query and nothing adjacent — checking whether a specific error string appears anywhere, for example.

Wrapping it as a tool

import time

TERMINAL = {"SUCCESS", "FAILURE", "REVOKED"}


def web_search(query: str, num_results: int = 8, timeout_s: int = 180) -> list[dict]:
    """Tool function: returns analyzed search results or raises."""
    job_id, estimated = start_search(query, num_results)

    deadline = time.monotonic() + timeout_s
    delay = 2.0
    while time.monotonic() < deadline:
        resp = requests.get(f"{API}/jobs/{job_id}", headers=HEADERS, timeout=30)
        if resp.status_code == 429:
            time.sleep(delay)
            delay = min(delay * 2, 20.0)
            continue
        resp.raise_for_status()
        job = resp.json()
        if job["status"] in TERMINAL:
            if job["status"] != "SUCCESS":
                raise RuntimeError(job.get("error") or job["status"])
            return job["results"] or []
        time.sleep(delay)
        delay = min(delay * 1.5, 10.0)

    raise TimeoutError(f"search job {job_id} exceeded {timeout_s}s")

The tool docstring an agent reads should state the credit cost. A model that knows a call costs 8 credits behaves differently from one that thinks search is free — it batches questions instead of firing one per thought.

Rate limit shapes the orchestration

/search allows 2 enqueues per minute per account, the tightest limit on the public API. That is deliberate: each call fans out to multiple sources and runs analysis, so it is the most expensive operation available.

For an agent loop this means search cannot be the default action. Two patterns that survive the limit:

Search once, scrape many. One /search to find the right five domains, then /batch at 1 credit per page to read them.
Cache by normalized query. Lowercase, strip punctuation, sort tokens, and key a local dict on the result. Agents re-ask the same question with different phrasing constantly.

Budget guard worth adding

class SearchBudget:
    def __init__(self, max_credits: int):
        self.remaining = max_credits

    def spend(self, credits: int) -> None:
        if credits > self.remaining:
            raise RuntimeError(
                f"search budget exhausted: needs {credits}, has {self.remaining}"
            )
        self.remaining -= credits

Call spend(credits_estimated) right after the enqueue returns. A 402 from the API means the entire account balance is gone, which is a much worse place to discover the problem than a per-session ceiling you set yourself.

What the results carry

Job results arrive under results once the status reaches SUCCESS, with per-item content already analyzed rather than raw snippets. Feed them to the model directly; the second retrieval pass most agent frameworks add after a search call is unnecessary here, and each skipped pass is one less place to lose the thread.

Scraping 50 URLs in one job instead of 50 sequential requests

Roberto Francisco junior — Thu, 03 Sep 2026 16:47:43 +0000

A loop over 50 URLs calling a synchronous scrape endpoint has a wall-clock cost equal to the sum of every page's latency. With an average of 4 seconds per page, that is 200 seconds of your process sitting in requests.post, plus a rate limit you will hit around request ten.

POST /batch moves the fan-out server-side. You submit a list, get a job identifier back, and poll once.

Submit the list

import os
import requests

API = "https://api.messora.dev"
HEADERS = {"X-API-Key": os.environ["MESSORA_API_KEY"]}

urls = [
    "https://example.com/blog/post-1",
    "https://example.com/blog/post-2",
    # ... up to 50
]

resp = requests.post(
    f"{API}/batch",
    headers=HEADERS,
    json={"urls": urls, "formats": ["markdown"]},
    timeout=30,
)
resp.raise_for_status()             # 202
job_id = resp.json()["job_id"]

The cap is 50 URLs per request. A list of 51 returns 422, not a truncated job — the validation is strict so you never silently lose the tail of your input.

Chunking a larger workload

from itertools import islice

BATCH_MAX = 50


def chunked(iterable, size):
    it = iter(iterable)
    while chunk := list(islice(it, size)):
        yield chunk


def submit_all(all_urls: list[str]) -> list[str]:
    job_ids = []
    for chunk in chunked(all_urls, BATCH_MAX):
        resp = requests.post(
            f"{API}/batch",
            headers=HEADERS,
            json={"urls": chunk, "formats": ["markdown"]},
            timeout=30,
        )
        resp.raise_for_status()
        job_ids.append(resp.json()["job_id"])
    return job_ids

Deduplicate before chunking. Every duplicate URL is a charged page, and a list built by concatenating sitemap fragments routinely contains the same canonical URL three times under different query strings.

Rate limits scale with plan

/batch is limited per account by plan, unlike /scrape which is a flat 10 per minute:

Plan	Requests per minute
free	10
starter	60
growth	120
scale	300

The limit counts enqueues, not URLs. On the free plan, 10 batches per minute is 500 URLs per minute of submitted work — the constraint is almost never the enqueue rate, it is the credit balance.

Collecting results

job = wait_for_job(job_id)          # same polling loop as /crawl

ok, failed = [], []
for item in job["results"]:
    (ok if item["scrape_status"] == "success" else failed).append(item)

print(f"{len(ok)} succeeded, {len(failed)} failed, {job['credits_used']} credits")

for item in failed:
    print(item["url"], item["scrape_status"], item.get("error"))

results preserves one entry per submitted URL, each with its own scrape_status. Batch jobs report SUCCESS when the job ran, not when every page worked — a batch where 40 of 50 pages were blocked is a successful job with 10 credits charged.

Partial failure is the normal case

At 50 URLs from a mixed set of domains, expect a handful of blocked_antibot and timeout entries. Design around it:

def retry_failures(job: dict) -> list[str]:
    """URLs worth a second attempt. Extraction failures are not."""
    retriable = {"blocked_antibot", "timeout"}
    return [
        item["url"]
        for item in job["results"]
        if item["scrape_status"] in retriable
    ]

extraction_failed is not retriable — the fetch already worked, so a second identical request produces the same empty conversion. Those URLs need formats: ["raw"] and a manual parse, or they need dropping.

Choosing between batch and crawl

Both return a job_id and both poll through GET /jobs/{job_id}, but they answer different questions:

/batch — you already know the URLs. Sitemap entries, a database column, a list of competitor product pages.
/crawl — you know one URL and want the reachable set from it, filtered by url_regex.

Running a crawl when you have an explicit list wastes discovery work and gives you less control over which pages are visited. Running a batch when you do not have the list means writing your own link discovery, which is the part /crawl already does with a page cap and a frontier filter.

Polling an asynchronous crawl job without hammering the API

Roberto Francisco junior — Wed, 02 Sep 2026 16:56:04 +0000

A breadth-first crawl over 50 pages does not fit inside an HTTP request. The MESSORA /crawl endpoint returns 202 Accepted with a job identifier and runs the work on a queue, which means your client needs a polling loop that terminates.

Most polling loops written in a hurry share the same three bugs: they treat an unknown state as terminal, they poll at a fixed interval that trips rate limits, and they never time out. Here is one that handles all three.

Enqueue

import os
import time
import requests

API = "https://api.messora.dev"
HEADERS = {"X-API-Key": os.environ["MESSORA_API_KEY"]}


def start_crawl(seed: str, max_pages: int = 25) -> str:
    resp = requests.post(
        f"{API}/crawl",
        headers=HEADERS,
        json={
            "url": seed,
            "max_pages": max_pages,
            "only_main_content": True,
            "url_regex": r"/docs/.*",
        },
        timeout=30,
    )
    resp.raise_for_status()          # 202 on success
    return resp.json()["job_id"]

url and max_pages are both required; there is no implicit page budget. max_pages caps at 50. url_regex filters discovered links before they enter the frontier, which is the difference between crawling a documentation tree and crawling an entire marketing site.

The four job states

GET /jobs/{job_id} reports Celery task states, in uppercase:

State	Terminal	Meaning
`PENDING`	no	Queued, no worker has picked it up
`STARTED`	no	Worker is crawling
`SUCCESS`	yes	`results` is populated
`FAILURE`	yes	`error` carries a sanitized message
`REVOKED`	yes	Cancelled; no partial results exposed

PENDING is ambiguous in Celery generally — it also covers task IDs that never existed. Here it does not, because the job row is looked up per tenant before the queue is consulted. A 404 means the job belongs to a different account or was never created; it never means "still starting".

A polling loop that terminates

TERMINAL = {"SUCCESS", "FAILURE", "REVOKED"}


def wait_for_job(job_id: str, timeout_s: int = 600) -> dict:
    deadline = time.monotonic() + timeout_s
    delay = 2.0

    while time.monotonic() < deadline:
        resp = requests.get(f"{API}/jobs/{job_id}", headers=HEADERS, timeout=30)

        if resp.status_code == 429:
            time.sleep(delay)
            delay = min(delay * 2, 30.0)
            continue

        resp.raise_for_status()
        job = resp.json()

        if job["status"] in TERMINAL:
            return job

        time.sleep(delay)
        delay = min(delay * 1.5, 15.0)

    raise TimeoutError(f"job {job_id} still {job['status']} after {timeout_s}s")

Three decisions worth naming:

Exponential backoff with a ceiling. A crawl of 50 pages takes minutes, not seconds. Polling every 500 ms produces hundreds of pointless requests and eventually a 429.
429 does not consume the retry budget. It is a signal to slow down, not a job failure.
time.monotonic(), not time.time(). A clock adjustment mid-crawl should not extend or truncate your timeout.

Reading the results

job = wait_for_job(start_crawl("https://example.com/docs/"))

if job["status"] == "FAILURE":
    raise RuntimeError(job["error"])

print(f"pages crawled: {job['pages_crawled']}")
print(f"stopped because: {job['stopped_reason']}")
print(f"credits: {job['credits_used']}")

for item in job["results"]:
    if item["scrape_status"] == "success":
        print(item["url"], len(item["markdown"]), "chars")
    else:
        print(item["url"], "skipped:", item.get("reason"))

stopped_reason is the field that answers "why did I get 12 pages instead of 25". Canonical values are completed, timeout, token_cap, and discovery_failed. A crawl that returns fewer pages than requested with stopped_reason: "completed" simply ran out of links matching your regex — the crawl worked, the frontier was empty.

Per-item statuses matter

results is a list where each entry carries its own scrape_status. A crawl is rarely all-or-nothing: a documentation tree with two PDF downloads and one page behind bot protection returns SUCCESS at the job level with three non-success items inside.

Credits are charged per successfully crawled page, so a job that discovers 25 URLs and extracts 22 costs 22 credits. Counting len(results) overstates the spend and understates your failure rate.

Rate limits in practice

/crawl allows 10 enqueues per minute per account. /jobs/{job_id} is bounded separately as anti-polling protection. If you are orchestrating dozens of crawls, queue the enqueues client-side rather than firing them in a loop and catching 429 — the backoff you write is always cheaper than the one the server imposes.