Outscraper MCP Server

Instructions

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