google_maps_reviews

Instructions

Extract reviews from Google Maps places using Outscraper

Args:
    query: Place query, place ID, or business name (e.g., 'ChIJrc9T9fpYwokRdvjYRHT8nI4', 'Memphis Seoul brooklyn usa')
    reviews_limit: Number of reviews to extract per place (default: 10, 0 for unlimited)
    limit: Number of places to process (default: 1)
    sort: Sort order for reviews ('most_relevant', 'newest', 'highest_rating', 'lowest_rating')
    language: Language code (default: 'en')
    region: Country/region code (e.g., 'US', 'GB', 'DE')
    cutoff: Unix timestamp to get only reviews after this date

Returns:
    Formatted reviews data with place information and individual reviews

Implementation Reference

• outscraper_mcp/server.py:333-431 (handler)

  Primary MCP tool handler for 'google_maps_reviews'. Includes decorator for registration, input validation, API call via client, result formatting, and error handling. The docstring provides the input schema.

  @mcp.tool()
  def google_maps_reviews(query: str, reviews_limit: int = 10, limit: int = 1, 
                         sort: str = "most_relevant", language: str = "en", 
                         region: Optional[str] = None, cutoff: Optional[int] = None) -> str:
      """
      Extract reviews from Google Maps places using Outscraper
      
      Args:
          query: Place query, place ID, or business name (e.g., 'ChIJrc9T9fpYwokRdvjYRHT8nI4', 'Memphis Seoul brooklyn usa')
          reviews_limit: Number of reviews to extract per place (default: 10, 0 for unlimited)
          limit: Number of places to process (default: 1)
          sort: Sort order for reviews ('most_relevant', 'newest', 'highest_rating', 'lowest_rating')
          language: Language code (default: 'en')
          region: Country/region code (e.g., 'US', 'GB', 'DE')
          cutoff: Unix timestamp to get only reviews after this date
      
      Returns:
          Formatted reviews data with place information and individual reviews
      """
      # Input validation
      if not query or not query.strip():
          return "Error: Query cannot be empty."
      
      if reviews_limit < 0 or reviews_limit > MAX_REVIEWS_LIMIT:
          return f"Error: Reviews limit must be between 0 and {MAX_REVIEWS_LIMIT}."
      
      if limit < 1:
          return "Error: Limit must be at least 1."
      
      if sort not in VALID_SORT_OPTIONS:
          return f"Error: Sort must be one of {VALID_SORT_OPTIONS}."
      
      if cutoff and cutoff < 0:
          return "Error: Cutoff timestamp must be positive."
      try:
          logger.info(f"Getting reviews for: {query}")
          
          results = client.google_maps_reviews(
              query=query,
              reviews_limit=reviews_limit,
              limit=limit,
              sort=sort,
              language=language,
              region=region,
              cutoff=cutoff
          )
          
          if not results:
              return "No reviews found for the given query."
          
          # Handle async response
          if isinstance(results, str) and "Request processing asynchronously" in results:
              return results
          
          # Format results for better readability
          formatted_results = []
          
          # Ensure results is a list
          if not isinstance(results, list):
              results = [results]
          
          if isinstance(results, list):
              for place_data in results:
                  if isinstance(place_data, dict):
                      # Place information
                      place_info = f"**{place_data.get('name', 'Unknown Place')}**\n"
                      place_info += f"📍 Address: {place_data.get('address', 'N/A')}\n"
                      place_info += f"⭐ Rating: {place_data.get('rating', 'N/A')} ({place_data.get('reviews', 0)} total reviews)\n"
                      place_info += f"📞 Phone: {place_data.get('phone', 'N/A')}\n"
                      place_info += f"🌐 Website: {place_data.get('site', 'N/A')}\n\n"
                      
                      formatted_results.append(place_info)
                      
                      # Reviews
                      reviews_data = place_data.get('reviews_data', [])
                      if reviews_data:
                          formatted_results.append(f"**Reviews (showing {len(reviews_data)} reviews):**\n")
                          
                          for i, review in enumerate(reviews_data, 1):
                              review_text = f"{i}. **{review.get('autor_name', 'Anonymous')}** - {review.get('review_rating', 'N/A')}⭐\n"
                              review_text += f"   📅 Date: {review.get('review_datetime_utc', 'N/A')}\n"
                              review_text += f"   💬 Review: {review.get('review_text', 'No text')[:200]}{'...' if len(review.get('review_text', '')) > 200 else ''}\n"
                              
                              if review.get('review_likes'):
                                  review_text += f"   👍 Likes: {review.get('review_likes')}\n"
                              
                              review_text += "\n"
                              formatted_results.append(review_text)
                      else:
                          formatted_results.append("No reviews found.\n")
                      
                      formatted_results.append("=" * 50 + "\n")
          
          return f"Reviews for '{query}':\n\n" + "".join(formatted_results)
          
      except Exception as e:
          logger.error(f"Error in google_maps_reviews: {str(e)}")
          return f"Error getting reviews: {str(e)}"

