crew/tools.py

from typing import List
from crewai_tools import ScrapeWebsiteTool
from pydantic import BaseModel, HttpUrl, Field
from googlesearch import search
from crewai.tools import tool
from langchain_community.tools import DuckDuckGoSearchRun
from datetime import datetime

@tool("get_current_time")
def get_current_time() -> str:
    """Returns the current date and time in a human-readable format."""
    now = datetime.now()
    return now.strftime("%A, %d %B %Y %I:%M:%S %p")

class SearchResponse(BaseModel):
    """google search response"""
    urls: List[str] = Field(default_factory=list, description="List of URLs scraped")
    content: str = Field(..., description="Formatted search results")

    
@tool("Google Search Tool")
def google_website_search_tool(query: str) -> SearchResponse:
    """Search the web for information on a given topic"""
    search_results = list(search(query, num_results=3))
    results, urls = [], []
    results.append(f"Google Search Results for: '{query}'")
    results.append("=" * 50)
    
    for i, url in enumerate(search_results, 1):
        results.append(f"\n{i}. URL: {url}")
        urls.append(str(url))  # Convert to string for simple model
        try:
            scraper = ScrapeWebsiteTool(website_url=url)
            content = scraper.run()
            
            if len(content) > 1000:
                content = content[:1000] + "... [Content truncated]"
            
            results.append(f"Content: {content}")
            results.append("-" * 30)
            
        except Exception as scrape_error:
            results.append(f"Error scraping content: {str(scrape_error)}")
            results.append("-" * 30)
    
    return SearchResponse(
        urls=urls,
        content="\n".join(results)
    )

@tool("Website Search Tool")
def website_search_tool(query: str) -> str:
    """Search the web for information on a given topic"""
    # return DuckDuckGoSearchRun().invoke(question)
    search_results = list(search(query, num_results=2))
    results, urls = [], []
    results.append(f"Google Search Results for: '{query}'")
    results.append("=" * 50)
    
    for i, url in enumerate(search_results, 1):
        results.append(f"\n{i}. URL: {url}")
        urls.append(f"{url}")
        try:
            # Use CrewAI's ScrapeWebsiteTool to get content
            scraper = ScrapeWebsiteTool(website_url=url)
            content = scraper.run()
            
            # Limit content length for readability
            if len(content) > 1000:
                content = content[:1000] + "... [Content truncated]"
            
            results.append(f"Content: {content}")
            results.append("-" * 30)
            
        except Exception as scrape_error:
            results.append(f"Error scraping content: {str(scrape_error)}")
            results.append("-" * 30)
    
    return "\n".join(results)