Build a Multi-Agent Web Scraper with CrewAI

Multi-Agent Web Scraper with CrewAI

Many web scraping tutorials leave you with raw HTML or unstructured text, requiring significant post-processing. Additionally, most AI agent demonstrations rely on paid APIs. In this tutorial, you will build a multi-agent web scraper pipeline that locates pages, extracts content, summarizes information, and saves a clean dataset. This guide uses only free tools, including CrewAI, BeautifulSoup, and a local Llama 3.2 model via Ollama.

Multi-Agent Web Scraper with CrewAI

Before you begin, install Ollama, download the model, and install the required Python packages:

ollama pull llama3.2
pip install crewai requests beautifulsoup4 pandas

Ensure Ollama is running in the background. The script communicates with it on localhost:11434.

Step 1: Configuration

import json
import re
from urllib.parse import urljoin, urlparse, urldefrag

import requests
import pandas as pd
from bs4 import BeautifulSoup

from crewai import Agent, Task, Crew, Process, LLM
from crewai.tools import tool

HEADERS = {
    "User-Agent": "Mozilla/5.0 (compatible; EducationalResearchBot/1.0)"
}

REQUEST_TIMEOUT = 15
MAX_URLS_PER_STARTING_PAGE = 10
MAX_CONTENT_LENGTH = 12000

STARTING_URLS = [
    "https://amanxai.com"
]

OUTPUT_CSV = "research_dataset.csv"
OUTPUT_JSON = "research_dataset.json"

Two limits are important: MAX_URLS_PER_STARTING_PAGE restricts the crawl size, and MAX_CONTENT_LENGTH prevents long pages from exceeding the model’s context window. Always configure both when using local models.

Step 2: Build the Scraping Tools

Agents do not scrape websites independently. You must provide them with tools, which are standard Python functions. The first tool identifies links within the same domain:

@tool("discover_urls")
def discover_urls(base_url: str) -> str:
    """
    Discover unique HTTP/HTTPS links on the same domain
    as the supplied starting URL. Returns a JSON array
    of discovered URLs.
    """
    try:
        response = requests.get(
            base_url,
            headers=HEADERS,
            timeout=REQUEST_TIMEOUT
        )
        response.raise_for_status()

        content_type = response.headers.get("Content-Type", "")
        if "html" not in content_type.lower():
            return json.dumps([])

        soup = BeautifulSoup(response.text, "html.parser")

        base_domain = urlparse(base_url).netloc.lower()
        discovered = []

        for link in soup.find_all("a", href=True):
            url = urljoin(base_url, link["href"])
            url, _ = urldefrag(url)
            parsed = urlparse(url)

            if parsed.scheme not in ("http", "https"):
                continue

            if parsed.netloc.lower() != base_domain:
                continue

            if url not in discovered:
                discovered.append(url)

            if len(discovered) >= MAX_URLS_PER_STARTING_PAGE:
                break

        return json.dumps(discovered, ensure_ascii=False)

    except requests.RequestException as exc:
        return json.dumps({
            "error": str(exc),
            "url": base_url
        })

The second tool extracts readable text from web pages:

@tool("extract_content")
def extract_content(url: str) -> str:
    """
    Fetch an HTML webpage and extract its title and readable
    text. Returns a JSON object containing url, title and
    content. Returns an error object if extraction fails.
    """
    try:
        response = requests.get(
            url,
            headers=HEADERS,
            timeout=REQUEST_TIMEOUT
        )
        response.raise_for_status()

        content_type = response.headers.get("Content-Type", "")
        if "html" not in content_type.lower():
            return json.dumps({
                "url": url,
                "error": "The URL did not return an HTML page."
            })

        soup = BeautifulSoup(response.text, "html.parser")

        title = (
            soup.title.get_text(" ", strip=True)
            if soup.title
            else ""
        )

        for tag in soup([
            "script", "style", "nav", "footer",
            "header", "aside", "noscript", "svg", "form"
        ]):
            tag.decompose()

        main = soup.find("main") or soup.find("article") or soup.body

        if main:
            text = main.get_text(" ", strip=True)
        else:
            text = soup.get_text(" ", strip=True)

        text = re.sub(r"\s+", " ", text).strip()

        return json.dumps({
            "url": url,
            "title": title,
            "content": text[:MAX_CONTENT_LENGTH]
        }, ensure_ascii=False)

    except requests.RequestException as exc:
        return json.dumps({
            "url": url,
            "error": str(exc)
        })

Both tools return errors as JSON rather than raising exceptions. This approach allows the agent to report failed pages without interrupting the entire process. The extractor also removes navigation, footers, and scripts, ensuring the summarizer processes only relevant content.

