Custom Document Loader for LangChain and LangGraph with Markdown extraction

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

1. Bot Protection: Cloudflare or DataDome blocks standard Python requests.
2. 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 = {"Authorization": f"Bearer {self.api_key}"}
        for url in self.urls:
            try:
                resp = requests.post(
                    "https://api.messora.dev/v1/extract",
                    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("status_code", 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.")