DEV Community: Roberto Francisco junior

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.

Extracting typed JSON from any web page with a Pydantic schema

Roberto Francisco junior — Wed, 02 Sep 2026 13:21:41 +0000

Parsing product pages with CSS selectors breaks the first time a designer renames a class. The selector-based approach encodes the page's current DOM shape into your code, and that shape is not part of any contract the site owner agreed to.

Schema-driven extraction inverts the dependency: you declare the fields you want, and the extraction layer figures out where they live. Here is how to do it with a Pydantic model, and what it costs.

Declare the shape you want

import json
import os
import requests
from pydantic import BaseModel, Field

API = "https://api.messora.dev"


class Product(BaseModel):
    name: str = Field(description="Product display name")
    price_brl: float = Field(description="Price in BRL, numeric only")
    in_stock: bool = Field(description="True when purchasable right now")
    sku: str | None = Field(default=None, description="Manufacturer SKU if shown")


def extract(url: str, model: type[BaseModel]) -> dict:
    resp = requests.post(
        f"{API}/scrape",
        headers={"X-API-Key": os.environ["MESSORA_API_KEY"]},
        json={
            "url": url,
            "formats": ["json"],
            "json_schema": model.model_json_schema(),
            "only_main_content": True,
        },
        timeout=120,
    )
    resp.raise_for_status()
    data = resp.json()

    if data["scrape_status"] != "success":
        raise RuntimeError(f"{url} -> {data['scrape_status']}")

    if data.get("json_extraction_error_code"):
        raise ValueError(f"extraction rejected: {data['json_extraction_error_code']}")

    return data["json"]


payload = extract("https://example.com/produto/teclado-mecanico", Product)
print(json.dumps(payload, indent=2, ensure_ascii=False))

model_json_schema() from Pydantic emits standard JSON Schema, which is exactly what the json_schema field expects. The description on each field is not decoration — it is the disambiguation signal when a page shows three numbers that could all be prices.

Validate on the way back in

The response is a dict, not a model instance. Round-trip it through Pydantic so a missing or mistyped field fails loudly at the boundary instead of three layers deeper:

product = Product.model_validate(payload)
assert product.price_brl > 0

This is the step most integrations skip. Without it, a null price silently becomes 0.0 somewhere downstream and your pricing dashboard reports a free keyboard.

Cost model

Structured extraction is billed at a flat 10 credits per request, regardless of page size. Markdown and raw scraping cost 1 credit per page. The gap reflects the inference pass that maps page content onto your schema.

Two consequences for how you design jobs:

Do not request json when you only need text. Ten credits for a page you were going to embed anyway is a 10x overspend.
Batch fields, not requests. One schema with twelve fields costs 10 credits. Twelve single-field requests cost 120 credits for the same page.

When there is no schema yet

json_prompt accepts a natural-language description instead of a schema. It is the right tool while you are still exploring what a page even contains:

json={
    "url": url,
    "formats": ["json"],
    "json_prompt": "Return the author name, publication date in ISO 8601, and a list of cited URLs.",
}

The tradeoff is that the output shape is not guaranteed across calls. Use json_prompt to discover the shape, then freeze it into a Pydantic model for anything that runs on a schedule.

Failure signal to watch

json_extraction_error_code is populated when the fetch succeeded but the mapping did not. That is a different problem from scrape_status and needs different handling: a blocked_antibot means try another URL, while an extraction error usually means your schema asks for a field the page does not have.

Checking only scrape_status and then reading data["json"] gives you None with no explanation. Check both.

Practical field design

Fields that work well are the ones a human could point at on the rendered page. Fields that fail are the ones requiring inference across pages — "is this cheaper than competitors" is not on the page, so it does not belong in the schema.

Keep Optional on anything that is genuinely absent on some pages. A required field that is missing forces the extractor to invent a value, and an invented SKU is worse than a null one.

Turning a web page into LLM-ready Markdown with one HTTP call

Roberto Francisco junior — Tue, 01 Sep 2026 20:31:07 +0000

Feeding raw HTML to a language model wastes tokens on markup that carries no meaning. A typical documentation page is 180 KB of HTML and roughly 6 KB of actual prose. The rest is <script>, inline CSS, navigation, cookie banners, and analytics beacons that inflate your context window and dilute retrieval quality.

The fix is to convert the page to Markdown server-side and send only the content. Here is the exact request against the MESSORA API, including the failure modes you have to handle in production.

The request

