crawl4ai-mcp

Instructions

Implementation Reference

• main.py:26-43 (handler)

  The FastMCP tool handler for 'crawl_website', registered via @mcp.tool() decorator. Defines input schema via type annotations and delegates execution to the imported crawl_website_async helper.

  @mcp.tool()
  async def crawl_website(
      url: str,
      crawl_depth: int = 1,
      max_pages: int = 5,
  ) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
      """
      Crawl a website starting from the given URL up to a specified depth and page limit.
  
      Args:
          url: The starting URL to crawl.
          crawl_depth: The maximum depth to crawl relative to the starting URL (default: 1).
          max_pages: The maximum number of pages to scrape during the crawl (default: 5).
  
      Returns:
          List containing TextContent with a JSON array of results for crawled pages.
      """
      return await crawl_website_async(url, crawl_depth, max_pages)

• tools/crawl.py:13-126 (helper)

  Core helper function implementing the website crawling logic using Crawl4AI library, including BFS deep crawling strategy, error handling, and result processing into MCP TextContent format.

  async def crawl_website_async(url: str, crawl_depth: int, max_pages: int) -> List[Any]:
      """Crawl a website using crawl4ai.
  
      Args:
          url: The starting URL to crawl.
          crawl_depth: The maximum depth to crawl.
          max_pages: The maximum number of pages to crawl.
  
      Returns:
          A list containing TextContent objects with the results as JSON.
      """
  
      normalized_url = validate_and_normalize_url(url)
      if not normalized_url:
          return [
              types.TextContent(
                  type="text",
                  text=json.dumps(
                      {
                          "success": False,
                          "url": url,
                          "error": "Invalid URL format",
                      }
                  ),
              )
          ]
  
      try:
          # Use default configurations with minimal customization
          browser_config = BrowserConfig(
              browser_type="chromium",
              headless=True,
              ignore_https_errors=True,
              verbose=False,
              extra_args=[
                  "--no-sandbox",
                  "--disable-setuid-sandbox",
                  "--disable-dev-shm-usage",
              ],
          )
  
          # 1. Create the deep crawl strategy with depth and page limits
          crawl_strategy = BFSDeepCrawlStrategy(
              max_depth=crawl_depth, max_pages=max_pages
          )
  
          # 2. Create the run config, passing the strategy
          run_config = CrawlerRunConfig(
              cache_mode=CacheMode.BYPASS,
              verbose=False,
              page_timeout=30 * 1000,  # 30 seconds per page
              deep_crawl_strategy=crawl_strategy,  # Pass the strategy here
          )
  
          results_list = []
          async with AsyncWebCrawler(config=browser_config) as crawler:
              # 3. Use arun and wrap in asyncio.wait_for for overall timeout
              crawl_results: List[CrawlResult] = await asyncio.wait_for(
                  crawler.arun(
                      url=normalized_url,
                      config=run_config,
                  ),
                  timeout=CRAWL_TIMEOUT_SECONDS,
              )
  
              # Process results, checking 'success' attribute
              for result in crawl_results:
                  if result.success:  # Check .success instead of .status
                      results_list.append(
                          {
                              "url": result.url,
                              "success": True,
                              "markdown": result.markdown,
                          }
                      )
                  else:
                      results_list.append(
                          {
                              "url": result.url,
                              "success": False,
                              "error": result.error,  # Assume .error holds the message
                          }
                      )
  
              # Return a single TextContent with a JSON array of results
              return [
                  types.TextContent(
                      type="text", text=json.dumps({"results": results_list})
                  )
              ]
  
      except asyncio.TimeoutError:
          return [
              types.TextContent(
                  type="text",
                  text=json.dumps(
                      {
                          "success": False,
                          "url": normalized_url,
                          "error": f"Crawl operation timed out after {CRAWL_TIMEOUT_SECONDS} seconds.",
                      }
                  ),
              )
          ]
      except Exception as e:
          return [
              types.TextContent(
                  type="text",
                  text=json.dumps(
                      {"success": False, "url": normalized_url, "error": str(e)}
                  ),
              )
          ]

• tools/utils.py:4-32 (helper)

  Utility function used by crawl_website_async to validate and normalize the input URL before crawling.

  def validate_and_normalize_url(url: str) -> str | None:
      """Validate and normalize a URL.
  
      Args:
          url: The URL string to validate.
  
      Returns:
          The normalized URL with https scheme if valid, otherwise None.
      """
      # Simple validation for domains/subdomains with http(s)
      # Allows for optional paths
      url_pattern = re.compile(
          r"^(?:https?://)?(?:(?:[A-Z0-9](?:[A-Z0-9-]{0,61}[A-Z0-9])?\.)+(?:[A-Z]{2,6}\.?|[A-Z0-9-]{2,}\.?)|"  # domain...
          r"localhost|"  # localhost...
          r"\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})"  # ...or ip
          r"(?::\d+)?"  # optional port
          r"(?:/?|[/?]\S+)$",
          re.IGNORECASE,
      )
  
      if not url_pattern.match(url):
          return None
  
      # Add https:// if missing
      if not url.startswith("http://") and not url.startswith("https://"):
          url = f"https://{url}"
  
      return url