test_brave_search

Instructions

Implementation Reference

• src/url_text_fetcher/server.py:354-387 (handler)

  The handler function for the 'test_brave_search' tool. It tests the Brave Search API by performing a search with the provided query and returns formatted results or error messages. Registered via @mcp.tool() decorator.

  @mcp.tool()
  async def test_brave_search(query: str = "test") -> str:
      """Test the Brave Search API connection and configuration.
      
      Args:
          query: Test query to search for (default: "test")
      """
      if not BRAVE_API_KEY:
          return "❌ Error: BRAVE_API_KEY environment variable not set"
      
      try:
          logger.info(f"Testing Brave Search API with query: '{query}'")
          results = brave_search(query, count=1)
          
          if results:
              result = results[0]
              return f"""✅ Brave Search API Test Successful!
              
  Query: {query}
  Found: {len(results)} result(s)
  
  First Result:
  Title: {result.get('title', 'No title')}
  URL: {result.get('url', 'No URL')}
  Description: {result.get('description', 'No description')}
  
  API Key: ✓ Valid (length: {len(BRAVE_API_KEY)})
  Rate Limit: {BRAVE_RATE_LIMIT_RPS} requests/second"""
          else:
              return f"⚠️ API connection successful but no results found for query: '{query}'"
              
      except Exception as e:
          return f"❌ Brave Search API Test Failed: {str(e)}"

• src/url_text_fetcher/server.py:246-319 (helper)

  Supporting helper function 'brave_search' that implements the core logic for querying the Brave Search API, including rate limiting, request handling, and result parsing. Directly called by the test_brave_search handler.

  def brave_search(query: str, count: int = 10) -> List[dict]:
      """Perform a Brave search and return results with thread-safe rate limiting."""
      if not BRAVE_API_KEY:
          logger.error("Brave Search API key not configured")
          raise ValueError("BRAVE_API_KEY environment variable is required")
      
      # Thread-safe rate limiting: ensure minimum interval between requests
      with rate_limit_lock:
          current_time = time.time()
          time_since_last_request = current_time - last_brave_request[0]
          if time_since_last_request < MIN_REQUEST_INTERVAL:
              sleep_time = MIN_REQUEST_INTERVAL - time_since_last_request
              logger.info(f"Rate limiting: sleeping for {sleep_time:.3f} seconds (limit: {BRAVE_RATE_LIMIT_RPS} req/s)")
              time.sleep(sleep_time)
          last_brave_request[0] = time.time()
      
      url = "https://api.search.brave.com/res/v1/web/search"
      headers = {
          'User-Agent': 'Mozilla/5.0 (compatible; MCP-URL-Fetcher/1.0)',
          'Accept': 'application/json',  # Brave API requires application/json or */*
          "X-Subscription-Token": BRAVE_API_KEY
      }
      params = {
          "q": query,
          "count": count,
          "search_lang": "en",
          "country": "US",
          "safesearch": "moderate"
      }
      
      try:
          logger.info(f"SEARCH_REQUEST: Making Brave Search for '{query}' (count={count})")
          response = requests.get(url, headers=headers, params=params, timeout=REQUEST_TIMEOUT)
          
          # Log response details for debugging
          logger.info(f"SEARCH_RESPONSE: Status {response.status_code}, Content-Type: {response.headers.get('Content-Type', 'unknown')}")
          
          response.raise_for_status()
          data = response.json()
          
          results = []
          if 'web' in data and 'results' in data['web']:
              for result in data['web']['results']:
                  results.append({
                      'title': result.get('title', ''),
                      'url': result.get('url', ''),
                      'description': result.get('description', ''),
                  })
          else:
              logger.warning(f"Unexpected response structure: {data}")
          
          logger.info(f"SEARCH_SUCCESS: Found {len(results)} results for '{query}'")
          return results
      except requests.HTTPError as e:
          logger.error(f"Brave Search API error: {e.response.status_code}")
          logger.error(f"Response headers: {dict(e.response.headers)}")
          logger.error(f"Response body: {e.response.text}")
          if e.response.status_code == 401:
              raise Exception("Invalid or missing API key - check your BRAVE_API_KEY")
          elif e.response.status_code == 403:
              raise Exception("API key does not have permission for this endpoint")
          elif e.response.status_code == 422:
              raise Exception("Search request was rejected - please check your query")
          elif e.response.status_code == 429:
              raise Exception("Rate limit exceeded - please wait before making another request")
          else:
              raise Exception(f"Search service error: {e.response.status_code} - {e.response.text}")
      except requests.RequestException as e:
          logger.error(f"Network error during search: {e}")
          raise Exception("Network error occurred during search")
      except Exception as e:
          logger.error(f"Unexpected error in brave_search: {e}", exc_info=True)
          raise Exception("An unexpected error occurred during search")