How to Scrape Web Data With CrewAI & Oxylabs Scraper API

uv tool install crewai
uv tool list

crewai create crew search_crew

API_USERNAME=YOUR_USERNAME_HERE
API_PASSWORD=YOUR_PASSWORD_HERE

# src/search_crew/tools/custom_tool.py
import os
import json
from typing import Type

import requests
from pydantic import BaseModel, Field, PrivateAttr

from crewai.tools import BaseTool


class OxylabsSearchInput(BaseModel):
    query: str
    geo_location: str
    pages: int


class OxylabsSearch(BaseTool):
    _username: str = PrivateAttr()
    _password: str = PrivateAttr()

    name: str = 'Oxylabs Web Search'
    description: str = (
        'Searches the web using Oxylabs Web Scraper API to find content for '
        'user-defined queries. This tool returns the latest information from '
        'Google search results.'
    )
    args_schema: Type[BaseModel] = OxylabsSearchInput

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self._username = os.environ.get('API_USERNAME')
        self._password = os.environ.get('API_PASSWORD')

        if not self._username or not self._password:
            raise ValueError('Oxylabs credentials not provided.')

    def _run(self, query: str, geo_location: str, pages: int) -> str:
        payload = {
            'source': 'google_search',
            'query': query,
            'geo_location': geo_location,
            'pages': pages,
            'parse': True
        }

        try:
            response = requests.post(
                'https://realtime.oxylabs.io/v1/queries',
                auth=(self._username, self._password),
                json=payload
            )
            response.raise_for_status()

            search_results = []
            for result in response.json()['results']:
                search_results.append(result['content'])

            return json.dumps(search_results, indent=2)

        except requests.exceptions.RequestException as e:
            return f'Error making request to Oxylabs API: {str(e)}'
        except (KeyError, json.JSONDecodeError) as e:
            return f'Error processing API response: {str(e)}'

researcher:
  role: >
    Senior Researcher
  goal: >
    Scrape Google search results with strategic search parameters and extract
    high-value, factual information.
  backstory: >
    You can craft the perfect search query and other parameters to get the most
    relevant results. You pick the most relevant and accurate information from
    reliable sources.

analyst:
  role: >
    Analytics Expert
  goal: >
    Transform raw data points into cohesive, actionable intelligence through
    concise structuring
  backstory: >
    You see patterns where others see only scattered facts. Your reports reveal
    connections between facts and their implications. You excel at converting
    information overload into strategic clarity.

search_task:
  description: >
    Use OxylabsSearch according to the user's {query}. Craft API parameters as
    follows:
    - 'query': craft a concise and accurate Google search term.
    - 'geo_location': 'United States' by default or a relevant country name.
    - 'pages': 2 by default or 1–5, depending on search complexity.
    Filter the scraped Google search results by finding highly valuable
    information.

  expected_output: >
    A structured JSON object containing neatly organized comprehensive data.
    Don't rephrase, rewrite, or summarize the content.
    Provide accurate scraped information and URLs.

  agent: researcher
  output_file: output/search_data.json

report_task:
  description: >
    Create a concise, actionable intelligence report that's easy to read.
    Answer the user's {query} with concrete data. Identify valuable patterns and
    trends in the data, including related searches. Don't make up any information.

  expected_output: >
    A scannable report focusing on the most informative and relevant
    information. Choose the structure and format that best fits the data and the user's query. Prioritize clarity and brevity throughout,
    structurize in a visual way. The entire report should be readable in under 2 minutes.

  agent: analyst
  context: [search_task]
  output_file: output/report.md

# src/search_crew/crew.py

from crewai import Agent, Crew, Process, Task
from crewai.project import CrewBase, agent, crew, task

from .tools.custom_tool import OxylabsSearch


@CrewBase
class WebIntelligenceCrew:
    @agent
    def researcher(self) -> Agent:
        return Agent(
            config=self.agents_config['researcher'],
            tools=[OxylabsSearch()],
            verbose=True
        )

    @agent
    def analyst(self) -> Agent:
        return Agent(
            config=self.agents_config['analyst'],
            verbose=True
        )

    @task
    def search_task(self) -> Task:
        return Task(
            config=self.tasks_config['search_task'],
            output_file='output/search_data.json'
        )

    @task
    def report_task(self) -> Task:
        return Task(
            config=self.tasks_config['report_task'],
            output_file='output/report.md'
        )

    @crew
    def crew(self) -> Crew:
        return Crew(
            agents=self.agents,
            tasks=self.tasks,
            process=Process.sequential,
            verbose=True
        )

# src/search_crew/main.py

import os
from search_crew.crew import WebIntelligenceCrew


os.makedirs('output', exist_ok=True)

def run():
    try:
        result = WebIntelligenceCrew().crew().kickoff(
            inputs={'query': input('\nSearch for: ').strip()}
        )
        return result
    except Exception as e:
        print(f'An error occurred: {str(e)}')


if __name__ == '__main__':
    run()

cd search_crew
crewai install

crewai run