• outscraper_mcp/server.py:178-223 (helper)

  Helper method in OutscraperClient class that constructs the API request parameters and makes the HTTP GET request to Outscraper's /maps/reviews-v3 endpoint, handling retries and responses via _handle_response.

  def google_maps_reviews(self, query: Union[List[str], str], reviews_limit: int = 10,
                         limit: int = 1, sort: str = 'most_relevant', 
                         language: str = 'en', region: str = None,
                         cutoff: int = None) -> Union[List, Dict]:
      """Get reviews from Google Maps places"""
      if isinstance(query, str):
          queries = [query]
      else:
          queries = query
          
      wait_async = reviews_limit > 499 or reviews_limit == 0 or len(queries) > 10
      
      params = {
          'query': queries,
          'reviewsLimit': reviews_limit,
          'limit': limit,
          'sort': sort,
          'language': language,
          'async': wait_async
      }
      
      if region:
          params['region'] = region
      if cutoff:
          params['cutoff'] = cutoff
          
      try:
          response = self.session.get(
              f'{OUTSCRAPER_API_BASE}/maps/reviews-v3',
              params=params,
              headers=self.headers,
              timeout=60 if wait_async else 30
          )
          return self._handle_response(response, wait_async)
      except requests.exceptions.Timeout:
          logger.error(f"Request timeout during Google Maps reviews")
          raise Exception("Request timed out. Please try again with fewer reviews or places.")
      except requests.exceptions.ConnectionError:
          logger.error(f"Connection error during Google Maps reviews")
          raise Exception("Connection error. Please check your internet connection.")
      except requests.exceptions.RequestException as e:
          logger.error(f"Network error during Google Maps reviews: {e}")
          raise Exception(f"Network error: {str(e)}")
      except Exception as e:
          logger.error(f"Unexpected error during Google Maps reviews: {e}")
          raise

• outscraper_mcp/server.py:68-132 (helper)

  Shared helper method in OutscraperClient for handling API responses, including JSON parsing, async handling, error extraction, and specific error messages for common HTTP status codes.

  def _handle_response(self, response: requests.Response, wait_async: bool = False) -> Union[List, Dict, str]:
      """Handle API response and return data
      
      Args:
          response: The HTTP response from the API
          wait_async: Whether this was an async request
          
      Returns:
          Parsed response data or formatted message for async requests
          
      Raises:
          Exception: For API errors or invalid responses
      """
      logger.info(f"API Response - Status: {response.status_code}, URL: {response.url}")
      
      if 199 < response.status_code < 300:
          try:
              response_json = response.json()
              logger.info(f"Response JSON keys: {list(response_json.keys()) if isinstance(response_json, dict) else 'Not a dict'}")
              
              # Handle async responses (status 202)
              if response.status_code == 202:
                  if isinstance(response_json, dict) and response_json.get('status') == 'Pending':
                      return f"⏳ **Request processing asynchronously**\n\n📋 **Available tools:**\n• google_maps_search\n• google_maps_reviews\n\n🔗 **Request ID:** {response_json.get('id', 'unknown')}\n📍 **Results URL:** {response_json.get('results_location', 'N/A')}\n\n💡 **Note:** This server only provides Google Maps Search and Reviews tools."
              
              if wait_async:
                  return response_json
              else:
                  # Handle different response structures
                  if isinstance(response_json, dict):
                      # Try 'data' field first, then return whole response
                      return response_json.get('data', response_json)
                  else:
                      return response_json
          except ValueError as e:
              logger.error(f"Failed to parse JSON response: {e}")
              logger.error(f"Raw response: {response.text[:500]}")
              raise Exception(f'Invalid JSON response from API: {e}')
          except Exception as e:
              logger.error(f"Unexpected error parsing response: {e}")
              logger.error(f"Raw response: {response.text[:500]}")
              raise Exception(f'Failed to parse API response: {e}')
      else:
          error_msg = f"API request failed with status {response.status_code}"
          try:
              error_json = response.json()
              if isinstance(error_json, dict):
                  if 'message' in error_json:
                      error_msg += f": {error_json['message']}"
                  elif 'error' in error_json:
                      error_msg += f": {error_json['error']}"
          except:
              error_msg += f": {response.text[:200]}"
          
          # Add specific error handling for common status codes
          if response.status_code == 401:
              error_msg = "Invalid API key. Please check your OUTSCRAPER_API_KEY."
          elif response.status_code == 429:
              error_msg = "Rate limit exceeded. Please wait before making more requests."
          elif response.status_code == 402:
              error_msg = "Insufficient credits. Please check your Outscraper account balance."
          
          logger.error(error_msg)
          raise Exception(error_msg)

• outscraper_mcp/server.py:40-43 (schema)

  Validation constants used in the tool handler for input schema enforcement, such as max limits and valid sort options.

  MAX_SEARCH_LIMIT = 400
  MAX_REVIEWS_LIMIT = 10000
  VALID_SORT_OPTIONS = ['most_relevant', 'newest', 'highest_rating', 'lowest_rating']
  VALID_LANGUAGE_CODES = ['en', 'es', 'fr', 'de', 'it', 'pt', 'ru', 'ja', 'ko', 'zh']  # Common language codes