Python SDK

pip install reader-py

import os
from reader_py import ReaderClient

reader = ReaderClient(api_key=os.environ["READER_KEY"])

result = reader.read(url="https://example.com")
if result.kind == "scrape":
    print(result.data.markdown)
    print(f"scraped in {result.data.metadata.duration}ms")

import asyncio
from reader_py import AsyncReaderClient

async def main():
    async with AsyncReaderClient(api_key=os.environ["READER_KEY"]) as reader:
        result = await reader.read(url="https://example.com")
        if result.kind == "scrape":
            print(result.data.markdown)

asyncio.run(main())

reader = ReaderClient(
    api_key="rdr_your_key",              # required
    base_url="https://api.reader.dev",   # optional, override for self-hosted
    timeout=60,                          # per-request timeout in seconds (default 60)
)

result = reader.read(
    url="https://example.com",
    formats=["markdown"],
    only_main_content=True,
)

if result.kind == "scrape":
    print(result.data.url)                         # canonical URL after redirects
    print(result.data.markdown)                    # clean markdown
    print(result.data.metadata.title)              # page title
    print(result.data.metadata.status_code)        # 200
    print(result.data.metadata.duration)           # ms
    print(result.data.metadata.cached)             # True if served from cache
    print(result.data.metadata.proxy_mode)         # "standard" | "premium"

import base64

result = reader.read(
    url="https://example.com",
    formats=["markdown", "screenshot"],
)

if result.kind == "scrape" and result.data.screenshot:
    # screenshot is a base64-encoded PNG
    png_bytes = base64.b64decode(result.data.screenshot)
    with open("screenshot.png", "wb") as f:
        f.write(png_bytes)

result = reader.read(
    urls=[
        "https://example.com/page-1",
        "https://example.com/page-2",
        "https://example.com/page-3",
    ],
)

if result.kind == "job":
    print(f"completed {result.data.completed} / {result.data.total}")
    for page in result.data.results:
        if page.error:
            print(f"{page.url}: {page.error}")
        else:
            print(page.url, len(page.markdown or ""), "chars")

result = reader.read(
    url="https://docs.example.com",
    max_depth=3,
    max_pages=100,
)

# Default: standard. Fast and affordable, works for most sites
reader.read(url=url)

# Explicit standard: force the cheaper tier (error if blocked)
reader.read(url=url, proxy_mode="standard")

# Explicit premium: residential proxies, bypasses bot walls (3x credits)
reader.read(url=url, proxy_mode="premium")

# Fetch a single page of a job's results
job, has_more = reader.get_job(job_id, skip=0, limit=20)

# Collect every page automatically
all_pages = reader.get_all_job_results(job_id)

# Poll a job by ID until it terminates (collects all results on completion)
job = reader.wait_for_job(job_id, poll_interval=2, timeout=300)

# Cancel a queued or processing job
reader.cancel_job(job_id)

# Retry failed URLs in a completed job
retry_info = reader.retry_job(job_id)
print(f"retrying {retry_info['retrying']} failed URLs")

for event in reader.stream(job_id):
    if event.type == "progress":
        print(f"{event.completed} / {event.total}")
    elif event.type == "page":
        print("page done:", event.data.url)
    elif event.type == "error":
        print("page failed:", event.url, event.error)
    elif event.type == "done":
        print("job finished:", event.status)
        break

credits = reader.get_credits()
print(f"{credits.balance} / {credits.limit}, tier: {credits.tier}")
print(f"resets at: {credits.reset_at}")

if credits.balance < 100:
    # Warn, pause workers, upgrade tier, etc.
    pass

from reader_py import (
    ReaderApiError,
    InvalidRequestError,
    UnauthenticatedError,
    InsufficientCreditsError,
    UrlBlockedError,
    NotFoundError,
    ConflictError,
    RateLimitedError,
    ConcurrencyLimitedError,
    InternalServerError,
    UpstreamUnavailableError,
    ScrapeTimeoutError,
)

try:
    result = reader.read(url=url)
except InsufficientCreditsError as err:
    print(f"Need {err.required} credits, have {err.available}")
    print(f"Resets at {err.reset_at}")
except RateLimitedError as err:
    print(f"Rate limited. Retry after {err.retry_after_seconds}s")
except UrlBlockedError as err:
    print(f"URL blocked: {err.reason}")
except ScrapeTimeoutError as err:
    print(f"Scrape exceeded {err.timeout_ms}ms")
except ReaderApiError as err:
    # Catch-all for known API errors
    print(f"[{err.code}] {err}")
    print(f"Docs: {err.docs_url}")
    print(f"Request ID: {err.request_id}")

reader.read(
    urls=many_urls,
    webhook={
        "url": "https://your-app.example.com/hooks/reader",
        "events": ["job.completed", "job.failed"],
        "secret": os.environ["READER_WEBHOOK_SECRET"],
    },
)

from reader_py import (
    ReadParams,
    ReadResult,      # kind: "scrape" | "job", data: ScrapeResult | Job
    ScrapeResult,
    ScrapeMetadata,
    Job,
    Page,
    Credits,
    UsageEntry,
    Pagination,
)

from reader_py import ReaderClient
from playwright.sync_api import sync_playwright

reader = ReaderClient(api_key="rdr_...")

session = reader.sessions.create()
print(session.ws_endpoint)

with sync_playwright() as p:
    browser = p.chromium.connect_over_cdp(session.ws_endpoint)
    page = browser.contexts[0].new_page()
    page.goto("https://example.com")
    print(page.title())
    browser.close()

reader.sessions.stop(session.session_id)

from reader_py import AsyncReaderClient
from playwright.async_api import async_playwright

reader = AsyncReaderClient(api_key="rdr_...")

session = await reader.sessions.create()

async with async_playwright() as p:
    browser = await p.chromium.connect_over_cdp(session.ws_endpoint)
    page = await (await browser.new_context()).new_page()
    await page.goto("https://example.com")
    print(await page.title())
    await browser.close()

await reader.sessions.stop(session.session_id)