Authentication uses the X-API-Key header. There is no Authorization: Bearer variant — a bearer token returns 401.

import os
import requests

API = "https://api.messora.dev"


def to_markdown(url: str) -> str:
    resp = requests.post(
        f"{API}/scrape",
        headers={"X-API-Key": os.environ["MESSORA_API_KEY"]},
        json={
            "url": url,
            "formats": ["markdown"],
            "only_main_content": True,
        },
        timeout=90,
    )
    resp.raise_for_status()
    data = resp.json()

    if data["scrape_status"] != "success":
        raise RuntimeError(f"{url} -> {data['scrape_status']}: {data.get('error')}")

    return data["markdown"]


print(to_markdown("https://docs.python.org/3/library/asyncio-task.html")[:400])

only_main_content drops header, footer, nav, and sidebar nodes before conversion. Leave it off when you actually need the chrome — comparing navigation structures across competitor sites, for example.

What comes back

{
  "success": true,
  "scrape_status": "success",
  "markdown": "# asyncio — Task object\n...",
  "rawHtml": null,
  "json": null,
  "metadata": {
    "url": "https://docs.python.org/3/library/asyncio-task.html",
    "title": "Coroutines and Tasks",
    "statusCode": 200,
    "fetchedAt": "2026-09-02T13:04:11.882Z"
  },
  "credits_used": 1,
  "remaining_credits": 987
}

formats accepts markdown, raw, and json. Asking for raw alongside markdown populates rawHtml in the same response, which is useful when you want to keep the original for auditing and the Markdown for embedding.

The four statuses you must branch on

scrape_status is an enum with exactly four values. Treating a 200 OK as a successful scrape is the most common integration bug, because the HTTP layer succeeded while the fetch did not.

`scrape_status`	Meaning	Sensible reaction
`success`	Content extracted, 1 credit charged	Use `markdown`
`blocked_antibot`	Target refused the fetch	Retry later or drop the URL
`timeout`	Page exceeded the `timeout` budget	Raise `timeout`, or set `render_js: false`
`extraction_failed`	Fetch worked, conversion produced nothing usable	Fall back to `raw` and parse manually

Credits are only debited on success, so a blocked page costs nothing. That matters when you are crawling a domain of unknown difficulty and want to size the job before committing budget.

HTTP errors worth handling separately

STATUS_ACTIONS = {
    401: "X-API-Key header missing",
    402: "monthly credit balance exhausted",
    403: "API key invalid or revoked",
    422: "malformed payload or SSRF-blocked URL",
    429: "rate limit exceeded (10 req/min on /scrape)",
}


def scrape_with_diagnostics(url: str) -> dict:
    resp = requests.post(
        f"{API}/scrape",
        headers={"X-API-Key": os.environ["MESSORA_API_KEY"]},
        json={"url": url, "formats": ["markdown"], "only_main_content": True},
        timeout=90,
    )
    if resp.status_code in STATUS_ACTIONS:
        raise RuntimeError(f"{resp.status_code}: {STATUS_ACTIONS[resp.status_code]}")
    resp.raise_for_status()
    return resp.json()

The distinction between 402 and 429 drives very different retry logic. A 429 clears on its own within the minute, so backing off works. A 402 means the monthly balance is gone and retrying just burns requests — that one needs to page a human or switch plans.

Caching with max_age_ms

Re-scraping a page that has not changed is wasted credit. max_age_ms returns a cached copy when one exists within the window:

json={"url": url, "formats": ["markdown"], "max_age_ms": 86_400_000}

A cache hit still reports scrape_status: "success" and returns the stored Markdown. Use a wide window for reference documentation and a narrow one for pricing pages or anything with a timestamp your users will notice.

Token math

For a RAG index over 5,000 documentation pages, the difference between raw HTML and cleaned Markdown at roughly 4 characters per token is about 220 million tokens versus 7.5 million. That gap shows up twice: once at embedding time, and again on every retrieval that stuffs a chunk into the prompt.

Cleaning at fetch time is cheaper than cleaning at inference time, and it happens once instead of on every query.

Custom Document Loader for LangChain and LangGraph with Markdown extraction

Roberto Francisco junior — Thu, 27 Aug 2026 22:42:30 +0000

LangChain's default WebBaseLoader uses BeautifulSoup under the hood. While simple for static HTML pages, it breaks on two common production scenarios:

Bot Protection: Cloudflare or DataDome blocks standard Python requests.
Context Pollution: It preserves raw navigation links, scripts, and styling attributes in the Document page_content.

