firecrawl_search_component.py

from langflow.custom import Component
from langflow.io import SecretStrInput, MessageTextInput, IntInput, BoolInput, DropdownInput, StrInput, Output
from langflow.schema import Data
import httpx
from loguru import logger

class FirecrawlSearchComponent(Component):
    display_name = "Firecrawl Search"
    description = "Search the web and get full content using Firecrawl"
    icon = "flame"
    name = "FirecrawlSearch"
    
    inputs = [
        SecretStrInput(
            name="api_key",
            display_name="Firecrawl API Key",
            required=True,
            info="Your Firecrawl API key from firecrawl.dev"
        ),
        MessageTextInput(
            name="query",
            display_name="Search Query",
            info="The search query to execute",
            tool_mode=True
        ),
        IntInput(
            name="limit",
            display_name="Results Limit",
            info="Maximum number of search results",
            value=5,
            advanced=True
        ),
        StrInput(
            name="location",
            display_name="Search Location",
            info="Geographic location for search results (e.g., 'Germany', 'San Francisco')",
            advanced=True
        ),
        DropdownInput(
            name="time_range",
            display_name="Time Range",
            options=["", "qdr:h", "qdr:d", "qdr:w", "qdr:m", "qdr:y"],
            value="",
            info="Filter results by time period",
            advanced=True
        ),
        BoolInput(
            name="scrape_content",
            display_name="Scrape Full Content",
            info="Get complete page content, not just search snippets",
            value=True,
            advanced=True
        )
    ]
    
    outputs = [
        Output(
            name="search_results",
            display_name="Search Results",
            method="search_web"
        )
    ]
    
    def search_web(self) -> Data:
        try:
            url = "https://api.firecrawl.dev/v1/search"
            headers = {
                "Authorization": f"Bearer {self.api_key}",
                "Content-Type": "application/json"
            }
            
            payload = {
                "query": self.query,
                "limit": self.limit
            }
            
            if self.location:
                payload["location"] = self.location
                
            if self.time_range:
                payload["tbs"] = self.time_range
                
            if self.scrape_content:
                payload["scrapeOptions"] = {
                    "formats": ["markdown", "links"]
                }
            
            with httpx.Client(timeout=90.0) as client:
                response = client.post(url, json=payload, headers=headers)
                response.raise_for_status()
                search_results = response.json()
            
            results = []
            
            for item in search_results.get("data", []):
                result_data = {
                    "title": item.get("title", ""),
                    "url": item.get("url", ""),
                    "description": item.get("description", "")
                }
                
                if "markdown" in item:
                    result_data["content"] = item["markdown"]
                    results.append(Data(text=item["markdown"], data=result_data))
                else:
                    results.append(Data(text=item.get("description", ""), data=result_data))
            
            self.status = f"Found {len(results)} search results"
            return Data(data={"results": results, "count": len(results)})
            
        except httpx.TimeoutException:
            error_msg = "Search request timed out. Try reducing the limit or simplifying the query."
            logger.error(error_msg)
            return Data(data={"error": error_msg, "results": []})
        
        except httpx.HTTPStatusError as e:
            error_msg = f"Search failed: {e.response.status_code} - {e.response.text}"
            logger.error(error_msg)
            return Data(data={"error": error_msg, "results": []})