This domain is for use in illustrative examples.
- More information... - - -""" - -# Use the tool with HTML content -result_html = tool.invoke( - { - "website_url": website_url, # Still required but will be overridden - "website_html": html_content, - "user_prompt": user_prompt, - } -) -print("\nResult from HTML:", result_html) diff --git a/langchain_scrapegraph/__init__.py b/langchain_scrapegraph/__init__.py index 30767d1..f45870f 100644 --- a/langchain_scrapegraph/__init__.py +++ b/langchain_scrapegraph/__init__.py @@ -1,3 +1,3 @@ """langchain-scrapegraph package.""" -__version__ = "1.10.0" +__version__ = "2.0.0" diff --git a/langchain_scrapegraph/tools/__init__.py b/langchain_scrapegraph/tools/__init__.py index edc9e53..f956e8b 100644 --- a/langchain_scrapegraph/tools/__init__.py +++ b/langchain_scrapegraph/tools/__init__.py @@ -1,37 +1,34 @@ -from .agentic_scraper import AgenticScraperTool +from .crawl import CrawlResumeTool, CrawlStartTool, CrawlStatusTool, CrawlStopTool from .credits import GetCreditsTool +from .extract import ExtractTool +from .history import HistoryTool from .markdownify import MarkdownifyTool -from .scheduled_jobs import ( - CreateScheduledJobTool, - DeleteScheduledJobTool, - GetJobExecutionsTool, - GetScheduledJobsTool, - GetScheduledJobTool, - PauseScheduledJobTool, - ResumeScheduledJobTool, - TriggerScheduledJobTool, - UpdateScheduledJobTool, +from .monitor import ( + MonitorCreateTool, + MonitorDeleteTool, + MonitorGetTool, + MonitorListTool, + MonitorPauseTool, + MonitorResumeTool, ) from .scrape import ScrapeTool -from .searchscraper import SearchScraperTool -from .smartcrawler import SmartCrawlerTool -from .smartscraper import SmartScraperTool +from .search import SearchTool __all__ = [ - "AgenticScraperTool", - "CreateScheduledJobTool", - "DeleteScheduledJobTool", + "CrawlResumeTool", + "CrawlStartTool", + "CrawlStatusTool", + "CrawlStopTool", + "ExtractTool", "GetCreditsTool", - "GetJobExecutionsTool", - "GetScheduledJobsTool", - "GetScheduledJobTool", + "HistoryTool", "MarkdownifyTool", - "PauseScheduledJobTool", - "ResumeScheduledJobTool", + "MonitorCreateTool", + "MonitorDeleteTool", + "MonitorGetTool", + "MonitorListTool", + "MonitorPauseTool", + "MonitorResumeTool", "ScrapeTool", - "SearchScraperTool", - "SmartCrawlerTool", - "SmartScraperTool", - "TriggerScheduledJobTool", - "UpdateScheduledJobTool", + "SearchTool", ] diff --git a/langchain_scrapegraph/tools/agentic_scraper.py b/langchain_scrapegraph/tools/agentic_scraper.py deleted file mode 100644 index 397dec2..0000000 --- a/langchain_scrapegraph/tools/agentic_scraper.py +++ /dev/null @@ -1,256 +0,0 @@ -# Models for agentic scraper endpoint - -from typing import Any, Dict, List, Optional, Type -from uuid import UUID - -from langchain_core.callbacks import ( - AsyncCallbackManagerForToolRun, - CallbackManagerForToolRun, -) -from langchain_core.tools import BaseTool -from langchain_core.utils import get_from_dict_or_env -from pydantic import BaseModel, Field, model_validator -from scrapegraph_py import Client - - -class AgenticScraperRequest(BaseModel): - url: str = Field( - ..., - example="https://dashboard.scrapegraphai.com/", - description="The URL to scrape", - ) - use_session: bool = Field( - default=True, description="Whether to use session for the scraping" - ) - steps: List[str] = Field( - ..., - example=[ - "Type email@gmail.com in email input box", - "Type test-password@123 in password inputbox", - "click on login", - ], - description="List of steps to perform on the webpage", - ) - user_prompt: Optional[str] = Field( - default=None, - example="Extract user information and available dashboard sections", - description="Prompt for AI extraction (only used when ai_extraction=True)", - ) - output_schema: Optional[Dict[str, Any]] = Field( - default=None, - example={ - "user_info": { - "type": "object", - "properties": { - "username": {"type": "string"}, - "email": {"type": "string"}, - "dashboard_sections": { - "type": "array", - "items": {"type": "string"}, - }, - }, - } - }, - description="Schema for structured data extraction (only used when ai_extraction=True)", - ) - ai_extraction: bool = Field( - default=False, - description="Whether to use AI for data extraction from the scraped content", - ) - - @model_validator(mode="after") - def validate_url(self) -> "AgenticScraperRequest": - if not self.url.strip(): - raise ValueError("URL cannot be empty") - if not (self.url.startswith("http://") or self.url.startswith("https://")): - raise ValueError("Invalid URL - must start with http:// or https://") - return self - - @model_validator(mode="after") - def validate_steps(self) -> "AgenticScraperRequest": - if not self.steps: - raise ValueError("Steps cannot be empty") - if any(not step.strip() for step in self.steps): - raise ValueError("All steps must contain valid instructions") - return self - - @model_validator(mode="after") - def validate_ai_extraction(self) -> "AgenticScraperRequest": - if self.ai_extraction: - if not self.user_prompt or not self.user_prompt.strip(): - raise ValueError("user_prompt is required when ai_extraction=True") - return self - - def model_dump(self, *args, **kwargs) -> dict: - # Set exclude_none=True to exclude None values from serialization - kwargs.setdefault("exclude_none", True) - return super().model_dump(*args, **kwargs) - - -class GetAgenticScraperRequest(BaseModel): - """Request model for get_agenticscraper endpoint""" - - request_id: str = Field(..., example="123e4567-e89b-12d3-a456-426614174000") - - @model_validator(mode="after") - def validate_request_id(self) -> "GetAgenticScraperRequest": - try: - # Validate the request_id is a valid UUID - UUID(self.request_id) - except ValueError: - raise ValueError("request_id must be a valid UUID") - return self - - -class AgenticScraperTool(BaseTool): - """Tool for performing agentic web scraping using ScrapeGraph AI. - - This tool allows you to define a series of steps to perform on a webpage, - such as filling forms, clicking buttons, and extracting data. - - Setup: - Install ``langchain-scrapegraph`` python package: - - .. code-block:: bash - - pip install langchain-scrapegraph - - Get your API key from ScrapeGraph AI (https://scrapegraphai.com) - and set it as an environment variable: - - .. code-block:: bash - - export SGAI_API_KEY="your-api-key" - - Key init args: - api_key: Your ScrapeGraph AI API key. If not provided, will look for SGAI_API_KEY env var. - client: Optional pre-configured ScrapeGraph client instance. - llm_output_schema: Optional Pydantic model or dictionary schema to structure the output. - If provided, the tool will ensure the output conforms to this schema. - - Instantiate: - .. code-block:: python - - from langchain_scrapegraph.tools import AgenticScraperTool - - # Will automatically get SGAI_API_KEY from environment - tool = AgenticScraperTool() - - # Or provide API key directly - tool = AgenticScraperTool(api_key="your-api-key") - - Use the tool: - .. code-block:: python - - # Basic usage with steps - result = tool.invoke({ - "url": "https://example.com/login", - "steps": [ - "Type 'user@example.com' in email input box", - "Type 'password123' in password input box", - "Click on login button" - ] - }) - - # With AI extraction - result = tool.invoke({ - "url": "https://dashboard.example.com", - "steps": [ - "Navigate to user profile section", - "Click on settings tab" - ], - "ai_extraction": True, - "user_prompt": "Extract user profile information and available settings", - "output_schema": { - "user_info": { - "type": "object", - "properties": { - "username": {"type": "string"}, - "email": {"type": "string"}, - "settings": {"type": "array", "items": {"type": "string"}} - } - } - } - }) - - """ - - name: str = "agentic_scraper" - description: str = ( - "Perform agentic web scraping by executing a series of steps on a webpage. " - "Supports form filling, button clicking, navigation, and AI-powered data extraction." - ) - args_schema: Type[BaseModel] = AgenticScraperRequest - return_direct: bool = False - - api_key: Optional[str] = Field(default=None, description="ScrapeGraph AI API key") - client: Optional[Client] = Field( - default=None, description="ScrapeGraph client instance" - ) - llm_output_schema: Optional[Type[BaseModel]] = Field( - default=None, description="Optional Pydantic model to structure the output" - ) - - def __init__(self, **kwargs): - super().__init__(**kwargs) - self.api_key = get_from_dict_or_env(kwargs, "api_key", "SGAI_API_KEY") - if not self.client: - self.client = Client(api_key=self.api_key) - - def _run( - self, - url: str, - steps: List[str], - use_session: bool = True, - user_prompt: Optional[str] = None, - output_schema: Optional[Dict[str, Any]] = None, - ai_extraction: bool = False, - run_manager: Optional[CallbackManagerForToolRun] = None, - ) -> Dict[str, Any]: - """Run the agentic scraper tool.""" - try: - # Prepare the request payload - payload = { - "url": url, - "use_session": use_session, - "steps": steps, - "ai_extraction": ai_extraction, - } - - if ai_extraction and user_prompt: - payload["user_prompt"] = user_prompt - if output_schema: - payload["output_schema"] = output_schema - - # Call the ScrapeGraph API - response = self.client.agentic_scraper(**payload) - - return response - - except Exception as e: - if run_manager: - run_manager.on_tool_error(e, tool_name=self.name) - raise e - - async def _arun( - self, - url: str, - steps: List[str], - use_session: bool = True, - user_prompt: Optional[str] = None, - output_schema: Optional[Dict[str, Any]] = None, - ai_extraction: bool = False, - run_manager: Optional[AsyncCallbackManagerForToolRun] = None, - ) -> Dict[str, Any]: - """Run the agentic scraper tool asynchronously.""" - # For now, just call the sync version - # In a real implementation, you might want to use async HTTP client - return self._run( - url=url, - steps=steps, - use_session=use_session, - user_prompt=user_prompt, - output_schema=output_schema, - ai_extraction=ai_extraction, - run_manager=run_manager, - ) diff --git a/langchain_scrapegraph/tools/crawl.py b/langchain_scrapegraph/tools/crawl.py new file mode 100644 index 0000000..b9698e5 --- /dev/null +++ b/langchain_scrapegraph/tools/crawl.py @@ -0,0 +1,216 @@ +from typing import Any, Dict, List, Optional, Type + +from langchain_core.callbacks import ( + AsyncCallbackManagerForToolRun, + CallbackManagerForToolRun, +) +from langchain_core.tools import BaseTool +from langchain_core.utils import get_from_dict_or_env +from pydantic import BaseModel, Field, model_validator +from scrapegraph_py import Client + + +class CrawlStartInput(BaseModel): + url: str = Field(description="The starting URL for the crawl") + depth: int = Field(default=2, description="Maximum crawl depth (1-10)") + max_pages: int = Field(default=10, description="Maximum pages to crawl (1-100)") + format: str = Field( + default="markdown", description="Output format: 'markdown' or 'html'" + ) + include_patterns: Optional[List[str]] = Field( + default=None, description="URL patterns to include" + ) + exclude_patterns: Optional[List[str]] = Field( + default=None, description="URL patterns to exclude" + ) + + +class CrawlIdInput(BaseModel): + crawl_id: str = Field(description="The crawl job ID") + + +class CrawlStartTool(BaseTool): + """Tool for starting a crawl job using ScrapeGraph AI v2 API.""" + + name: str = "CrawlStart" + description: str = ( + "Start a crawl job that will crawl multiple pages from a starting URL. " + "Returns a job ID that can be used to check status." + ) + args_schema: Type[BaseModel] = CrawlStartInput + return_direct: bool = True + client: Optional[Client] = None + api_key: str + + @model_validator(mode="before") + @classmethod + def validate_environment(cls, values: Dict) -> Dict: + values["api_key"] = get_from_dict_or_env(values, "api_key", "SGAI_API_KEY") + values["client"] = Client(api_key=values["api_key"]) + return values + + def __init__(self, **data: Any): + super().__init__(**data) + + def _run( + self, + url: str, + depth: int = 2, + max_pages: int = 10, + format: str = "markdown", + include_patterns: Optional[List[str]] = None, + exclude_patterns: Optional[List[str]] = None, + run_manager: Optional[CallbackManagerForToolRun] = None, + ) -> dict: + if not self.client: + raise ValueError("Client not initialized") + + response = self.client.crawl.start( + url=url, + depth=depth, + max_pages=max_pages, + format=format, + include_patterns=include_patterns, + exclude_patterns=exclude_patterns, + ) + return response + + async def _arun( + self, + url: str, + depth: int = 2, + max_pages: int = 10, + format: str = "markdown", + include_patterns: Optional[List[str]] = None, + exclude_patterns: Optional[List[str]] = None, + run_manager: Optional[AsyncCallbackManagerForToolRun] = None, + ) -> dict: + return self._run( + url=url, + depth=depth, + max_pages=max_pages, + format=format, + include_patterns=include_patterns, + exclude_patterns=exclude_patterns, + run_manager=run_manager.get_sync() if run_manager else None, + ) + + +class CrawlStatusTool(BaseTool): + """Tool for getting the status of a crawl job.""" + + name: str = "CrawlStatus" + description: str = "Get the status and results of a crawl job by its ID." + args_schema: Type[BaseModel] = CrawlIdInput + return_direct: bool = True + client: Optional[Client] = None + api_key: str + + @model_validator(mode="before") + @classmethod + def validate_environment(cls, values: Dict) -> Dict: + values["api_key"] = get_from_dict_or_env(values, "api_key", "SGAI_API_KEY") + values["client"] = Client(api_key=values["api_key"]) + return values + + def __init__(self, **data: Any): + super().__init__(**data) + + def _run( + self, + crawl_id: str, + run_manager: Optional[CallbackManagerForToolRun] = None, + ) -> dict: + if not self.client: + raise ValueError("Client not initialized") + return self.client.crawl.status(crawl_id) + + async def _arun( + self, + crawl_id: str, + run_manager: Optional[AsyncCallbackManagerForToolRun] = None, + ) -> dict: + return self._run( + crawl_id=crawl_id, + run_manager=run_manager.get_sync() if run_manager else None, + ) + + +class CrawlStopTool(BaseTool): + """Tool for stopping a running crawl job.""" + + name: str = "CrawlStop" + description: str = "Stop a running crawl job by its ID." + args_schema: Type[BaseModel] = CrawlIdInput + return_direct: bool = True + client: Optional[Client] = None + api_key: str + + @model_validator(mode="before") + @classmethod + def validate_environment(cls, values: Dict) -> Dict: + values["api_key"] = get_from_dict_or_env(values, "api_key", "SGAI_API_KEY") + values["client"] = Client(api_key=values["api_key"]) + return values + + def __init__(self, **data: Any): + super().__init__(**data) + + def _run( + self, + crawl_id: str, + run_manager: Optional[CallbackManagerForToolRun] = None, + ) -> dict: + if not self.client: + raise ValueError("Client not initialized") + return self.client.crawl.stop(crawl_id) + + async def _arun( + self, + crawl_id: str, + run_manager: Optional[AsyncCallbackManagerForToolRun] = None, + ) -> dict: + return self._run( + crawl_id=crawl_id, + run_manager=run_manager.get_sync() if run_manager else None, + ) + + +class CrawlResumeTool(BaseTool): + """Tool for resuming a stopped crawl job.""" + + name: str = "CrawlResume" + description: str = "Resume a stopped crawl job by its ID." + args_schema: Type[BaseModel] = CrawlIdInput + return_direct: bool = True + client: Optional[Client] = None + api_key: str + + @model_validator(mode="before") + @classmethod + def validate_environment(cls, values: Dict) -> Dict: + values["api_key"] = get_from_dict_or_env(values, "api_key", "SGAI_API_KEY") + values["client"] = Client(api_key=values["api_key"]) + return values + + def __init__(self, **data: Any): + super().__init__(**data) + + def _run( + self, + crawl_id: str, + run_manager: Optional[CallbackManagerForToolRun] = None, + ) -> dict: + if not self.client: + raise ValueError("Client not initialized") + return self.client.crawl.resume(crawl_id) + + async def _arun( + self, + crawl_id: str, + run_manager: Optional[AsyncCallbackManagerForToolRun] = None, + ) -> dict: + return self._run( + crawl_id=crawl_id, + run_manager=run_manager.get_sync() if run_manager else None, + ) diff --git a/langchain_scrapegraph/tools/credits.py b/langchain_scrapegraph/tools/credits.py index d4ea94e..d6aec2c 100644 --- a/langchain_scrapegraph/tools/credits.py +++ b/langchain_scrapegraph/tools/credits.py @@ -27,32 +27,18 @@ class GetCreditsTool(BaseTool): export SGAI_API_KEY="your-api-key" - Key init args: - api_key: Your ScrapeGraph AI API key. If not provided, will look for SGAI_API_KEY env var. - client: Optional pre-configured ScrapeGraph client instance. - Instantiate: .. code-block:: python from langchain_scrapegraph.tools import GetCreditsTool - # Will automatically get SGAI_API_KEY from environment tool = GetCreditsTool() - # Or provide API key directly - tool = GetCreditsTool(api_key="your-api-key") - Use the tool: .. code-block:: python result = tool.invoke({}) - print(result) - # { - # "remaining_credits": 100, - # "total_credits_used": 50 - # } - Async usage: .. code-block:: python @@ -70,7 +56,6 @@ class GetCreditsTool(BaseTool): @model_validator(mode="before") @classmethod def validate_environment(cls, values: Dict) -> Dict: - """Validate that api key exists in environment.""" values["api_key"] = get_from_dict_or_env(values, "api_key", "SGAI_API_KEY") values["client"] = Client(api_key=values["api_key"]) return values @@ -79,14 +64,12 @@ def __init__(self, **data: Any): super().__init__(**data) def _run(self, run_manager: Optional[CallbackManagerForToolRun] = None) -> dict: - """Get the available credits.""" if not self.client: raise ValueError("Client not initialized") - return self.client.get_credits() + return self.client.credits() async def _arun( self, run_manager: Optional[AsyncCallbackManagerForToolRun] = None, - ) -> int: - """Get the available credits asynchronously.""" + ) -> dict: return self._run(run_manager=run_manager.get_sync() if run_manager else None) diff --git a/langchain_scrapegraph/tools/extract.py b/langchain_scrapegraph/tools/extract.py new file mode 100644 index 0000000..8b426fe --- /dev/null +++ b/langchain_scrapegraph/tools/extract.py @@ -0,0 +1,123 @@ +from typing import Any, Dict, Optional, Type, TypeVar + +from langchain_core.callbacks import ( + AsyncCallbackManagerForToolRun, + CallbackManagerForToolRun, +) +from langchain_core.tools import BaseTool +from langchain_core.utils import get_from_dict_or_env +from pydantic import BaseModel, Field, model_validator +from scrapegraph_py import Client + +T = TypeVar("T", bound=BaseModel) + + +class ExtractInput(BaseModel): + url: str = Field(description="URL of the webpage to extract data from") + prompt: str = Field( + description="Prompt describing what to extract from the webpage and how to structure the output" + ) + + +class ExtractTool(BaseTool): + """Tool for extracting structured data from websites using ScrapeGraph AI v2 API. + + Setup: + Install ``langchain-scrapegraph`` python package: + + .. code-block:: bash + + pip install langchain-scrapegraph + + Get your API key from ScrapeGraph AI (https://scrapegraphai.com) + and set it as an environment variable: + + .. code-block:: bash + + export SGAI_API_KEY="your-api-key" + + Key init args: + api_key: Your ScrapeGraph AI API key. If not provided, will look for SGAI_API_KEY env var. + client: Optional pre-configured ScrapeGraph client instance. + llm_output_schema: Optional Pydantic model class to structure the output. + + Instantiate: + .. code-block:: python + + from langchain_scrapegraph.tools import ExtractTool + + tool = ExtractTool() + + # With output schema: + from pydantic import BaseModel, Field + + class WebsiteInfo(BaseModel): + title: str = Field(description="The main title") + description: str = Field(description="The main description") + + tool_with_schema = ExtractTool(llm_output_schema=WebsiteInfo) + + Use the tool: + .. code-block:: python + + result = tool.invoke({ + "url": "https://example.com", + "prompt": "Extract the main heading and first paragraph" + }) + + Async usage: + .. code-block:: python + + result = await tool.ainvoke({ + "url": "https://example.com", + "prompt": "Extract the main heading" + }) + """ + + name: str = "Extract" + description: str = ( + "Extract structured data from a webpage using AI, by providing a URL and an extraction prompt" + ) + args_schema: Type[BaseModel] = ExtractInput + return_direct: bool = True + client: Optional[Client] = None + api_key: str + llm_output_schema: Optional[Type[BaseModel]] = None + + @model_validator(mode="before") + @classmethod + def validate_environment(cls: Type[T], values: Dict[str, Any]) -> Dict[str, Any]: + values["api_key"] = get_from_dict_or_env(values, "api_key", "SGAI_API_KEY") + values["client"] = Client(api_key=values["api_key"]) + return values + + def __init__(self, **data: Any): + super().__init__(**data) + + def _run( + self, + url: str, + prompt: str, + run_manager: Optional[CallbackManagerForToolRun] = None, + ) -> Dict[str, Any]: + if not self.client: + raise ValueError("Client not initialized") + + response = self.client.extract( + url=url, + prompt=prompt, + output_schema=self.llm_output_schema, + ) + return response + + async def _arun( + self, + url: str, + prompt: str, + run_manager: Optional[AsyncCallbackManagerForToolRun] = None, + ) -> Dict[str, Any]: + return self._run( + url, + prompt, + run_manager=run_manager.get_sync() if run_manager else None, + ) diff --git a/langchain_scrapegraph/tools/history.py b/langchain_scrapegraph/tools/history.py new file mode 100644 index 0000000..9acdf38 --- /dev/null +++ b/langchain_scrapegraph/tools/history.py @@ -0,0 +1,79 @@ +from typing import Any, Dict, Optional, Type + +from langchain_core.callbacks import ( + AsyncCallbackManagerForToolRun, + CallbackManagerForToolRun, +) +from langchain_core.tools import BaseTool +from langchain_core.utils import get_from_dict_or_env +from pydantic import BaseModel, Field, model_validator +from scrapegraph_py import Client + + +class HistoryInput(BaseModel): + endpoint: Optional[str] = Field( + default=None, + description="Filter by endpoint name (e.g. 'scrape', 'extract', 'search')", + ) + status: Optional[str] = Field(default=None, description="Filter by request status") + limit: Optional[int] = Field( + default=None, description="Maximum number of results (1-100)" + ) + offset: Optional[int] = Field(default=None, description="Number of results to skip") + + +class HistoryTool(BaseTool): + """Tool for retrieving request history from ScrapeGraph AI v2 API.""" + + name: str = "History" + description: str = ( + "Retrieve your ScrapeGraph AI request history with optional filters." + ) + args_schema: Type[BaseModel] = HistoryInput + return_direct: bool = True + client: Optional[Client] = None + api_key: str + + @model_validator(mode="before") + @classmethod + def validate_environment(cls, values: Dict) -> Dict: + values["api_key"] = get_from_dict_or_env(values, "api_key", "SGAI_API_KEY") + values["client"] = Client(api_key=values["api_key"]) + return values + + def __init__(self, **data: Any): + super().__init__(**data) + + def _run( + self, + endpoint: Optional[str] = None, + status: Optional[str] = None, + limit: Optional[int] = None, + offset: Optional[int] = None, + run_manager: Optional[CallbackManagerForToolRun] = None, + ) -> dict: + if not self.client: + raise ValueError("Client not initialized") + + return self.client.history( + endpoint=endpoint, + status=status, + limit=limit, + offset=offset, + ) + + async def _arun( + self, + endpoint: Optional[str] = None, + status: Optional[str] = None, + limit: Optional[int] = None, + offset: Optional[int] = None, + run_manager: Optional[AsyncCallbackManagerForToolRun] = None, + ) -> dict: + return self._run( + endpoint=endpoint, + status=status, + limit=limit, + offset=offset, + run_manager=run_manager.get_sync() if run_manager else None, + ) diff --git a/langchain_scrapegraph/tools/markdownify.py b/langchain_scrapegraph/tools/markdownify.py index 4750f5b..fa4041d 100644 --- a/langchain_scrapegraph/tools/markdownify.py +++ b/langchain_scrapegraph/tools/markdownify.py @@ -17,6 +17,8 @@ class MarkdownifyInput(BaseModel): class MarkdownifyTool(BaseTool): """Tool for converting webpages to Markdown format using ScrapeGraph AI. + In v2, this uses the scrape endpoint with format='markdown'. + Setup: Install ``langchain-scrapegraph`` python package: @@ -31,21 +33,13 @@ class MarkdownifyTool(BaseTool): export SGAI_API_KEY="your-api-key" - Key init args: - api_key: Your ScrapeGraph AI API key. If not provided, will look for SGAI_API_KEY env var. - client: Optional pre-configured ScrapeGraph client instance. - Instantiate: .. code-block:: python from langchain_scrapegraph.tools import MarkdownifyTool - # Will automatically get SGAI_API_KEY from environment tool = MarkdownifyTool() - # Or provide API key directly - tool = MarkdownifyTool(api_key="your-api-key") - Use the tool: .. code-block:: python @@ -53,11 +47,6 @@ class MarkdownifyTool(BaseTool): "website_url": "https://example.com" }) - print(result) - # # Example Domain - # - # This domain is for use in illustrative examples... - Async usage: .. code-block:: python @@ -78,7 +67,6 @@ class MarkdownifyTool(BaseTool): @model_validator(mode="before") @classmethod def validate_environment(cls, values: Dict) -> Dict: - """Validate that api key exists in environment.""" values["api_key"] = get_from_dict_or_env(values, "api_key", "SGAI_API_KEY") values["client"] = Client(api_key=values["api_key"]) return values @@ -91,18 +79,16 @@ def _run( website_url: str, run_manager: Optional[CallbackManagerForToolRun] = None, ) -> dict: - """Use the tool to extract data from a website.""" if not self.client: raise ValueError("Client not initialized") - response = self.client.markdownify(website_url=website_url) - return response["result"] + response = self.client.scrape(url=website_url, format="markdown") + return response async def _arun( self, website_url: str, run_manager: Optional[AsyncCallbackManagerForToolRun] = None, - ) -> str: - """Use the tool asynchronously.""" + ) -> dict: return self._run( website_url, run_manager=run_manager.get_sync() if run_manager else None, diff --git a/langchain_scrapegraph/tools/monitor.py b/langchain_scrapegraph/tools/monitor.py new file mode 100644 index 0000000..0f3ef84 --- /dev/null +++ b/langchain_scrapegraph/tools/monitor.py @@ -0,0 +1,282 @@ +from typing import Any, Dict, Optional, Type + +from langchain_core.callbacks import ( + AsyncCallbackManagerForToolRun, + CallbackManagerForToolRun, +) +from langchain_core.tools import BaseTool +from langchain_core.utils import get_from_dict_or_env +from pydantic import BaseModel, Field, model_validator +from scrapegraph_py import Client + + +class MonitorCreateInput(BaseModel): + name: str = Field(description="Name of the monitor") + url: str = Field(description="URL to monitor") + prompt: str = Field(description="Prompt for AI extraction") + cron: str = Field( + description="Cron expression for scheduling (e.g., '0 9 * * *' for daily at 9 AM)" + ) + output_schema: Optional[Dict[str, Any]] = Field( + default=None, description="Optional JSON Schema for structured output" + ) + + +class MonitorIdInput(BaseModel): + monitor_id: str = Field(description="The monitor ID") + + +class MonitorCreateTool(BaseTool): + """Tool for creating a monitor using ScrapeGraph AI v2 API.""" + + name: str = "MonitorCreate" + description: str = ( + "Create a new monitor that automatically extracts data from a URL on a schedule." + ) + args_schema: Type[BaseModel] = MonitorCreateInput + return_direct: bool = True + client: Optional[Client] = None + api_key: str + + @model_validator(mode="before") + @classmethod + def validate_environment(cls, values: Dict) -> Dict: + values["api_key"] = get_from_dict_or_env(values, "api_key", "SGAI_API_KEY") + values["client"] = Client(api_key=values["api_key"]) + return values + + def __init__(self, **data: Any): + super().__init__(**data) + + def _run( + self, + name: str, + url: str, + prompt: str, + cron: str, + output_schema: Optional[Dict[str, Any]] = None, + run_manager: Optional[CallbackManagerForToolRun] = None, + ) -> dict: + if not self.client: + raise ValueError("Client not initialized") + + response = self.client.monitor.create( + name=name, + url=url, + prompt=prompt, + cron=cron, + output_schema=output_schema, + ) + return response + + async def _arun( + self, + name: str, + url: str, + prompt: str, + cron: str, + output_schema: Optional[Dict[str, Any]] = None, + run_manager: Optional[AsyncCallbackManagerForToolRun] = None, + ) -> dict: + return self._run( + name=name, + url=url, + prompt=prompt, + cron=cron, + output_schema=output_schema, + run_manager=run_manager.get_sync() if run_manager else None, + ) + + +class MonitorListTool(BaseTool): + """Tool for listing all monitors.""" + + name: str = "MonitorList" + description: str = "List all monitors in your ScrapeGraph AI account." + return_direct: bool = True + client: Optional[Client] = None + api_key: str + + @model_validator(mode="before") + @classmethod + def validate_environment(cls, values: Dict) -> Dict: + values["api_key"] = get_from_dict_or_env(values, "api_key", "SGAI_API_KEY") + values["client"] = Client(api_key=values["api_key"]) + return values + + def __init__(self, **data: Any): + super().__init__(**data) + + def _run( + self, + run_manager: Optional[CallbackManagerForToolRun] = None, + ) -> dict: + if not self.client: + raise ValueError("Client not initialized") + return self.client.monitor.list() + + async def _arun( + self, + run_manager: Optional[AsyncCallbackManagerForToolRun] = None, + ) -> dict: + return self._run(run_manager=run_manager.get_sync() if run_manager else None) + + +class MonitorGetTool(BaseTool): + """Tool for getting a specific monitor.""" + + name: str = "MonitorGet" + description: str = "Get details of a specific monitor by its ID." + args_schema: Type[BaseModel] = MonitorIdInput + return_direct: bool = True + client: Optional[Client] = None + api_key: str + + @model_validator(mode="before") + @classmethod + def validate_environment(cls, values: Dict) -> Dict: + values["api_key"] = get_from_dict_or_env(values, "api_key", "SGAI_API_KEY") + values["client"] = Client(api_key=values["api_key"]) + return values + + def __init__(self, **data: Any): + super().__init__(**data) + + def _run( + self, + monitor_id: str, + run_manager: Optional[CallbackManagerForToolRun] = None, + ) -> dict: + if not self.client: + raise ValueError("Client not initialized") + return self.client.monitor.get(monitor_id) + + async def _arun( + self, + monitor_id: str, + run_manager: Optional[AsyncCallbackManagerForToolRun] = None, + ) -> dict: + return self._run( + monitor_id=monitor_id, + run_manager=run_manager.get_sync() if run_manager else None, + ) + + +class MonitorPauseTool(BaseTool): + """Tool for pausing a monitor.""" + + name: str = "MonitorPause" + description: str = "Pause a monitor so it stops running until resumed." + args_schema: Type[BaseModel] = MonitorIdInput + return_direct: bool = True + client: Optional[Client] = None + api_key: str + + @model_validator(mode="before") + @classmethod + def validate_environment(cls, values: Dict) -> Dict: + values["api_key"] = get_from_dict_or_env(values, "api_key", "SGAI_API_KEY") + values["client"] = Client(api_key=values["api_key"]) + return values + + def __init__(self, **data: Any): + super().__init__(**data) + + def _run( + self, + monitor_id: str, + run_manager: Optional[CallbackManagerForToolRun] = None, + ) -> dict: + if not self.client: + raise ValueError("Client not initialized") + return self.client.monitor.pause(monitor_id) + + async def _arun( + self, + monitor_id: str, + run_manager: Optional[AsyncCallbackManagerForToolRun] = None, + ) -> dict: + return self._run( + monitor_id=monitor_id, + run_manager=run_manager.get_sync() if run_manager else None, + ) + + +class MonitorResumeTool(BaseTool): + """Tool for resuming a paused monitor.""" + + name: str = "MonitorResume" + description: str = "Resume a paused monitor so it starts running again." + args_schema: Type[BaseModel] = MonitorIdInput + return_direct: bool = True + client: Optional[Client] = None + api_key: str + + @model_validator(mode="before") + @classmethod + def validate_environment(cls, values: Dict) -> Dict: + values["api_key"] = get_from_dict_or_env(values, "api_key", "SGAI_API_KEY") + values["client"] = Client(api_key=values["api_key"]) + return values + + def __init__(self, **data: Any): + super().__init__(**data) + + def _run( + self, + monitor_id: str, + run_manager: Optional[CallbackManagerForToolRun] = None, + ) -> dict: + if not self.client: + raise ValueError("Client not initialized") + return self.client.monitor.resume(monitor_id) + + async def _arun( + self, + monitor_id: str, + run_manager: Optional[AsyncCallbackManagerForToolRun] = None, + ) -> dict: + return self._run( + monitor_id=monitor_id, + run_manager=run_manager.get_sync() if run_manager else None, + ) + + +class MonitorDeleteTool(BaseTool): + """Tool for deleting a monitor.""" + + name: str = "MonitorDelete" + description: str = "Delete a monitor permanently." + args_schema: Type[BaseModel] = MonitorIdInput + return_direct: bool = True + client: Optional[Client] = None + api_key: str + + @model_validator(mode="before") + @classmethod + def validate_environment(cls, values: Dict) -> Dict: + values["api_key"] = get_from_dict_or_env(values, "api_key", "SGAI_API_KEY") + values["client"] = Client(api_key=values["api_key"]) + return values + + def __init__(self, **data: Any): + super().__init__(**data) + + def _run( + self, + monitor_id: str, + run_manager: Optional[CallbackManagerForToolRun] = None, + ) -> dict: + if not self.client: + raise ValueError("Client not initialized") + return self.client.monitor.delete(monitor_id) + + async def _arun( + self, + monitor_id: str, + run_manager: Optional[AsyncCallbackManagerForToolRun] = None, + ) -> dict: + return self._run( + monitor_id=monitor_id, + run_manager=run_manager.get_sync() if run_manager else None, + ) diff --git a/langchain_scrapegraph/tools/scheduled_jobs.py b/langchain_scrapegraph/tools/scheduled_jobs.py deleted file mode 100644 index bf36fa7..0000000 --- a/langchain_scrapegraph/tools/scheduled_jobs.py +++ /dev/null @@ -1,520 +0,0 @@ -from typing import Any, Dict, Optional, Type - -from langchain_core.callbacks import ( - AsyncCallbackManagerForToolRun, - CallbackManagerForToolRun, -) -from langchain_core.tools import BaseTool -from langchain_core.utils import get_from_dict_or_env -from pydantic import BaseModel, Field, model_validator -from scrapegraph_py import Client - - -class ServiceType: - """Service types for scheduled jobs.""" - - SMARTSCRAPER = "smartscraper" - SEARCHSCRAPER = "searchscraper" - SMARTCRAWLER = "smartcrawler" - MARKDOWNIFY = "markdownify" - - -class CreateScheduledJobInput(BaseModel): - job_name: str = Field(description="Name of the scheduled job") - service_type: str = Field( - description="Type of service to run (smartscraper, searchscraper, smartcrawler, markdownify)" - ) - cron_expression: str = Field( - description="Cron expression for scheduling (e.g., '0 9 * * *' for daily at 9 AM)" - ) - job_config: Dict[str, Any] = Field( - description="Configuration dictionary for the job (varies by service type)" - ) - is_active: bool = Field( - default=True, description="Whether the job should be active" - ) - - -class GetScheduledJobsInput(BaseModel): - page: int = Field(default=1, description="Page number for pagination") - page_size: int = Field(default=10, description="Number of jobs per page") - service_type: Optional[str] = Field( - default=None, - description="Filter by service type (smartscraper, searchscraper, etc.)", - ) - is_active: Optional[bool] = Field( - default=None, description="Filter by active status" - ) - - -class GetScheduledJobInput(BaseModel): - job_id: str = Field(description="ID of the scheduled job to retrieve") - - -class UpdateScheduledJobInput(BaseModel): - job_id: str = Field(description="ID of the scheduled job to update") - job_name: Optional[str] = Field(default=None, description="New job name") - cron_expression: Optional[str] = Field( - default=None, description="New cron expression" - ) - job_config: Optional[Dict[str, Any]] = Field( - default=None, description="New job configuration" - ) - is_active: Optional[bool] = Field(default=None, description="New active status") - - -class JobControlInput(BaseModel): - job_id: str = Field(description="ID of the scheduled job") - - -class GetJobExecutionsInput(BaseModel): - job_id: str = Field(description="ID of the scheduled job") - page: int = Field(default=1, description="Page number for pagination") - page_size: int = Field(default=10, description="Number of executions per page") - - -class CreateScheduledJobTool(BaseTool): - """Tool for creating scheduled jobs with ScrapeGraph AI. - - This tool allows you to create recurring jobs that will automatically - run at specified intervals using cron expressions. - """ - - name: str = "CreateScheduledJob" - description: str = ( - "Create a new scheduled job that will run automatically at specified intervals. " - "Supports SmartScraper, SearchScraper, SmartCrawler, and Markdownify services." - ) - args_schema: Type[BaseModel] = CreateScheduledJobInput - return_direct: bool = True - client: Optional[Client] = None - api_key: str - - @model_validator(mode="before") - @classmethod - def validate_environment(cls, values: Dict) -> Dict: - """Validate that api key exists in environment.""" - values["api_key"] = get_from_dict_or_env(values, "api_key", "SGAI_API_KEY") - values["client"] = Client(api_key=values["api_key"]) - return values - - def _run( - self, - job_name: str, - service_type: str, - cron_expression: str, - job_config: Dict[str, Any], - is_active: bool = True, - run_manager: Optional[CallbackManagerForToolRun] = None, - ) -> dict: - """Create a scheduled job.""" - if not self.client: - raise ValueError("Client not initialized") - - response = self.client.create_scheduled_job( - job_name=job_name, - service_type=service_type, - cron_expression=cron_expression, - job_config=job_config, - is_active=is_active, - ) - return response - - async def _arun( - self, - job_name: str, - service_type: str, - cron_expression: str, - job_config: Dict[str, Any], - is_active: bool = True, - run_manager: Optional[AsyncCallbackManagerForToolRun] = None, - ) -> dict: - """Create a scheduled job asynchronously.""" - return self._run( - job_name=job_name, - service_type=service_type, - cron_expression=cron_expression, - job_config=job_config, - is_active=is_active, - run_manager=run_manager.get_sync() if run_manager else None, - ) - - -class GetScheduledJobsTool(BaseTool): - """Tool for retrieving scheduled jobs from ScrapeGraph AI.""" - - name: str = "GetScheduledJobs" - description: str = ( - "Retrieve a list of scheduled jobs with optional filtering by service type and active status." - ) - args_schema: Type[BaseModel] = GetScheduledJobsInput - return_direct: bool = True - client: Optional[Client] = None - api_key: str - - @model_validator(mode="before") - @classmethod - def validate_environment(cls, values: Dict) -> Dict: - """Validate that api key exists in environment.""" - values["api_key"] = get_from_dict_or_env(values, "api_key", "SGAI_API_KEY") - values["client"] = Client(api_key=values["api_key"]) - return values - - def _run( - self, - page: int = 1, - page_size: int = 10, - service_type: Optional[str] = None, - is_active: Optional[bool] = None, - run_manager: Optional[CallbackManagerForToolRun] = None, - ) -> dict: - """Get scheduled jobs.""" - if not self.client: - raise ValueError("Client not initialized") - - response = self.client.get_scheduled_jobs( - page=page, - page_size=page_size, - service_type=service_type, - is_active=is_active, - ) - return response - - async def _arun( - self, - page: int = 1, - page_size: int = 10, - service_type: Optional[str] = None, - is_active: Optional[bool] = None, - run_manager: Optional[AsyncCallbackManagerForToolRun] = None, - ) -> dict: - """Get scheduled jobs asynchronously.""" - return self._run( - page=page, - page_size=page_size, - service_type=service_type, - is_active=is_active, - run_manager=run_manager.get_sync() if run_manager else None, - ) - - -class GetScheduledJobTool(BaseTool): - """Tool for retrieving a specific scheduled job by ID.""" - - name: str = "GetScheduledJob" - description: str = "Retrieve details of a specific scheduled job by its ID." - args_schema: Type[BaseModel] = GetScheduledJobInput - return_direct: bool = True - client: Optional[Client] = None - api_key: str - - @model_validator(mode="before") - @classmethod - def validate_environment(cls, values: Dict) -> Dict: - """Validate that api key exists in environment.""" - values["api_key"] = get_from_dict_or_env(values, "api_key", "SGAI_API_KEY") - values["client"] = Client(api_key=values["api_key"]) - return values - - def _run( - self, - job_id: str, - run_manager: Optional[CallbackManagerForToolRun] = None, - ) -> dict: - """Get a specific scheduled job.""" - if not self.client: - raise ValueError("Client not initialized") - - response = self.client.get_scheduled_job(job_id) - return response - - async def _arun( - self, - job_id: str, - run_manager: Optional[AsyncCallbackManagerForToolRun] = None, - ) -> dict: - """Get a specific scheduled job asynchronously.""" - return self._run( - job_id=job_id, - run_manager=run_manager.get_sync() if run_manager else None, - ) - - -class UpdateScheduledJobTool(BaseTool): - """Tool for updating a scheduled job.""" - - name: str = "UpdateScheduledJob" - description: str = "Update properties of an existing scheduled job." - args_schema: Type[BaseModel] = UpdateScheduledJobInput - return_direct: bool = True - client: Optional[Client] = None - api_key: str - - @model_validator(mode="before") - @classmethod - def validate_environment(cls, values: Dict) -> Dict: - """Validate that api key exists in environment.""" - values["api_key"] = get_from_dict_or_env(values, "api_key", "SGAI_API_KEY") - values["client"] = Client(api_key=values["api_key"]) - return values - - def _run( - self, - job_id: str, - job_name: Optional[str] = None, - cron_expression: Optional[str] = None, - job_config: Optional[Dict[str, Any]] = None, - is_active: Optional[bool] = None, - run_manager: Optional[CallbackManagerForToolRun] = None, - ) -> dict: - """Update a scheduled job.""" - if not self.client: - raise ValueError("Client not initialized") - - response = self.client.update_scheduled_job( - job_id=job_id, - job_name=job_name, - cron_expression=cron_expression, - job_config=job_config, - is_active=is_active, - ) - return response - - async def _arun( - self, - job_id: str, - job_name: Optional[str] = None, - cron_expression: Optional[str] = None, - job_config: Optional[Dict[str, Any]] = None, - is_active: Optional[bool] = None, - run_manager: Optional[AsyncCallbackManagerForToolRun] = None, - ) -> dict: - """Update a scheduled job asynchronously.""" - return self._run( - job_id=job_id, - job_name=job_name, - cron_expression=cron_expression, - job_config=job_config, - is_active=is_active, - run_manager=run_manager.get_sync() if run_manager else None, - ) - - -class PauseScheduledJobTool(BaseTool): - """Tool for pausing a scheduled job.""" - - name: str = "PauseScheduledJob" - description: str = "Pause a scheduled job so it won't run until resumed." - args_schema: Type[BaseModel] = JobControlInput - return_direct: bool = True - client: Optional[Client] = None - api_key: str - - @model_validator(mode="before") - @classmethod - def validate_environment(cls, values: Dict) -> Dict: - """Validate that api key exists in environment.""" - values["api_key"] = get_from_dict_or_env(values, "api_key", "SGAI_API_KEY") - values["client"] = Client(api_key=values["api_key"]) - return values - - def _run( - self, - job_id: str, - run_manager: Optional[CallbackManagerForToolRun] = None, - ) -> dict: - """Pause a scheduled job.""" - if not self.client: - raise ValueError("Client not initialized") - - response = self.client.pause_scheduled_job(job_id) - return response - - async def _arun( - self, - job_id: str, - run_manager: Optional[AsyncCallbackManagerForToolRun] = None, - ) -> dict: - """Pause a scheduled job asynchronously.""" - return self._run( - job_id=job_id, - run_manager=run_manager.get_sync() if run_manager else None, - ) - - -class ResumeScheduledJobTool(BaseTool): - """Tool for resuming a paused scheduled job.""" - - name: str = "ResumeScheduledJob" - description: str = "Resume a paused scheduled job so it will start running again." - args_schema: Type[BaseModel] = JobControlInput - return_direct: bool = True - client: Optional[Client] = None - api_key: str - - @model_validator(mode="before") - @classmethod - def validate_environment(cls, values: Dict) -> Dict: - """Validate that api key exists in environment.""" - values["api_key"] = get_from_dict_or_env(values, "api_key", "SGAI_API_KEY") - values["client"] = Client(api_key=values["api_key"]) - return values - - def _run( - self, - job_id: str, - run_manager: Optional[CallbackManagerForToolRun] = None, - ) -> dict: - """Resume a scheduled job.""" - if not self.client: - raise ValueError("Client not initialized") - - response = self.client.resume_scheduled_job(job_id) - return response - - async def _arun( - self, - job_id: str, - run_manager: Optional[AsyncCallbackManagerForToolRun] = None, - ) -> dict: - """Resume a scheduled job asynchronously.""" - return self._run( - job_id=job_id, - run_manager=run_manager.get_sync() if run_manager else None, - ) - - -class TriggerScheduledJobTool(BaseTool): - """Tool for manually triggering a scheduled job.""" - - name: str = "TriggerScheduledJob" - description: str = "Manually trigger a scheduled job to run immediately." - args_schema: Type[BaseModel] = JobControlInput - return_direct: bool = True - client: Optional[Client] = None - api_key: str - - @model_validator(mode="before") - @classmethod - def validate_environment(cls, values: Dict) -> Dict: - """Validate that api key exists in environment.""" - values["api_key"] = get_from_dict_or_env(values, "api_key", "SGAI_API_KEY") - values["client"] = Client(api_key=values["api_key"]) - return values - - def _run( - self, - job_id: str, - run_manager: Optional[CallbackManagerForToolRun] = None, - ) -> dict: - """Trigger a scheduled job.""" - if not self.client: - raise ValueError("Client not initialized") - - response = self.client.trigger_scheduled_job(job_id) - return response - - async def _arun( - self, - job_id: str, - run_manager: Optional[AsyncCallbackManagerForToolRun] = None, - ) -> dict: - """Trigger a scheduled job asynchronously.""" - return self._run( - job_id=job_id, - run_manager=run_manager.get_sync() if run_manager else None, - ) - - -class GetJobExecutionsTool(BaseTool): - """Tool for getting execution history of a scheduled job.""" - - name: str = "GetJobExecutions" - description: str = "Retrieve execution history for a scheduled job." - args_schema: Type[BaseModel] = GetJobExecutionsInput - return_direct: bool = True - client: Optional[Client] = None - api_key: str - - @model_validator(mode="before") - @classmethod - def validate_environment(cls, values: Dict) -> Dict: - """Validate that api key exists in environment.""" - values["api_key"] = get_from_dict_or_env(values, "api_key", "SGAI_API_KEY") - values["client"] = Client(api_key=values["api_key"]) - return values - - def _run( - self, - job_id: str, - page: int = 1, - page_size: int = 10, - run_manager: Optional[CallbackManagerForToolRun] = None, - ) -> dict: - """Get job executions.""" - if not self.client: - raise ValueError("Client not initialized") - - response = self.client.get_job_executions( - job_id=job_id, - page=page, - page_size=page_size, - ) - return response - - async def _arun( - self, - job_id: str, - page: int = 1, - page_size: int = 10, - run_manager: Optional[AsyncCallbackManagerForToolRun] = None, - ) -> dict: - """Get job executions asynchronously.""" - return self._run( - job_id=job_id, - page=page, - page_size=page_size, - run_manager=run_manager.get_sync() if run_manager else None, - ) - - -class DeleteScheduledJobTool(BaseTool): - """Tool for deleting a scheduled job.""" - - name: str = "DeleteScheduledJob" - description: str = "Delete a scheduled job permanently." - args_schema: Type[BaseModel] = JobControlInput - return_direct: bool = True - client: Optional[Client] = None - api_key: str - - @model_validator(mode="before") - @classmethod - def validate_environment(cls, values: Dict) -> Dict: - """Validate that api key exists in environment.""" - values["api_key"] = get_from_dict_or_env(values, "api_key", "SGAI_API_KEY") - values["client"] = Client(api_key=values["api_key"]) - return values - - def _run( - self, - job_id: str, - run_manager: Optional[CallbackManagerForToolRun] = None, - ) -> dict: - """Delete a scheduled job.""" - if not self.client: - raise ValueError("Client not initialized") - - response = self.client.delete_scheduled_job(job_id) - return response - - async def _arun( - self, - job_id: str, - run_manager: Optional[AsyncCallbackManagerForToolRun] = None, - ) -> dict: - """Delete a scheduled job asynchronously.""" - return self._run( - job_id=job_id, - run_manager=run_manager.get_sync() if run_manager else None, - ) diff --git a/langchain_scrapegraph/tools/scrape.py b/langchain_scrapegraph/tools/scrape.py index 5821459..526f088 100644 --- a/langchain_scrapegraph/tools/scrape.py +++ b/langchain_scrapegraph/tools/scrape.py @@ -11,18 +11,17 @@ class ScrapeInput(BaseModel): - website_url: str = Field(description="URL of the website to scrape") - render_heavy_js: bool = Field( - default=False, - description="Whether to render heavy JavaScript content (slower but more complete)", - ) - headers: Optional[Dict[str, str]] = Field( - default=None, description="Optional headers to send with the request" + url: str = Field(description="URL of the website to scrape") + format: str = Field( + default="markdown", + description="Output format: 'markdown', 'html', 'screenshot', or 'branding'", ) class ScrapeTool(BaseTool): - """Tool for getting HTML content from websites using ScrapeGraph AI. + """Tool for scraping web pages using ScrapeGraph AI v2 API. + + Returns content in the specified format (markdown, html, screenshot, or branding). Setup: Install ``langchain-scrapegraph`` python package: @@ -47,55 +46,34 @@ class ScrapeTool(BaseTool): from langchain_scrapegraph.tools import ScrapeTool - # Will automatically get SGAI_API_KEY from environment tool = ScrapeTool() - # Or provide API key directly - tool = ScrapeTool(api_key="your-api-key") - Use the tool: .. code-block:: python - # Basic scraping + # Get markdown content result = tool.invoke({ - "website_url": "https://example.com" + "url": "https://example.com" }) - # With heavy JavaScript rendering + # Get HTML content result = tool.invoke({ - "website_url": "https://example.com", - "render_heavy_js": True + "url": "https://example.com", + "format": "html" }) - # With custom headers - result = tool.invoke({ - "website_url": "https://example.com", - "headers": { - "User-Agent": "Custom Bot 1.0", - "Accept": "text/html" - } - }) - - print(result) - # { - # "html": "...", - # "scrape_request_id": "req_123", - # "status": "success", - # "error": None - # } - Async usage: .. code-block:: python result = await tool.ainvoke({ - "website_url": "https://example.com" + "url": "https://example.com" }) """ name: str = "Scrape" description: str = ( - "Get HTML content from a website. Useful when you need to retrieve the raw HTML " - "content of a webpage, with optional heavy JavaScript rendering and custom headers." + "Scrape a web page and return its content in the specified format " + "(markdown, html, screenshot, or branding)." ) args_schema: Type[BaseModel] = ScrapeInput return_direct: bool = True @@ -105,7 +83,6 @@ class ScrapeTool(BaseTool): @model_validator(mode="before") @classmethod def validate_environment(cls, values: Dict) -> Dict: - """Validate that api key exists in environment.""" values["api_key"] = get_from_dict_or_env(values, "api_key", "SGAI_API_KEY") values["client"] = Client(api_key=values["api_key"]) return values @@ -115,34 +92,24 @@ def __init__(self, **data: Any): def _run( self, - website_url: str, - render_heavy_js: bool = False, - headers: Optional[Dict[str, str]] = None, + url: str, + format: str = "markdown", run_manager: Optional[CallbackManagerForToolRun] = None, ) -> dict: - """Use the tool to scrape HTML content from a website.""" if not self.client: raise ValueError("Client not initialized") - response = self.client.scrape( - website_url=website_url, - render_heavy_js=render_heavy_js, - headers=headers, - ) - + response = self.client.scrape(url=url, format=format) return response async def _arun( self, - website_url: str, - render_heavy_js: bool = False, - headers: Optional[Dict[str, str]] = None, + url: str, + format: str = "markdown", run_manager: Optional[AsyncCallbackManagerForToolRun] = None, ) -> dict: - """Use the tool asynchronously.""" return self._run( - website_url=website_url, - render_heavy_js=render_heavy_js, - headers=headers, + url=url, + format=format, run_manager=run_manager.get_sync() if run_manager else None, ) diff --git a/langchain_scrapegraph/tools/search.py b/langchain_scrapegraph/tools/search.py new file mode 100644 index 0000000..b53f132 --- /dev/null +++ b/langchain_scrapegraph/tools/search.py @@ -0,0 +1,115 @@ +from typing import Any, Dict, Optional, Type, TypeVar + +from langchain_core.callbacks import ( + AsyncCallbackManagerForToolRun, + CallbackManagerForToolRun, +) +from langchain_core.tools import BaseTool +from langchain_core.utils import get_from_dict_or_env +from pydantic import BaseModel, Field, model_validator +from scrapegraph_py import Client + +T = TypeVar("T", bound=BaseModel) + + +class SearchInput(BaseModel): + query: str = Field( + description="The search query describing what information to find on the web" + ) + num_results: int = Field( + default=5, + description="Number of search results to return (3-20)", + ) + + +class SearchTool(BaseTool): + """Tool for searching the web and extracting structured data using ScrapeGraph AI v2 API. + + Setup: + Install ``langchain-scrapegraph`` python package: + + .. code-block:: bash + + pip install langchain-scrapegraph + + Get your API key from ScrapeGraph AI (https://scrapegraphai.com) + and set it as an environment variable: + + .. code-block:: bash + + export SGAI_API_KEY="your-api-key" + + Key init args: + api_key: Your ScrapeGraph AI API key. If not provided, will look for SGAI_API_KEY env var. + client: Optional pre-configured ScrapeGraph client instance. + llm_output_schema: Optional Pydantic model class to structure the output. + + Instantiate: + .. code-block:: python + + from langchain_scrapegraph.tools import SearchTool + + tool = SearchTool() + + Use the tool: + .. code-block:: python + + result = tool.invoke({ + "query": "What are the key features of ChatGPT Plus?" + }) + + Async usage: + .. code-block:: python + + result = await tool.ainvoke({ + "query": "What are the key features of Product X?" + }) + """ + + name: str = "Search" + description: str = ( + "Search the web and extract structured information about a specific topic or query" + ) + args_schema: Type[BaseModel] = SearchInput + return_direct: bool = True + client: Optional[Client] = None + api_key: str + llm_output_schema: Optional[Type[BaseModel]] = None + + @model_validator(mode="before") + @classmethod + def validate_environment(cls: Type[T], values: Dict[str, Any]) -> Dict[str, Any]: + values["api_key"] = get_from_dict_or_env(values, "api_key", "SGAI_API_KEY") + values["client"] = Client(api_key=values["api_key"]) + return values + + def __init__(self, **data: Any): + super().__init__(**data) + + def _run( + self, + query: str, + num_results: int = 5, + run_manager: Optional[CallbackManagerForToolRun] = None, + ) -> Dict[str, Any]: + if not self.client: + raise ValueError("Client not initialized") + + response = self.client.search( + query=query, + num_results=num_results, + output_schema=self.llm_output_schema, + ) + return response + + async def _arun( + self, + query: str, + num_results: int = 5, + run_manager: Optional[AsyncCallbackManagerForToolRun] = None, + ) -> Dict[str, Any]: + return self._run( + query, + num_results=num_results, + run_manager=run_manager.get_sync() if run_manager else None, + ) diff --git a/langchain_scrapegraph/tools/searchscraper.py b/langchain_scrapegraph/tools/searchscraper.py deleted file mode 100644 index 9b80ef5..0000000 --- a/langchain_scrapegraph/tools/searchscraper.py +++ /dev/null @@ -1,181 +0,0 @@ -from typing import Any, Dict, Optional, Type, TypeVar - -from langchain_core.callbacks import ( - AsyncCallbackManagerForToolRun, - CallbackManagerForToolRun, -) -from langchain_core.tools import BaseTool -from langchain_core.utils import get_from_dict_or_env -from pydantic import BaseModel, Field, model_validator -from scrapegraph_py import Client - -T = TypeVar("T", bound=BaseModel) - - -class SearchScraperInput(BaseModel): - user_prompt: str = Field( - description="Prompt describing what information to search for and extract from the web" - ) - extraction_mode: bool = Field( - default=True, - description="If True, use AI extraction mode (10 credits/page). If False, use markdown mode (2 credits/page).", - ) - - -class SearchScraperTool(BaseTool): - """Tool for searching and extracting structured data from the web using ScrapeGraph AI. - - Setup: - Install ``langchain-scrapegraph`` python package: - - .. code-block:: bash - - pip install langchain-scrapegraph - - Get your API key from ScrapeGraph AI (https://scrapegraphai.com) - and set it as an environment variable: - - .. code-block:: bash - - export SGAI_API_KEY="your-api-key" - - Key init args: - api_key: Your ScrapeGraph AI API key. If not provided, will look for SGAI_API_KEY env var. - client: Optional pre-configured ScrapeGraph client instance. - llm_output_schema: Optional Pydantic model or dictionary schema to structure the output. - If provided, the tool will ensure the output conforms to this schema. - - Instantiate: - .. code-block:: python - - from langchain_scrapegraph.tools import SearchScraperTool - - # Will automatically get SGAI_API_KEY from environment - tool = SearchScraperTool() - - # Or provide API key directly - tool = SearchScraperTool(api_key="your-api-key") - - # Optionally, you can provide an output schema: - from pydantic import BaseModel, Field - from typing import List - - class ProductInfo(BaseModel): - name: str = Field(description="Product name") - features: List[str] = Field(description="List of product features") - pricing: Dict[str, Any] = Field(description="Pricing information") - - tool_with_schema = SearchScraperTool(llm_output_schema=ProductInfo) - - Use the tool: - .. code-block:: python - - result = tool.invoke({ - "user_prompt": "What are the key features and pricing of ChatGPT Plus?" - }) - - print(result) - # { - # "product": { - # "name": "ChatGPT Plus", - # "description": "Premium version of ChatGPT...", - # ... - # }, - # "features": [...], - # "pricing": {...}, - # "reference_urls": [ - # "https://openai.com/chatgpt", - # ... - # ] - # } - - Async usage: - .. code-block:: python - - result = await tool.ainvoke({ - "user_prompt": "What are the key features of Product X?" - }) - """ - - name: str = "SearchScraper" - description: str = ( - "Useful when you need to search and extract structured information from the web about a specific topic or query" - ) - args_schema: Type[BaseModel] = SearchScraperInput - return_direct: bool = True - client: Optional[Client] = None - api_key: str - llm_output_schema: Optional[Type[BaseModel]] = None - - @model_validator(mode="before") - @classmethod - def validate_environment(cls: Type[T], values: Dict[str, Any]) -> Dict[str, Any]: - """Validate that api key exists in environment.""" - values["api_key"] = get_from_dict_or_env(values, "api_key", "SGAI_API_KEY") - values["client"] = Client(api_key=values["api_key"]) - return values - - def __init__(self, **data: Any): - super().__init__(**data) - - def _run( - self, - user_prompt: str, - extraction_mode: bool = True, - run_manager: Optional[CallbackManagerForToolRun] = None, - ) -> Dict[str, Any]: - """Use the tool to search and extract data from the web. - - Args: - user_prompt: What information to search for and extract - extraction_mode: If True, use AI extraction (10 credits/page). If False, use markdown mode (2 credits/page). - run_manager: Optional callback manager - - Returns: - dict: In extraction mode, returns structured data. In markdown mode, returns markdown content. - """ - if not self.client: - raise ValueError("Client not initialized") - - # In markdown mode, we ignore the output schema since we're returning raw markdown - if not extraction_mode: - response = self.client.searchscraper( - user_prompt=user_prompt, - extraction_mode=False, - ) - return { - "markdown_content": response.get("markdown_content", ""), - "reference_urls": response.get("reference_urls", []), - } - - # In extraction mode, we can use the output schema if provided - if self.llm_output_schema is None: - response = self.client.searchscraper( - user_prompt=user_prompt, - extraction_mode=True, - ) - elif isinstance(self.llm_output_schema, type) and issubclass( - self.llm_output_schema, BaseModel - ): - response = self.client.searchscraper( - user_prompt=user_prompt, - extraction_mode=True, - output_schema=self.llm_output_schema, - ) - else: - raise ValueError("llm_output_schema must be a Pydantic model class") - - return response["result"] - - async def _arun( - self, - user_prompt: str, - extraction_mode: bool = True, - run_manager: Optional[AsyncCallbackManagerForToolRun] = None, - ) -> Dict[str, Any]: - """Use the tool asynchronously.""" - return self._run( - user_prompt, - extraction_mode=extraction_mode, - run_manager=run_manager.get_sync() if run_manager else None, - ) diff --git a/langchain_scrapegraph/tools/smartcrawler.py b/langchain_scrapegraph/tools/smartcrawler.py deleted file mode 100644 index c626850..0000000 --- a/langchain_scrapegraph/tools/smartcrawler.py +++ /dev/null @@ -1,188 +0,0 @@ -from typing import Any, Dict, Optional, Type - -from langchain_core.callbacks import ( - AsyncCallbackManagerForToolRun, - CallbackManagerForToolRun, -) -from langchain_core.tools import BaseTool -from langchain_core.utils import get_from_dict_or_env -from pydantic import BaseModel, Field, model_validator -from scrapegraph_py import Client - - -class SmartCrawlerInput(BaseModel): - prompt: str = Field( - description="Prompt describing what to extract from the websites and how to structure the output" - ) - url: str = Field(description="URL of the website to start crawling from") - cache_website: bool = Field( - default=True, - description="Whether to cache the website content for faster subsequent requests", - ) - depth: int = Field( - default=2, description="Maximum depth to crawl from the starting URL" - ) - max_pages: int = Field(default=2, description="Maximum number of pages to crawl") - same_domain_only: bool = Field( - default=True, - description="Whether to only crawl pages from the same domain as the starting URL", - ) - - -class SmartCrawlerTool(BaseTool): - """Tool for crawling and extracting structured data from multiple related webpages using ScrapeGraph AI. - - Setup: - Install ``langchain-scrapegraph`` python package: - - .. code-block:: bash - - pip install langchain-scrapegraph - - Get your API key from ScrapeGraph AI (https://scrapegraphai.com) - and set it as an environment variable: - - .. code-block:: bash - - export SGAI_API_KEY="your-api-key" - - Key init args: - api_key: Your ScrapeGraph AI API key. If not provided, will look for SGAI_API_KEY env var. - client: Optional pre-configured ScrapeGraph client instance. - llm_output_schema: Optional Pydantic model or dictionary schema to structure the output. - If provided, the tool will ensure the output conforms to this schema. - - Instantiate: - .. code-block:: python - - from langchain_scrapegraph.tools import SmartCrawlerTool - - # Will automatically get SGAI_API_KEY from environment - tool = SmartCrawlerTool() - - # Or provide API key directly - tool = SmartCrawlerTool(api_key="your-api-key") - - # Optionally, you can provide an output schema: - from pydantic import BaseModel, Field - - class CompanyInfo(BaseModel): - company_description: str = Field(description="What the company does") - privacy_policy: str = Field(description="Privacy policy content") - terms_of_service: str = Field(description="Terms of service content") - - tool_with_schema = SmartCrawlerTool(llm_output_schema=CompanyInfo) - - Use the tool: - .. code-block:: python - - # Basic crawling - result = tool.invoke({ - "prompt": "What does the company do? Extract privacy and terms content", - "url": "https://scrapegraphai.com/", - "depth": 2, - "max_pages": 5 - }) - - # Crawling with custom parameters - result = tool.invoke({ - "prompt": "Extract product information and pricing", - "url": "https://example.com/products", - "cache_website": False, - "depth": 3, - "max_pages": 10, - "same_domain_only": False - }) - - print(result) - - Async usage: - .. code-block:: python - - result = await tool.ainvoke({ - "prompt": "Extract company information", - "url": "https://example.com" - }) - """ - - name: str = "SmartCrawler" - description: str = ( - "Useful when you need to extract structured data from multiple related webpages by crawling through a website, applying LLM reasoning across pages, by providing a starting URL and extraction prompt" - ) - args_schema: Type[BaseModel] = SmartCrawlerInput - return_direct: bool = True - client: Optional[Client] = None - api_key: str - llm_output_schema: Optional[Type[BaseModel]] = None - - @model_validator(mode="before") - @classmethod - def validate_environment(cls, values: Dict) -> Dict: - """Validate that api key exists in environment.""" - values["api_key"] = get_from_dict_or_env(values, "api_key", "SGAI_API_KEY") - values["client"] = Client(api_key=values["api_key"]) - return values - - def __init__(self, **data: Any): - super().__init__(**data) - - def _run( - self, - prompt: str, - url: str, - cache_website: bool = True, - depth: int = 2, - max_pages: int = 2, - same_domain_only: bool = True, - run_manager: Optional[CallbackManagerForToolRun] = None, - ) -> dict: - """Use the tool to crawl and extract data from multiple webpages.""" - if not self.client: - raise ValueError("Client not initialized") - - if self.llm_output_schema is None: - response = self.client.crawl( - url=url, - prompt=prompt, - cache_website=cache_website, - depth=depth, - max_pages=max_pages, - same_domain_only=same_domain_only, - ) - elif isinstance(self.llm_output_schema, type) and issubclass( - self.llm_output_schema, BaseModel - ): - response = self.client.crawl( - url=url, - prompt=prompt, - cache_website=cache_website, - depth=depth, - max_pages=max_pages, - same_domain_only=same_domain_only, - output_schema=self.llm_output_schema, - ) - else: - raise ValueError("llm_output_schema must be a Pydantic model class") - - return response - - async def _arun( - self, - prompt: str, - url: str, - cache_website: bool = True, - depth: int = 2, - max_pages: int = 2, - same_domain_only: bool = True, - run_manager: Optional[AsyncCallbackManagerForToolRun] = None, - ) -> dict: - """Use the tool asynchronously.""" - return self._run( - prompt, - url, - cache_website=cache_website, - depth=depth, - max_pages=max_pages, - same_domain_only=same_domain_only, - run_manager=run_manager.get_sync() if run_manager else None, - ) diff --git a/langchain_scrapegraph/tools/smartscraper.py b/langchain_scrapegraph/tools/smartscraper.py deleted file mode 100644 index 11334a6..0000000 --- a/langchain_scrapegraph/tools/smartscraper.py +++ /dev/null @@ -1,198 +0,0 @@ -from typing import Any, Dict, Optional, Type, TypeVar - -from langchain_core.callbacks import ( - AsyncCallbackManagerForToolRun, - CallbackManagerForToolRun, -) -from langchain_core.tools import BaseTool -from langchain_core.utils import get_from_dict_or_env -from pydantic import BaseModel, Field, model_validator -from scrapegraph_py import Client - -T = TypeVar("T", bound=BaseModel) - - -class SmartScraperInput(BaseModel): - user_prompt: str = Field( - description="Prompt describing what to extract from the webpage and how to structure the output" - ) - website_url: str = Field(description="Url of the webpage to extract data from") - website_html: Optional[str] = Field( - default=None, - description="Optional HTML content to process instead of fetching from website_url", - ) - render_heavy_js: bool = Field( - default=False, - description="If True, enables full browser rendering for JavaScript-heavy websites", - ) - - -class SmartScraperTool(BaseTool): - """Tool for extracting structured data from websites using ScrapeGraph AI. - - Setup: - Install ``langchain-scrapegraph`` python package: - - .. code-block:: bash - - pip install langchain-scrapegraph - - Get your API key from ScrapeGraph AI (https://scrapegraphai.com) - and set it as an environment variable: - - .. code-block:: bash - - export SGAI_API_KEY="your-api-key" - - Key init args: - api_key: Your ScrapeGraph AI API key. If not provided, will look for SGAI_API_KEY env var. - client: Optional pre-configured ScrapeGraph client instance. - llm_output_schema: Optional Pydantic model or dictionary schema to structure the output. - If provided, the tool will ensure the output conforms to this schema. - - Instantiate: - .. code-block:: python - - from langchain_scrapegraph.tools import SmartScraperTool - - # Will automatically get SGAI_API_KEY from environment - tool = SmartScraperTool() - - # Or provide API key directly - tool = SmartScraperTool(api_key="your-api-key") - - # Optionally, you can provide an output schema: - from pydantic import BaseModel, Field - - class WebsiteInfo(BaseModel): - title: str = Field(description="The main title") - description: str = Field(description="The main description") - - tool_with_schema = SmartScraperTool(llm_output_schema=WebsiteInfo) - - Use the tool: - .. code-block:: python - - # Using website URL - result = tool.invoke({ - "user_prompt": "Extract the main heading and first paragraph", - "website_url": "https://example.com" - }) - - # Using HTML content directly - html_content = ''' - - -This domain is for use in illustrative examples...
- - - ''' - result = tool.invoke({ - "user_prompt": "Extract the main heading and first paragraph", - "website_url": "https://example.com", - "website_html": html_content # This will override website_url - }) - - print(result) - # Without schema: - # { - # "main_heading": "Example Domain", - # "first_paragraph": "This domain is for use in illustrative examples..." - # } - # - # With WebsiteInfo schema: - # { - # "title": "Example Domain", - # "description": "This domain is for use in illustrative examples..." - # } - - Async usage: - .. code-block:: python - - result = await tool.ainvoke({ - "user_prompt": "Extract the main heading", - "website_url": "https://example.com" - }) - """ - - name: str = "SmartScraper" - description: str = ( - "Useful when you need to extract structured data from a webpage, applying also some reasoning using LLM, by providing a webpage URL and an extraction prompt" - ) - args_schema: Type[BaseModel] = SmartScraperInput - return_direct: bool = True - client: Optional[Client] = None - api_key: str - llm_output_schema: Optional[Type[BaseModel]] = None - - @model_validator(mode="before") - @classmethod - def validate_environment(cls: Type[T], values: Dict[str, Any]) -> Dict[str, Any]: - """Validate that api key exists in environment.""" - values["api_key"] = get_from_dict_or_env(values, "api_key", "SGAI_API_KEY") - values["client"] = Client(api_key=values["api_key"]) - return values - - def __init__(self, **data: Any): - super().__init__(**data) - - def _run( - self, - user_prompt: str, - website_url: str, - website_html: Optional[str] = None, - render_heavy_js: bool = False, - run_manager: Optional[CallbackManagerForToolRun] = None, - ) -> Dict[str, Any]: - """Use the tool to extract data from a website. - - Args: - user_prompt: What to extract from the webpage - website_url: URL to scrape - website_html: Optional HTML content to process instead of fetching from URL - render_heavy_js: If True, enables full browser rendering for JavaScript-heavy sites - run_manager: Optional callback manager - - Returns: - dict: Extracted data in the requested format - """ - if not self.client: - raise ValueError("Client not initialized") - - if self.llm_output_schema is None: - response = self.client.smartscraper( - website_url=website_url, - user_prompt=user_prompt, - website_html=website_html, - ) - elif isinstance(self.llm_output_schema, type) and issubclass( - self.llm_output_schema, BaseModel - ): - response = self.client.smartscraper( - website_url=website_url, - user_prompt=user_prompt, - website_html=website_html, - output_schema=self.llm_output_schema, - ) - else: - raise ValueError("llm_output_schema must be a Pydantic model class") - - return response["result"] - - async def _arun( - self, - user_prompt: str, - website_url: str, - website_html: Optional[str] = None, - render_heavy_js: bool = False, - run_manager: Optional[AsyncCallbackManagerForToolRun] = None, - ) -> Dict[str, Any]: - """Use the tool asynchronously.""" - return self._run( - user_prompt, - website_url, - website_html=website_html, - render_heavy_js=render_heavy_js, - run_manager=run_manager.get_sync() if run_manager else None, - ) diff --git a/pyproject.toml b/pyproject.toml index 3199205..cf277fc 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [tool.poetry] name = "langchain-scrapegraph" -version = "1.11.0" +version = "2.0.0" description = "Library for extracting structured data from websites using ScrapeGraphAI" authors = ["Marco VinciguerraTest content
", - "error": None, + "monitors": [ + { + "id": "monitor-123", + "name": "Test Monitor", + "url": "https://example.com", + "status": "active", + } + ] } - def create_scheduled_job( - self, - job_name: str, - service_type: str, - cron_expression: str, - job_config: dict, - is_active: bool = True, - ) -> dict: - """Mock create_scheduled_job method""" + def get(self, monitor_id: str) -> dict: return { - "id": "test-job-id-123", - "job_name": job_name, - "service_type": service_type, - "cron_expression": cron_expression, - "job_config": job_config, - "is_active": is_active, - "created_at": "2024-01-01T00:00:00Z", - "next_run_at": "2024-01-02T09:00:00Z", + "id": monitor_id, + "name": "Test Monitor", + "url": "https://example.com", + "status": "active", } - def get_scheduled_jobs( - self, - page: int = 1, - page_size: int = 10, - service_type: str = None, - is_active: bool = None, - ) -> dict: - """Mock get_scheduled_jobs method""" - jobs = [ - { - "id": "test-job-1", - "job_name": "Test Job 1", - "service_type": "smartscraper", - "cron_expression": "0 9 * * *", - "is_active": True, - "created_at": "2024-01-01T00:00:00Z", - "next_run_at": "2024-01-02T09:00:00Z", - } - ] + def pause(self, monitor_id: str) -> dict: + return {"id": monitor_id, "status": "paused"} - # Apply filters - if service_type: - jobs = [job for job in jobs if job["service_type"] == service_type] - if is_active is not None: - jobs = [job for job in jobs if job["is_active"] == is_active] + def resume(self, monitor_id: str) -> dict: + return {"id": monitor_id, "status": "active"} - return {"jobs": jobs, "total": len(jobs), "page": page, "page_size": page_size} + def delete(self, monitor_id: str) -> dict: + return {"message": "Monitor deleted successfully", "id": monitor_id} - def get_scheduled_job(self, job_id: str) -> dict: - """Mock get_scheduled_job method""" - return { - "id": job_id, - "job_name": "Test Job", - "service_type": "smartscraper", - "cron_expression": "0 9 * * *", - "job_config": {"website_url": "https://example.com", "user_prompt": "test"}, - "is_active": True, - "created_at": "2024-01-01T00:00:00Z", - "updated_at": "2024-01-01T00:00:00Z", - "next_run_at": "2024-01-02T09:00:00Z", - } - def update_scheduled_job( +class MockClient: + def __init__(self, api_key: str = None, *args, **kwargs): + self._api_key = api_key + self.crawl = MockCrawlNamespace(self) + self.monitor = MockMonitorNamespace(self) + + def extract( self, - job_id: str, - job_name: str = None, - cron_expression: str = None, - job_config: dict = None, - is_active: bool = None, + url: str, + prompt: str, + output_schema=None, + fetch_config=None, + llm_config=None, ) -> dict: - """Mock update_scheduled_job method""" return { - "id": job_id, - "job_name": job_name or "Updated Test Job", - "service_type": "smartscraper", - "cron_expression": cron_expression or "0 8 * * *", - "is_active": is_active if is_active is not None else True, - "updated_at": "2024-01-01T00:00:00Z", + "id": "test-id", + "data": { + "main_heading": "Example Domain", + "first_paragraph": "Test paragraph", + }, } - def pause_scheduled_job(self, job_id: str) -> dict: - """Mock pause_scheduled_job method""" + def search( + self, + query: str, + num_results: int = 5, + output_schema=None, + llm_config=None, + ) -> dict: return { - "message": "Job paused successfully", - "job_id": job_id, - "is_active": False, + "id": "test-id", + "results": [ + { + "title": "Test Result", + "url": "https://example.com", + "content": "Test content", + } + ], } - def resume_scheduled_job(self, job_id: str) -> dict: - """Mock resume_scheduled_job method""" + def scrape(self, url: str, format: str = "markdown", fetch_config=None) -> dict: + if format == "html": + return { + "id": "test-id", + "format": "html", + "content": ["Test content
", - "error": None, + "id": "test-id", + "format": "markdown", + "content": ["# Example Domain\n\nTest paragraph"], } -class MockCreateScheduledJobInput(BaseModel): - job_name: str = Field(description="Test job name") - service_type: str = Field(description="Test service type") - cron_expression: str = Field(description="Test cron expression") - job_config: Dict[str, Any] = Field(description="Test job config") - is_active: bool = Field(default=True, description="Test active status") +class MockCrawlStartInput(BaseModel): + url: str = Field(description="Test URL") + depth: int = Field(default=2, description="Crawl depth") + max_pages: int = Field(default=10, description="Max pages") + format: str = Field(default="markdown", description="Output format") -class MockCreateScheduledJobTool(BaseTool): - name: str = "CreateScheduledJob" +class MockCrawlStartTool(BaseTool): + name: str = "CrawlStart" description: str = "Test description" - args_schema: type[BaseModel] = MockCreateScheduledJobInput + args_schema: type[BaseModel] = MockCrawlStartInput client: Optional[MockClient] = None api_key: str def _run(self, **kwargs: Any) -> Dict: return { - "id": "test-job-id-123", - "job_name": kwargs.get("job_name", "Test Job"), - "service_type": kwargs.get("service_type", "smartscraper"), - "cron_expression": kwargs.get("cron_expression", "0 9 * * *"), - "job_config": kwargs.get("job_config", {}), - "is_active": kwargs.get("is_active", True), - "created_at": "2024-01-01T00:00:00Z", - "next_run_at": "2024-01-02T09:00:00Z", + "id": "crawl-job-123", + "status": "running", + "url": kwargs.get("url", "https://example.com"), } -class MockGetScheduledJobsInput(BaseModel): - page: int = Field(default=1, description="Test page") - page_size: int = Field(default=10, description="Test page size") - service_type: Optional[str] = Field(default=None, description="Test service type") - is_active: Optional[bool] = Field(default=None, description="Test active status") +class MockMonitorCreateInput(BaseModel): + name: str = Field(description="Monitor name") + url: str = Field(description="Monitor URL") + prompt: str = Field(description="Monitor prompt") + cron: str = Field(description="Cron expression") -class MockGetScheduledJobsTool(BaseTool): - name: str = "GetScheduledJobs" +class MockMonitorCreateTool(BaseTool): + name: str = "MonitorCreate" description: str = "Test description" - args_schema: type[BaseModel] = MockGetScheduledJobsInput + args_schema: type[BaseModel] = MockMonitorCreateInput client: Optional[MockClient] = None api_key: str def _run(self, **kwargs: Any) -> Dict: - jobs = [ - { - "id": "test-job-1", - "job_name": "Test Job 1", - "service_type": "smartscraper", - "cron_expression": "0 9 * * *", - "is_active": True, - "created_at": "2024-01-01T00:00:00Z", - "next_run_at": "2024-01-02T09:00:00Z", - } - ] - return {"jobs": jobs, "total": len(jobs), "page": 1, "page_size": 10} + return { + "id": "monitor-123", + "name": kwargs.get("name", "Test Monitor"), + "url": kwargs.get("url", "https://example.com"), + "status": "active", + } diff --git a/tests/unit_tests/test_tools.py b/tests/unit_tests/test_tools.py index 55f7d6c..0feb911 100644 --- a/tests/unit_tests/test_tools.py +++ b/tests/unit_tests/test_tools.py @@ -4,101 +4,82 @@ from langchain_tests.unit_tests import ToolsUnitTests from langchain_scrapegraph.tools import ( - CreateScheduledJobTool, + CrawlStartTool, + ExtractTool, GetCreditsTool, - GetScheduledJobsTool, MarkdownifyTool, + MonitorCreateTool, ScrapeTool, - SearchScraperTool, - SmartScraperTool, + SearchTool, ) from tests.unit_tests.mocks import ( MockClient, - MockCreateScheduledJobTool, + MockCrawlStartTool, + MockExtractTool, MockGetCreditsTool, - MockGetScheduledJobsTool, MockMarkdownifyTool, + MockMonitorCreateTool, MockScrapeTool, - MockSearchScraperTool, - MockSmartScraperTool, + MockSearchTool, ) -class TestSmartScraperToolUnit(ToolsUnitTests): +class TestExtractToolUnit(ToolsUnitTests): @property - def tool_constructor(self) -> Type[SmartScraperTool]: - return MockSmartScraperTool + def tool_constructor(self) -> Type[ExtractTool]: + return MockExtractTool @property def tool_constructor_params(self) -> dict: - with patch("langchain_scrapegraph.tools.smartscraper.Client", MockClient): + with patch("langchain_scrapegraph.tools.extract.Client", MockClient): return {"api_key": "sgai-test-api-key"} @property def tool_invoke_params_example(self) -> dict: return { - "user_prompt": "Extract the main heading", - "website_url": "https://example.com", + "url": "https://example.com", + "prompt": "Extract the main heading", } -class TestSmartScraperToolCustom: - def test_invoke_with_html(self): - """Test invoking the tool with HTML content.""" - with patch("langchain_scrapegraph.tools.smartscraper.Client", MockClient): - tool = MockSmartScraperTool(api_key="sgai-test-api-key") +class TestExtractToolCustom: + def test_invoke_basic(self): + with patch("langchain_scrapegraph.tools.extract.Client", MockClient): + tool = MockExtractTool(api_key="sgai-test-api-key") result = tool.invoke( { - "user_prompt": "Extract the main heading", - "website_url": "https://example.com", - "website_html": "