Here is a custom LangChain Document Loader implementation backed by the MESSORA extraction API.

Implementation

import os
import requests
from typing import List, Optional, Iterator
from langchain_core.document_loaders import BaseLoader
from langchain_core.documents import Document

class MessoraWebLoader(BaseLoader):
    """LangChain Document Loader that returns clean Markdown with anti-bot bypass."""

    def __init__(self, urls: List[str], api_key: Optional[str] = None, only_main_content: bool = True):
        self.urls = urls
        self.api_key = api_key or os.environ.get("MESSORA_API_KEY")
        self.only_main_content = only_main_content
        if not self.api_key:
            raise ValueError("MESSORA_API_KEY is required.")

    def lazy_load(self) -> Iterator[Document]:
        headers = {"X-API-Key": self.api_key}
        for url in self.urls:
            try:
                resp = requests.post(
                    "https://api.messora.dev/scrape",
                    headers=headers,
                    json={"url": url, "only_main_content": self.only_main_content},
                    timeout=30,
                )
                resp.raise_for_status()
                data = resp.json()
                markdown_text = data.get("markdown", "")
                metadata = {
                    "source": url,
                    "title": data.get("metadata", {}).get("title", ""),
                    "status_code": data.get("metadata", {}).get("statusCode", 200),
                }
                yield Document(page_content=markdown_text, metadata=metadata)
            except Exception as err:
                yield Document(page_content=f"Error loading {url}: {err}", metadata={"source": url, "error": True})

Using in a RAG Vector Pipeline

from langchain_text_splitters import MarkdownHeaderTextSplitter
from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings

# 1. Load clean documents
loader = MessoraWebLoader(urls=[
    "https://en.wikipedia.org/wiki/Retrieval-augmented_generation",
    "https://docs.python.org/3/library/json.html",
])
docs = loader.load()

# 2. Split on markdown headers
headers_to_split_on = [("#", "Header 1"), ("##", "Header 2"), ("###", "Header 3")]
markdown_splitter = MarkdownHeaderTextSplitter(headers_to_split_on=headers_to_split_on)

split_docs = []
for doc in docs:
    splits = markdown_splitter.split_text(doc.page_content)
    for s in splits:
        s.metadata.update(doc.metadata)
    split_docs.extend(splits)

print(f"Generated {len(split_docs)} semantic markdown chunks for embedding.")

Why standard HTTP clients get blocked by anti-bot systems and how TLS fingerprinting works

Roberto Francisco junior — Wed, 26 Aug 2026 14:36:49 +0000

If you have ever written a web scraper using Python requests, urllib3, or aiohttp, you have likely encountered this:

import requests
# Headers look identical to Chrome...
headers = {"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7)..."}
response = requests.get("https://protected-site.com", headers=headers)
print(response.status_code) # 403 Forbidden

Setting a browser User-Agent header is no longer sufficient against modern WAFs (Cloudflare, DataDome, Akamai). The block happens before your HTTP headers are even evaluated.

What is TLS Fingerprinting (JA3 / JA4)?

When a client initiates an HTTPS connection, it sends a ClientHello packet as part of the TLS handshake. This packet contains:

TLS Version: (e.g., TLS 1.3, TLS 1.2)
Cipher Suites: The cryptographic algorithms the client supports, in order of preference.
Extensions: Supported extensions (SNI, ALPN, Supported Versions, Key Share).
Elliptic Curves & Formats: Supported curves for key exchange.

Because different software stacks (OpenSSL in Python vs. BoringSSL in Chrome vs. Apple SecureTransport) build this ClientHello differently, each stack produces a distinct fingerprint.

Python requests (OpenSSL) has a known TLS signature.
Headless Chromium has a distinct signature from desktop Chrome.
Anti-bot systems hash these parameters into a JA4 string. If your User-Agent claims to be Chrome but your TLS handshake looks like OpenSSL, the request is dropped immediately.

How to solve TLS mismatch

1. Using TLS impersonation libraries (`curl_cffi`)

In Python, curl_cffi binds against a custom curl binary patched with BoringSSL to emulate exact browser handshakes:

from curl_cffi import requests

# Impersonates Chrome 120 TLS handshake and HTTP/2 settings
response = requests.get("https://protected-site.com", impersonate="chrome120")
print(response.status_code) # 200 OK

2. Offloading to an ingestion gateway

For production workloads involving IP rotation, residential proxies, and dynamic JavaScript rendering, managing TLS profiles and browser pools adds infrastructure cost.