Step 3: Connect a Free Local LLM

llm = LLM(
    model="ollama/llama3.2",
    base_url="http://localhost:11434",
    temperature=0.1
)

Setting a low temperature of 0.1 ensures predictable output, which is essential for structured data tasks.

Step 4: Create the Specialized Agents

url_agent = Agent(
    role="URL Discovery Specialist",
    goal=(
        "Discover useful same-domain webpages from the "
        "starting URLs provided by the user."
    ),
    backstory=(
        "You identify useful research pages, avoid duplicate "
        "URLs, and prioritize relevant HTML pages."
    ),
    tools=[discover_urls],
    llm=llm,
    verbose=True
)

extractor = Agent(
    role="Web Content Extractor",
    goal=(
        "Extract the title and webpage text from every "
        "discovered URL using the provided extraction tool."
    ),
    backstory=(
        "You collect source content accurately and report "
        "failed pages instead of inventing their contents."
    ),
    tools=[extract_content],
    llm=llm,
    verbose=True
)

summarizer = Agent(
    role="Research Summarizer",
    goal=(
        "Produce accurate, concise summaries grounded "
        "in the extracted webpage content."
    ),
    backstory=(
        "You are a careful research analyst. You preserve "
        "important facts and never invent missing details."
    ),
    llm=llm,
    verbose=True
)

organizer = Agent(
    role="Research Data Organizer",
    goal=(
        "Convert the page summaries into consistent, "
        "machine-readable JSON records."
    ),
    backstory=(
        "You specialize in structured research datasets "
        "and valid JSON output."
    ),
    llm=llm,
    verbose=True
)

Only the first two agents use tools. The summarizer and organizer function as reasoning agents. The repeated instruction to “never invent” is intentional, as hallucination is a significant risk with smaller models.

Step 5: Define the Tasks

discover_task = Task(
    description="""
    Starting URLs:
    {starting_urls}

    For every starting URL, use the discover_urls tool.

    Requirements:
    1. Discover links on the same domain.
    2. Remove duplicate URLs.
    3. Include the supplied starting URLs themselves
       in the final list so they can also be extracted.
    4. Do not invent URLs.
    5. Return a JSON array of URL strings only.
    """,
    expected_output=(
        "A valid JSON array containing unique starting "
        "and discovered URLs."
    ),
    agent=url_agent
)

extract_task = Task(
    description="""
    Use the URL list from the previous task.

    For each URL:
    1. Call the extract_content tool.
    2. Collect the URL, title and extracted content.
    3. Skip failed pages and non-HTML pages.
    4. Do not fabricate page content.
    5. Keep the extracted text within the tool's limits.

    Return the successfully extracted records as JSON.
    Report failed URLs separately.
    """,
    expected_output=(
        "JSON containing successfully extracted page "
        "records and a list of failed URLs."
    ),
    agent=extractor,
    context=[discover_task]
)

summary_task = Task(
    description="""
    Summarize the successfully extracted webpage records
    provided by the previous task.

    For each page:
    1. Preserve its original URL and title.
    2. Write a factual summary in 3-5 sentences.
    3. Identify 3-7 concise key topics.
    4. Use only information supported by the source text.
    5. Skip failed pages.
    6. Do not invent facts or claim to have read
       content that was not provided.

    Return a JSON array. Each object must have:
    url, title, summary, key_topics.
    """,
    expected_output=(
        "A JSON array of summarized pages with url, "
        "title, summary and key_topics."
    ),
    agent=summarizer,
    context=[extract_task]
)

organize_task = Task(
    description="""
    Produce the final research dataset from the summaries
    generated in the previous task.

    Return a valid JSON array only.

    Every object must contain exactly these fields:
    - url: string
    - title: string
    - summary: string
    - key_topics: array of strings

    Preserve the source URLs. Remove duplicate records.
    Do not add explanations, Markdown fences, or text
    before or after the JSON array.
    """,
    expected_output=(
        "A valid JSON array of structured research records."
    ),
    agent=organizer,
    context=[summary_task]
)

The context parameter links the agents. Each task receives the output of the previous one, allowing the pipeline to progress from URLs to content, summaries, and final JSON. The {starting_urls} placeholder is populated at runtime.

Preparing for AI/ML Interviews?
Building multi-agent projects is a great way to strengthen your portfolio, but you should also be able to explain how your agents work and why you designed them that way. Cracking Your First AI/ML Interview is a resource I’d recommend if you’re preparing for your first AI/ML interview.

Step 6: Assemble the Crew

crew = Crew(
    agents=[url_agent, extractor, summarizer, organizer],
    tasks=[discover_task, extract_task, summary_task, organize_task],
    process=Process.sequential,
    verbose=True
)

