markdownpipe 0.1.2

Markdown Pipe Python SDK (beta, pre-1.0): Python client for structured web scraping and fluent queries. APIs may change; initial PyPI release is intended as 0.1.0.

Markdown Pipe — Python SDK

Deterministic content extraction for AI pipelines, SEO, and data intelligence.

Discover, fetch, and extract clean, structured data from thousands of pages — with schema checksums, credit-based billing, and a fluent DSL.

---

Why Markdown Pipe?

Traditional scrapers return noisy HTML. Markdown Pipe returns structured, stable contracts:

• ✅ Global preset configurations — high-clean content pipeline for articles, products, SERP results out-of-the-box
• ✅ Fluent DSL (Q builder) — surgical field-by-field extraction with type-safe syntax
• ✅ Full discovery engine — sitemap, RSS, and link graph traversal in one call
• ✅ Decoupled pipeline — fetch once, reprocess many times without extra network cost
• ✅ Schema checksums — detect layout changes, enable server-side caching
• ✅ Transparent pricing — pay for infrastructure (credits), not opaque AI tokens

---

Installation

# Recommended
uv add markdownpipe

# pip
pip install markdownpipe

Python 3.10+

---

Quick Examples

1. Article Extraction (Preset)

import asyncio
from markdownpipe import MarkdownPipe

async def main():
    async with MarkdownPipe(api_key="sk-...") as client:
        article = await client.article("https://techcrunch.com/2024/01/15/ai-news/")
        print(article.title)
        print(article.author)
        print(article.markdown[:300])

asyncio.run(main())

2. Custom Schema Extraction

Define exactly what data you need using the Q builder:

from markdownpipe import MarkdownPipe, Q

PRODUCT_SCHEMA = {
    "name":        Q("h1").text(),
    "price":       Q(".price-current").text(),
    "images":      Q("img.product-photo").all().attr("src"),
    "description": Q(".product-desc").cleanup(".ads").markdown(),
    "specs":       Q("table.specs-table").table(),   # → structured JSON
}

async with MarkdownPipe(api_key="sk-...") as client:
    result = await client.extract(
        "https://shop.example.com/product/widget-pro",
        schema=PRODUCT_SCHEMA
    )
    print(result.data["name"])    # "Widget Pro"
    print(result.data["specs"])   # [{"Feature": "Weight", "Value": "120g"}, ...]
    print(f"{result.usage.credits} credits · {result.usage.latency}ms")

3. Domain-Scale Harvesting

Discover every article → extract all of them, concurrently:

async with MarkdownPipe(api_key="sk-...") as client:
    async for result in client.harvest(
        "https://blog.example.com",
        depth=1,
        concurrency=10,
        filter_func=lambda item: "/2024/" in item.url,
        show_progress=True,
    ):
        save_to_database(result.data)

4. Fetch → Markdown Pipeline

Decouple acquisition from processing — useful for async queues and storage pipelines:

async with MarkdownPipe(api_key="sk-...") as client:
    # Fetch and store (1 credit) — HTML stored server-side
    fetch_res = await client.fetch("https://docs.python.org/3/library/asyncio.html")

    # Retrieve the HTML (decompressed automatically)
    html = await client.get_stored_html(fetch_res.storage_url)

    # Convert to clean Markdown (0.5 credits)
    md_res = await client.markdown(html, url="https://docs.python.org/3/")
    print(md_res.markdown)

5. Raw DSL Engine

Run DSL on HTML you already have — no network call for fetching:

html = open("cached_report.html").read()

async with MarkdownPipe(api_key="sk-...") as client:
    result = await client.engine(
        url="https://example.com/report",
        html=html,
        schema={
            "title":      "h1::text",
            "financials": "table.financials::table",
            "summary":    ".exec-summary::markdown",
        }
    )
    print(result.data["financials"])

6. CSV → Markdown Table

async with MarkdownPipe(api_key="sk-...") as client:
    result = await client.csv("Name,Price\nWidget Pro,$49.99\nWidget Lite,$19.99")
    print(result.markdown)
    # | Name | Price |
    # |---|---|
    # | Widget Pro | $49.99 |
    # | Widget Lite | $19.99 |

7. Schema Checksums for Production Caching

Pre-validate once → reuse on every request for reduced latency:

async with MarkdownPipe(api_key="sk-...") as client:
    # Validate schema (free)
    v = await client.validate(MY_SCHEMA)
    CHECKSUM = v.checksum
    print(f"Estimated cost: {v.credits} credits/request")

    # Use checksum to enable server-side caching
    result = await client.extract(url, schema=MY_SCHEMA, checksum=CHECKSUM)

---

API Surface

SDK Method	Endpoint	Description
client.pipe(url, namespace?, **opts)	POST /pipe/	Preset extraction → ArticleResponse | ProductResponse | SerpResponse (no custom schema)
client.process(url, namespace)	POST /pipe/	Same as pipe (legacy name)
client.extract(url, schema, *, …)	POST /extract/	Custom Chained DSL; optional checksum / version headers
client.engine(url, schema, html)	POST /engine/	DSL on raw HTML (no fetch)
client.fetch(url)	POST /fetch	Acquire + store → FetchFreshOut | FetchCachedOut
client.markdown(html)	POST /markdown	HTML → Markdown converter
client.csv(csv_text)	POST /markdown/csv	CSV → Markdown table
client.csv_batch(items)	POST /markdown/csv/batch	Batch CSV → Markdown tables
client.excel_sheets(file)	POST /markdown/excel/sheets/	Workbook sheet names
client.map(url, depth)	POST /map	URL discovery (sitemap + RSS + link graph)
client.rss(url)	POST /map/rss	RSS/Atom feed → MapRssResponse
client.sitemap(url)	POST /map/sitemap	Sitemap → MapSitemapResponse
client.validate(schema)	POST /validate/	Schema validation + checksum + cost estimate
client.get_stored_html(url)	HTTP GET	Retrieve gzip-compressed HTML from storage
client.get_job(job_id)	GET /jobs/<id>	Retrieve stored job result
client.farm(items, concurrency)	—	Concurrent batch extraction via process()
client.harvest(seed_url, ...)	—	Discovery → filter → farm pipeline
client.health()	GET /health	Service health check

---

Synchronous Client

All async methods are available synchronously via PipeClient:

from markdownpipe import PipeClient

with PipeClient(api_key="sk-...") as client:
    article = client.article("https://example.com/blog-post")
    print(article.title)

---

Documentation

Document	Description
Docs Overview	Introduction to the Micro-Orchestrator pattern
Quickstart	Step-by-step guide from zero to production
API Reference	Full method signatures and response models
DSL Guide	Fluent Q builder and logic configuration
Pseudo-Classes Reference	Complete operator catalogue
Advanced DSL Patterns	Surgical extraction recipes for complex sites
Infrastructure Patterns	Mass scale, concurrency, and observability
Cookbook	Ready-to-use recipes: RAG pipelines, monitoring, news scraping
Integrations	Framework-specific setups (FastAPI, LangChain, etc.)
Troubleshooting	Common errors and fixes

---

License

Apache 2.0 © Markdown Pipe Team