Services like MESSORA manage fingerprint rotation, residential routing, and markdown parsing behind an API call:

curl -X POST https://api.messora.dev/scrape \
  -H "X-API-Key: YOUR_MESSORA_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"url": "https://protected-site.com", "only_main_content": true}'

Understanding where the inspection happens (Network layer vs. TLS layer vs. Application layer) is key to diagnosing why requests fail.

Building an automated web ingestion pipeline in n8n for LLM knowledge bases

Roberto Francisco junior — Wed, 26 Aug 2026 14:30:24 +0000

Automating weekly competitor intelligence, price tracking, or documentation updates inside n8n often fails when scrapers hit anti-bot challenges or return uncleaned HTML that exceeds context boundaries.

Here is an architectural pattern for an automated ingestion workflow in n8n using HTTP nodes and MESSORA.

Workflow Architecture

[Schedule Trigger (Daily at 08:00)] 
       │
       ▼
[Read URL List / Google Sheet]
       │
       ▼
[HTTP Request: POST /scrape (MESSORA)]
       │
       ▼
[Generate Embeddings (OpenAI)]
       │
       ▼
[Upsert to Vector Store (Qdrant / Pinecone)]

The n8n HTTP Node Configuration

In your n8n HTTP Request node:

Method: POST
URL: https://api.messora.dev/scrape
Authentication: Generic Credential Type -> Header Auth
- Header Name: X-API-Key
- Header Value: YOUR_MESSORA_API_KEY
Body Parameters:

   {
     "url": "={{ $json.url }}",
     "only_main_content": true
   }

Handling Output in Downstream Nodes

The response returns clean Markdown under data.markdown:

// Function node: prepare chunks for embedding
const markdown = $input.item.json.markdown;
const sourceUrl = $input.item.json.metadata.url;

return {
  json: {
    text: markdown,
    source: sourceUrl,
    extracted_at: new Date().toISOString(),
  }
};

Advantages over native browser nodes

Zero infrastructure management: No headless Chromium instances hanging on your n8n host.
Bypass WAFs automatically: Cloudflare, DataDome, and TLS challenges are resolved at the gateway.
75% token reduction: Ingests clean Markdown instead of full DOM hierarchies.

How to extract structured JSON from dynamic websites using typed Pydantic schemas

Roberto Francisco junior — Wed, 26 Aug 2026 14:25:02 +0000

When scraping e-commerce sites, job boards, or financial portals, converting unstructured HTML into valid, typed JSON usually requires fragile CSS selectors or multi-step LLM extraction prompts that hallucinate or miss required fields.

A more reliable pattern is schema-constrained extraction at the gateway level, where the extraction engine enforces a JSON Schema directly.

Defining the target schema in Pydantic

from pydantic import BaseModel, Field
from typing import List, Optional

class ProductItem(BaseModel):
    title: str = Field(description="Product name without promotional badges")
    price_cents: int = Field(description="Price in integer cents (e.g. 1999 for $19.99)")
    currency: str = Field(default="USD", description="Three-letter ISO currency code")
    in_stock: bool = Field(description="Stock availability status")
    features: List[str] = Field(default_factory=list, description="Bullet points of key product specifications")

class ProductCatalog(BaseModel):
    store_name: str
    products: List[ProductItem]

Extracting typed data with MESSORA

Instead of writing custom BeautifulSoup parsers or regexes for every site layout, pass the JSON Schema directly to the extraction endpoint:

import os
import json
import requests

api_key = os.environ.get("MESSORA_API_KEY")

payload = {
    "url": "https://example-store.com/electronics",
    "formats": ["json"],
    "json_schema": ProductCatalog.model_json_schema(),
}

response = requests.post(
    "https://api.messora.dev/scrape",
    headers={"X-API-Key": api_key},
    json=payload,
    timeout=45,
)
response.raise_for_status()

# Parse directly into Pydantic model
catalog_data = response.json().get("json")
catalog = ProductCatalog.model_validate(catalog_data)

for product in catalog.products:
    print(f"{product.title}: ${product.price_cents / 100:.2f} (In Stock: {product.in_stock})")

Why schema-first extraction matters

Deterministic types: Booleans, integers, and nested lists conform to the schema type contracts on the first pass.
Zero parsing maintenance: When sites change class names or DOM hierarchies, extraction continues working without breaking selectors.
No prompt engineering required: Field descriptions in the schema serve as extraction instructions.