Process.sequential executes tasks in order, which is appropriate when each step relies on the previous one.

Step 7: Parse the Output Safely

Do not assume an LLM will return perfect JSON. This parser removes Markdown fences, extracts arrays from additional text, and validates each record:

def parse_json_array(result):
    output = (
        result.raw
        if hasattr(result, "raw")
        else str(result)
    )

    output = output.strip()

    output = re.sub(
        r"^```(?:json)?\s*", "", output, flags=re.IGNORECASE
    )
    output = re.sub(r"\s*```$", "", output)

    try:
        data = json.loads(output)
    except json.JSONDecodeError:
        start = output.find("[")
        end = output.rfind("]")

        if start == -1 or end == -1 or end < start:
            raise ValueError(
                "CrewAI did not return a valid JSON array. "
                "Inspect the final task output."
            )

        data = json.loads(output[start:end + 1])

    if not isinstance(data, list):
        raise ValueError(
            "Expected a JSON array from the organizer agent."
        )

    required_fields = {"url", "title", "summary", "key_topics"}

    for index, record in enumerate(data):
        if not isinstance(record, dict):
            raise ValueError(f"Record {index} is not a JSON object.")

        missing = required_fields - record.keys()

        if missing:
            raise ValueError(
                f"Record {index} is missing fields: {sorted(missing)}"
            )

        if not isinstance(record["key_topics"], list):
            raise ValueError(f"Record {index} has invalid key_topics.")

    return data

Step 8: Run the Workflow and Save the Dataset

def main():
    print("\nStarting the multi-agent web scraper...\n")

    result = crew.kickoff(
        inputs={"starting_urls": STARTING_URLS}
    )

    print("\nCrewAI workflow finished.")

    try:
        records = parse_json_array(result)
    except (ValueError, json.JSONDecodeError) as exc:
        print(f"\nCould not parse final dataset: {exc}")
        print("\nRaw CrewAI output:\n")
        print(result)
        return

    if not records:
        print("No research records were produced.")
        return

    with open(OUTPUT_JSON, "w", encoding="utf-8") as file:
        json.dump(records, file, indent=2, ensure_ascii=False)

    df = pd.DataFrame(records)
    df["key_topics"] = df["key_topics"].apply(
        lambda topics: json.dumps(topics, ensure_ascii=False)
    )

    df.to_csv(OUTPUT_CSV, index=False, encoding="utf-8-sig")

    print(f"\nSuccessfully processed {len(records)} pages.")
    print(f"JSON saved to: {OUTPUT_JSON}")
    print(f"CSV saved to: {OUTPUT_CSV}")

    print("\nDataset preview:")
    print(df.head().to_string(index=False))


if __name__ == "__main__":
    main()

Run your Python file. The script generates research_dataset.json and research_dataset.csv, each containing a URL, title, summary, and key topics for every page.

Final Output:

Successfully processed 1 pages.
JSON saved to: research_dataset.json
CSV saved to: research_dataset.csv

Dataset preview:
                url                   title                                                                                                                                        summary                                                                                          key_topics
https://amanxai.com AmanXai by Aman Kharwal Aman Kharwal provides tutorials and articles on AI, Machine Learning, and Data Science, focusing on practical and easy-to-follow explanations. ["AI", "Machine Learning", "Data Science", "Practical Guides", "AI Projects", "MLOps", "AI Agents"]

Where to Use This (and Where Not To)

This approach is effective for competitor research, content audits, building small RAG datasets, and summarizing documentation sites. It is suitable when you require an understanding of pages rather than simple data extraction.

This method is not suitable for large-scale crawling. For such tasks, a standard Python script is faster and more reliable. Always review a site’s robots.txt and terms before scraping. Some variability is expected from a 3B model. If output is malformed, rerun the process, reduce MAX_URLS_PER_STARTING_PAGE, or use a larger Ollama model.

The Takeaway

The key takeaway is not CrewAI itself, but the practice of dividing complex problems into small, well-defined roles and validating each output.

This mindset applies to every production AI system you develop, whether you use free local models or advanced ones.

Begin with this pipeline, apply it to a relevant site, and consider extending it with additional agents or output formats.

Thank you for reading this article on building a multi-agent web scraper with CrewAI. For further insights on AI and machine learning, you are welcome to follow me on Instagram.

Aman Kharwal
Aman Kharwal

AI/ML Engineer | Published Author. My aim is to decode data science for the real world in the most simple words.

Articles: 2232

Recommended For You

Leave a Reply

Discover more from AmanXai by Aman Kharwal

Subscribe now to keep reading and get access to the full archive.

Continue reading