URL Text Fetcher MCP Server

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")