Outscraper MCP Server

Instructions

Implementation Reference

• outscraper_mcp/server.py:228-332 (handler)

  Main handler function for the 'google_maps_search' tool. Decorated with @mcp.tool() for registration in FastMCP. Performs input validation, invokes the API client helper, handles responses (including async), and formats output for the user.
  @mcp.tool() def google_maps_search(query: str, limit: int = 20, language: str = "en", region: Optional[str] = None, drop_duplicates: bool = False, enrichment: Optional[List[str]] = None) -> str: """ Search for businesses and places on Google Maps using Outscraper Args: query: Search query (e.g., 'restaurants brooklyn usa', 'hotels paris france') limit: Number of results to return (default: 20, max: 400) language: Language code (default: 'en') region: Country/region code (e.g., 'US', 'GB', 'DE') drop_duplicates: Remove duplicate results (default: False) enrichment: Additional services to run (e.g., ['domains_service', 'emails_validator_service']) Returns: Formatted search results with business information """ # Input validation if not query or not query.strip(): return "Error: Search query cannot be empty." if limit < 1 or limit > MAX_SEARCH_LIMIT: return f"Error: Limit must be between 1 and {MAX_SEARCH_LIMIT}." if language and len(language) != 2: logger.warning(f"Unusual language code: {language}. Standard codes are 2 characters (e.g., 'en', 'es').") try: logger.info(f"Searching Google Maps for: {query}") results = client.google_maps_search( query=query, limit=limit, language=language, region=region, drop_duplicates=drop_duplicates, enrichment=enrichment ) if not results: return "No results 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 = [] if isinstance(results, list) and len(results) > 0: places = results[0] if isinstance(results[0], list) else results # Ensure places is a list if not isinstance(places, list): places = [places] for i, place in enumerate(places[:limit], 1): if isinstance(place, dict): name = place.get('name', 'Unknown') formatted_place = f"**{i}. {name}**\n" # Address address = place.get('full_address') or place.get('address', 'N/A') formatted_place += f" 📍 Address: {address}\n" # Rating and reviews rating = place.get('rating') reviews = place.get('reviews', 0) if rating: formatted_place += f" ⭐ Rating: {rating} ({reviews} reviews)\n" else: formatted_place += f" ⭐ Rating: No ratings yet\n" # Contact info phone = place.get('phone', 'N/A') formatted_place += f" 📞 Phone: {phone}\n" website = place.get('site', 'N/A') formatted_place += f" 🌐 Website: {website}\n" # Type/category place_type = place.get('type') or place.get('main_category', 'N/A') formatted_place += f" 🏷️ Type: {place_type}\n" if place.get('working_hours'): formatted_place += f" 🕒 Hours: {place.get('working_hours_old_format', 'N/A')}\n" formatted_place += f" 🆔 Place ID: {place.get('place_id', 'N/A')}\n" # Include enrichment data if available if place.get('emails'): formatted_place += f" 📧 Emails: {', '.join([email.get('value') for email in place.get('emails', [])])}\n" formatted_place += "---\n" formatted_results.append(formatted_place) return f"Found {len(places)} places for '{query}':\n\n" + "\n".join(formatted_results) else: return f"Search results for '{query}':\n\n" + str(results) except Exception as e: logger.error(f"Error in google_maps_search: {str(e)}") logger.error(f"Query: {query}, Limit: {limit}, Language: {language}") return f"Error searching Google Maps: {str(e)}"
• outscraper_mcp/server.py:133-177 (helper)

  Helper method in OutscraperClient class that constructs the API request to Outscraper's /maps/search-v3 endpoint, handles retries and errors, and parses the response using _handle_response.
  def google_maps_search(self, query: Union[List[str], str], limit: int = 20, language: str = 'en', region: str = None, drop_duplicates: bool = False, enrichment: List[str] = None) -> Union[List, Dict]: """Search Google Maps for places/businesses""" if isinstance(query, str): queries = [query] else: queries = query wait_async = len(queries) > 10 and limit > 1 params = { 'query': queries, 'language': language, 'organizationsPerQueryLimit': limit, 'async': wait_async, 'dropDuplicates': drop_duplicates } if region: params['region'] = region if enrichment: params['enrichment'] = enrichment try: response = self.session.get( f'{OUTSCRAPER_API_BASE}/maps/search-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 search") raise Exception("Request timed out. Please try again with fewer queries or lower limit.") except requests.exceptions.ConnectionError: logger.error(f"Connection error during Google Maps search") raise Exception("Connection error. Please check your internet connection.") except requests.exceptions.RequestException as e: logger.error(f"Network error during Google Maps search: {e}") raise Exception(f"Network error: {str(e)}") except Exception as e: logger.error(f"Unexpected error during Google Maps search: {e}") raise
• outscraper_mcp/server.py:40-43 (schema)

  Constants used for input validation in the google_maps_search tool, defining limits and valid options for parameters like limit, sort, language.
  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
• outscraper_mcp/server.py:29-228 (registration)

  The @mcp.tool() decorator registers the google_maps_search function as an MCP tool. The FastMCP instance is created here.
  mcp = FastMCP("outscraper-mcp") # Outscraper API configuration OUTSCRAPER_API_BASE = "https://api.app.outscraper.com" API_KEY = os.getenv("OUTSCRAPER_API_KEY") if not API_KEY: logger.warning("OUTSCRAPER_API_KEY environment variable not set. Server will start but tools will not function without a valid API key.") API_KEY = "placeholder_key_for_deployment" # Constants for validation 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 class OutscraperClient: """Outscraper API client for making requests with retry logic""" def __init__(self, api_key: str): self.api_key = api_key self.headers = { 'X-API-KEY': api_key, 'client': 'MCP Server', 'User-Agent': 'Outscraper-MCP/1.0.0' } # Configure session with retry logic self.session = requests.Session() retry_strategy = Retry( total=3, backoff_factor=1, status_forcelist=[429, 500, 502, 503, 504], allowed_methods=["GET", "POST"] ) adapter = HTTPAdapter(max_retries=retry_strategy) self.session.mount("http://", adapter) self.session.mount("https://", adapter) 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) def google_maps_search(self, query: Union[List[str], str], limit: int = 20, language: str = 'en', region: str = None, drop_duplicates: bool = False, enrichment: List[str] = None) -> Union[List, Dict]: """Search Google Maps for places/businesses""" if isinstance(query, str): queries = [query] else: queries = query wait_async = len(queries) > 10 and limit > 1 params = { 'query': queries, 'language': language, 'organizationsPerQueryLimit': limit, 'async': wait_async, 'dropDuplicates': drop_duplicates } if region: params['region'] = region if enrichment: params['enrichment'] = enrichment try: response = self.session.get( f'{OUTSCRAPER_API_BASE}/maps/search-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 search") raise Exception("Request timed out. Please try again with fewer queries or lower limit.") except requests.exceptions.ConnectionError: logger.error(f"Connection error during Google Maps search") raise Exception("Connection error. Please check your internet connection.") except requests.exceptions.RequestException as e: logger.error(f"Network error during Google Maps search: {e}") raise Exception(f"Network error: {str(e)}") except Exception as e: logger.error(f"Unexpected error during Google Maps search: {e}") raise 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 # Initialize client client = OutscraperClient(API_KEY) @mcp.tool()