Ctx Scrapling¶

ctx_scrapling -- mvp.ctx_scrapling

Cluster: Context & Retrieval | Type: component | MCP Tools: 25

Overview¶

Resilient web scraper with a two-tier fallback chain: Scrapling (JS rendering + anti-bot evasion) when installed, then httpx + regex HTML processing. Supports CSS selector targeting, plain-text extraction, and Markdown conversion via an internal regex transformer. Both backends honour a configurable timeout and return a structured ScrapeOutput with URL, title, content, content type, and character count.

When to use:

• Fetching live web content for a retrieval or grounding pipeline
• Extracting article text or documentation from URLs without a separate scraping service
• Getting Markdown-formatted content from HTML pages for downstream LLM consumption

Example:

from mvp.ctx_scrapling import CtxScraplingBlock, ScrapeInput

block = CtxScraplingBlock(name="scraper")
result = block.infer(ScrapeInput(url="https://example.com", extract_markdown=True))
# result.value.content → Markdown text; result.value.title → page title

Works well with: ctx_markitdown, ctx_langextract, ctx_rag

Public API¶

CtxScraplingDecisionError(ValueError)¶

The LLM did not produce a usable, validated search-rerank decision.

ScraplingRerankDecision¶

Validated advisory rerank verdict over a returned search-record set.

LLMScraplingRerankRuntime¶

Provider-neutral search-result-rerank runtime backed by G6's LLM caller.

Constructor:

Methods:

rerank(query: str, records: list[Any]) -> ScraplingRerankDecision¶

CtxScraplingRerankPatternRuntime¶

Stateless, load-bearing returned-set-ceiling enforcement.

Methods:

enforce_eligibility(decision: ScraplingRerankDecision, records: list[Any]) -> tuple[list[Any], bool, bool]¶

CtxScraplingPlanner¶

Runtime-first advisory result-rerank facade with returned-order fallback.

Constructor:

Methods:

rerank(query: str, records: list[Any]) -> list[Any]¶

ScrapeInput(BaseModel)¶

Input to CtxScraplingBlock.

ScrapeOutput(BaseModel)¶

Output from CtxScraplingBlock.

CtxScraplingBlock(AIBlock[ScrapeInput, ScrapeOutput, None])¶

Resilient web scraping block.

Methods:

infer(data: ScrapeInput) -> Result[ScrapeOutput]¶

MCPScraplingRecord(BaseModel)¶

MCPScraplingInput(BaseModel)¶

MCPScraplingOutput(BaseModel)¶

CtxScraplingMCPBlock(AIBlock[MCPScraplingInput, MCPScraplingOutput, dict])¶

Full-featured web scraping block with SQLite persistence.

Methods:

infer(data: MCPScraplingInput) -> Result[MCPScraplingOutput]¶

ScraplingStore¶

Sync SQLite scrapling store with 8 tables.

Constructor:

Methods:

add_page(url: str, domain: str, title: str, html: str, content: str, content_type: str, status_code: int, fetcher_type: str, tags: list[str], metadata_json: str = '{}') -> str¶

get_page(url: str = '', page_id: str = '') -> dict[str, Any] | None¶

query_pages(domain: str = '', tags: list[str] | None = None, query: str = '', limit: int = 50) -> list[dict[str, Any]]¶

add_item(page_id: str, url: str, selector: str, data: dict, extraction_method: str, tags: list[str]) -> str¶

query_items(page_id: str = '', limit: int = 50) -> list[dict[str, Any]]¶

upsert_selector(name: str, selector_type: str, selector_value: str, description: str, domain: str, tags: list[str]) -> str¶

get_selector(name: str) -> dict[str, Any] | None¶

query_selectors(domain: str = '', limit: int = 50) -> list[dict[str, Any]]¶

create_session(start_url: str, allowed_domains: list[str], max_depth: int, max_pages: int, fetcher_type: str, config_json: str = '{}') -> str¶

get_session(session_id: str) -> dict[str, Any] | None¶

pop_queue(session_id: str) -> dict[str, Any] | None¶

complete_queue_item(queue_id: str) -> None¶

add_to_queue(session_id: str, url: str, depth: int, parent_url: str) -> str¶

increment_session_pages(session_id: str) -> None¶

increment_session_items(session_id: str, count: int = 1) -> None¶

stop_session(session_id: str) -> None¶

get_queue_stats(session_id: str) -> dict[str, int]¶

add_crawl_item(session_id: str, url: str, data: dict) -> str¶

query_crawl_items(session_id: str, limit: int = 50, offset: int = 0) -> list[dict[str, Any]]¶

add_link(source_url: str, target_url: str, anchor_text: str, domain: str, session_id: str = '') -> str¶

query_links(source_url: str = '', domain: str = '', limit: int = 50) -> list[dict[str, Any]]¶

upsert_extraction_rule(name: str, description: str, url_pattern: str, selector_value: str, selector_type: str, output_fields: list[str], domain: str, tags: list[str]) -> str¶

text_search(query: str, top_k: int = 5) -> list[dict[str, Any]]¶

TF-IDF search across pages and selectors.

count_all() -> dict[str, int]¶

Functions¶

eligible_fingerprint(records: list[Any]) -> str¶

sha256 over the returned record texts (order-sensitive).

validate_scrapling_rerank_decision(decision: ScraplingRerankDecision, n_eligible: int, expected_fingerprint: str) -> None¶

Returned-set / anti-injection guard for a search-rerank decision.

planner_is_llm_trusted(planner: Any) -> bool¶

Whether the BLOCK may report llm_used=True for planner.

MCP Tools¶