wellmarked

Official Python SDK for the WellMarked API — convert any URL to clean Markdown.

pip install wellmarked

Quick start

from wellmarked import WellMarked

with WellMarked(api_key="wm_...") as wm:
    result = wm.extract("https://example.com/article")
    print(result.markdown)
    print(result.metadata.title, "by", result.metadata.author)
    print("retrieved at", result.metadata.retrieved_at)

result.metadata.retrieved_at is a datetime (UTC) recording when WellMarked actually fetched the page — distinct from result.metadata.date (the article's published date, often None). Useful for cache-freshness checks on the caller's side.

The API key can also be picked up from the WELLMARKED_API_KEY environment variable, in which case WellMarked() is enough.

Get a key at wellmarked.io.

Pricing

See additional pricing information at wellmarked.io/#pricing.

Async

AsyncWellMarked is a drop-in async equivalent — every endpoint method is a coroutine.

import asyncio
from wellmarked import AsyncWellMarked

async def main():
    async with AsyncWellMarked() as wm:
        result = await wm.extract("https://example.com/article")
        print(result.markdown)

asyncio.run(main())

Bulk extraction

Submit many URLs at once (Pro: up to 50; Growth: up to 200; Enterprise: unlimited). The call returns immediately with a job_id. Poll with get_job or block until done with wait_for_job.

job = wm.bulk([
    "https://example.com/article-1",
    "https://example.com/article-2",
])
job = wm.wait_for_job(job.job_id)         # blocks until status == "done"

for item in job.results:
    if item.ok:
        print(item.metadata.title)
    else:
        print(f"{item.url} failed: {item.error}")

Pass retry=N to any of extract, bulk, or crawl to re-attempt timed-out fetches server-side (target_timeout only, fresh connection per attempt, default 0). On the synchronous extract each timed-out attempt takes 20–30s before the next fires, so aggressive values belong here on bulk, where the async workers absorb the wait. search doesn't take retry — its 15-second per-result deadline can't absorb one.

get_job and wait_for_job are polymorphic — they work for both bulk and crawl job_ids. The SDK reads a kind discriminator from the API response and returns either a BulkJob or a CrawlJob. Use isinstance(job, CrawlJob) (or check job.kind == "crawl") before reading crawl-specific fields like job.truncated or item.depth.

Crawl

Crawl a site BFS-style from a root URL — same-site links only, with per-plan depth and page caps (Pro: depth 5, up to 2,000 pages; Growth: depth 10, up to 10,000 pages; Enterprise: unlimited). Like bulk, this returns a queued job; poll with get_job or block until done with wait_for_job — the same two functions work on both kinds. Pass render_js=True to fetch each page through Playwright instead of httpx; a single shared browser is launched at the start of the crawl and reused across pages.

job = wm.crawl("https://docs.example.com", depth=2)
job = wm.wait_for_job(job.job_id)          # works for crawl AND bulk job ids

for page in job.results:
    if page.ok:
        print(f"depth={page.depth} {page.metadata.title}")
    else:
        print(f"{page.url} failed: {page.error}")

if job.truncated:
    print(f"crawl stopped early: {job.truncated_reason}")

Pass max_pages=N to stop the crawl after N successful pages — it can only narrow your plan's page cap, never widen it. Each successful page consumes one request from your monthly quota — failed pages (timeouts, robots-disallowed, no-content) are not billed. If you run out of quota mid-crawl the job finishes with truncated=True, truncated_reason="quota_exhausted".

Webhooks

Instead of polling wait_for_job, pass a webhook_url to bulk() or crawl() and we'll POST a signed notification to that URL the moment the job reaches status="done".

job = wm.bulk(
    ["https://example.com/article-1", "https://example.com/article-2"],
    webhook_url="https://yourapp.com/hooks/wm",
)

# First time you ever submit with a webhook_url, the response carries
# your signing secret — store it now, you only see it once.
if job.webhook_signing_secret is not None:
    save_to_env("WELLMARKED_WEBHOOK_SECRET", job.webhook_signing_secret)

webhook_signing_secret is populated only on the submission that first minted it (one-time visibility, Stripe-style). Subsequent submissions return None for that field. Lost it? Call wm.rotate_webhook_secret():

rotated = wm.rotate_webhook_secret()
print("New secret:", rotated.webhook_signing_secret)  # save it

Rotation invalidates the previous secret immediately. Deliveries already queued for retry are re-signed with the new secret on their next attempt.

Receiving + verifying a delivery

The SDK ships a verifier — use it rather than reimplementing HMAC by hand:

from fastapi import FastAPI, Request, Response
from wellmarked import verify_webhook, WebhookVerificationError

app = FastAPI()
WELLMARKED_WEBHOOK_SECRET = os.environ["WELLMARKED_WEBHOOK_SECRET"]

@app.post("/hooks/wm")
async def wellmarked_hook(request: Request):
    try:
        payload = verify_webhook(
            secret=WELLMARKED_WEBHOOK_SECRET,
            headers=request.headers,
            body=await request.body(),    # MUST be raw bytes, not request.json()
        )
    except WebhookVerificationError:
        return Response(status_code=401)

    job_id = payload["job_id"]
    # Default payload is "thin": metadata only + results_url. Fetch
    # results with the same SDK against your normal API key.
    job = wm.get_job(job_id)
    for item in job.results:
        ...
    return Response(status_code=200)

Pass webhook_include_results=True on submission to inline the full results array (capped at ~5 MB; over the cap the payload silently falls back to the thin shape with results_truncated_for_size=True).

Delivery semantics

• Your endpoint must respond with a 2xx within 10 seconds.
• Retries on any other outcome (timeout, 4xx, 5xx, DNS): 30s, 5m, 30m, 2h, 12h, 24h — 7 attempts, ~38 hours, then dead-letter.
• X-WellMarked-Delivery-Id is stable across retries — use it as your idempotency key. X-WellMarked-Timestamp and X-WellMarked-Signature are recomputed every attempt.
• Treat delivery as at-least-once.

See the Webhooks documentation for the full signature scheme and header reference.

Custom headers

Pass extra HTTP headers on every request — useful for correlation IDs, multi-tenant identifiers, or a custom user-agent suffix:

with WellMarked(
    api_key="wm_...",
    headers={"X-Trace-Id": "req-abc-123", "X-Tenant": "acme"},
) as wm:
    wm.extract("https://example.com")

Headers can also be added or removed at runtime:

wm.set_header("X-Run-Id", "run-99")
wm.extract(...)                  # carries X-Run-Id
wm.remove_header("X-Run-Id")

Authorization, Content-Type, and Accept are reserved — the SDK manages them itself, and entries passed in headers= for those keys are silently ignored. To rotate the bearer token, use rotate_key().

Usage & rate limits

get_usage() is the source of truth for your current-period quota. The quota state belongs on the account, so call get_usage() when you want it:

usage = wm.get_usage()
print(f"{usage.used} / {usage.limit} used this period ({usage.plan}) — {usage.remaining} left")

GET /usage itself does not count toward your quota.

Key rotation

rotated = wm.rotate_key()
print("New key:", rotated.api_key)  # shown once — store it before the program exits

After rotate_key() the client automatically switches to the new key for subsequent calls; you still need to persist rotated.api_key somewhere durable, because the previous key stops working immediately and there is no recovery flow.

Errors

Every non-2xx response is translated into a typed exception. Catch the base class to handle anything, or the specific subclass to handle one failure mode:

from wellmarked import (
    WellMarked,
    AuthenticationError,
    PermissionDeniedError,
    NotFoundError,
    UnprocessableEntityError,
    RateLimitError,
    APIConnectionError,
)

with WellMarked() as wm:
    try:
        result = wm.extract("https://example.com/paywalled")
    except RateLimitError as e:
        print(f"Quota hit. Resets in {e.retry_after}s.")
    except UnprocessableEntityError as e:
        # e.code is one of: no_content, target_timeout, ...
        print(f"Extraction failed ({e.code}): {e.message}")

All inherit from WellMarkedError.

Configuration

WellMarked(
    api_key="wm_...",                   # or set WELLMARKED_API_KEY
    timeout=30.0,                       # seconds, per request
    max_retries=2,                      # retries for safely replayable requests
    headers={"X-Trace-Id": "..."},      # optional: extra headers on every request
)

The client always talks to https://api.wellmarked.io.

For Agents

If you are an agent, feel free to find additional context here!

License

Copyright © 2026 WellMarked. Released under the MIT License.

Source: https://github.com/WellMarkedAPI/Python-SDK

Use of the hosted API at api.wellmarked.io remains subject to the Terms of Service.