Guía de web scraping con LlamaIndex y CrawlForge MCP

from llama_index.core import Document
from crawlforge_llamaindex import CrawlForgeReader

reader = CrawlForgeReader(api_key="cf_live_your_key")
docs: list[Document] = reader.load_data(urls=["https://docs.stripe.com/api"])

pip install llama-index-core llama-index-embeddings-openai requests

export CRAWLFORGE_API_KEY="cf_live_your_key_here"
export OPENAI_API_KEY="sk-..."

# crawlforge_reader.py
import os
from typing import List, Optional
import requests
from llama_index.core.readers.base import BaseReader
from llama_index.core.schema import Document


class CrawlForgeReader(BaseReader):
    """LlamaIndex reader that uses CrawlForge for web scraping."""

    BASE_URL = "https://crawlforge.dev/api/v1/tools"

    def __init__(self, api_key: Optional[str] = None, use_stealth: bool = False):
        self.api_key = api_key or os.environ["CRAWLFORGE_API_KEY"]
        self.use_stealth = use_stealth

    def load_data(self, urls: List[str]) -> List[Document]:
        documents = []
        tool = "stealth_mode" if self.use_stealth else "extract_content"

        for url in urls:
            response = requests.post(
                f"{self.BASE_URL}/{tool}",
                headers={
                    "Authorization": f"Bearer {self.api_key}",
                    "Content-Type": "application/json",
                },
                json={
                    "url": url,
                    "options": {"format": "markdown"},
                },
                timeout=30,
            )
            response.raise_for_status()
            data = response.json()

            documents.append(
                Document(
                    text=data.get("content", ""),
                    metadata={
                        "source": url,
                        "title": data.get("title"),
                        "scraped_at": data.get("scraped_at"),
                    },
                )
            )

        return documents

# build_index.py
from llama_index.core import VectorStoreIndex
from crawlforge_reader import CrawlForgeReader

reader = CrawlForgeReader()
docs = reader.load_data(urls=[
    "https://docs.stripe.com/api/charges/create",
    "https://docs.stripe.com/api/payment_intents/create",
    "https://docs.stripe.com/api/refunds/create",
])

index = VectorStoreIndex.from_documents(docs)
index.storage_context.persist(persist_dir="./storage/stripe_docs")

# query_index.py
from llama_index.core import StorageContext, load_index_from_storage

storage = StorageContext.from_defaults(persist_dir="./storage/stripe_docs")
index = load_index_from_storage(storage)

query_engine = index.as_query_engine()
response = query_engine.query(
    "What are the required fields to create a Charge in Stripe's API?"
)
print(response)
# -> "To create a Charge, you must provide amount (integer in cents) and
#     currency (three-letter ISO code). Additionally you need a source
#     (payment method) or customer."

# docs_rag.py
import os
from datetime import datetime
from llama_index.core import VectorStoreIndex, StorageContext, load_index_from_storage
from crawlforge_reader import CrawlForgeReader

PERSIST_DIR = "./storage/stripe_docs"
TARGET_URLS = [
    "https://docs.stripe.com/api/charges/create",
    "https://docs.stripe.com/api/payment_intents/create",
    "https://docs.stripe.com/api/refunds/create",
    "https://docs.stripe.com/api/customers/create",
    "https://docs.stripe.com/api/subscriptions/create",
]


def refresh_index() -> VectorStoreIndex:
    """Re-scrape sources and rebuild the index."""
    reader = CrawlForgeReader()
    docs = reader.load_data(urls=TARGET_URLS)
    index = VectorStoreIndex.from_documents(docs)
    index.storage_context.persist(persist_dir=PERSIST_DIR)
    print(f"Indexed {len(docs)} docs at {datetime.utcnow().isoformat()}Z")
    return index


def load_index() -> VectorStoreIndex:
    """Load the persisted index from disk."""
    storage = StorageContext.from_defaults(persist_dir=PERSIST_DIR)
    return load_index_from_storage(storage)


def ask(question: str) -> str:
    index = load_index() if os.path.exists(PERSIST_DIR) else refresh_index()
    return str(index.as_query_engine().query(question))


if __name__ == "__main__":
    # Refresh once a day in a cron job: python docs_rag.py --refresh
    import sys
    if "--refresh" in sys.argv:
        refresh_index()
    else:
        print(ask("How do I create a refund for a charge?"))

# crawlforge_tools.py
from llama_index.core.tools import FunctionTool
from crawlforge_reader import CrawlForgeReader


def scrape_url(url: str) -> str:
    """Scrape a URL and return its main content as markdown."""
    reader = CrawlForgeReader()
    docs = reader.load_data(urls=[url])
    return docs[0].text if docs else ""


def search_and_scrape(query: str, n: int = 3) -> list[str]:
    """Search the web and return content from the top N results."""
    import os, requests
    resp = requests.post(
        "https://crawlforge.dev/api/v1/tools/search_web",
        headers={"Authorization": f"Bearer {os.environ['CRAWLFORGE_API_KEY']}"},
        json={"query": query, "limit": n},
        timeout=30,
    ).json()
    urls = [r["url"] for r in resp.get("results", [])]
    reader = CrawlForgeReader()
    return [d.text for d in reader.load_data(urls=urls)]


scrape_tool = FunctionTool.from_defaults(fn=scrape_url)
search_tool = FunctionTool.from_defaults(fn=search_and_scrape)

from llama_index.core.agent import ReActAgent
from llama_index.llms.openai import OpenAI

agent = ReActAgent.from_tools(
    tools=[scrape_tool, search_tool],
    llm=OpenAI(model="gpt-4o-mini"),
    verbose=True,
)

response = agent.chat(
    "Research the current state of Anthropic's MCP protocol adoption in 2026. "
    "Cite at least 3 sources."
)
print(response)