diff --git a/cookbook/tools/google_maps_tools.py b/cookbook/tools/google_maps_tools.py new file mode 100644 index 0000000000..d949f123df --- /dev/null +++ b/cookbook/tools/google_maps_tools.py @@ -0,0 +1,143 @@ +""" +Business Contact Search Agent for finding and extracting business contact information. +This example demonstrates various Google Maps API functionalities including business search, +directions, geocoding, address validation, and more. + +Prerequisites: +- Set the environment variable `GOOGLE_MAPS_API_KEY` with your Google Maps API key. + You can obtain the API key from the Google Cloud Console: + https://console.cloud.google.com/projectselector2/google/maps-apis/credentials + +- You also need to activate the Address Validation API for your . + https://console.developers.google.com/apis/api/addressvalidation.googleapis.com + +""" + +from agno.agent import Agent +from agno.tools.crawl4ai import Crawl4aiTools +from agno.tools.google_maps import GoogleMapTools + +agent = Agent( + name="Maps API Demo Agent", + tools=[ + GoogleMapTools(), # For on Google Maps + Crawl4aiTools(max_length=5000), # For scraping business websites + ], + description="You are a location and business information specialist that can help with various mapping and location-based queries.", + instructions=[ + "When given a search query:", + "1. Use appropriate Google Maps methods based on the query type", + "2. For place searches, combine Maps data with website data when available", + "3. Format responses clearly and provide relevant details based on the query", + "4. Handle errors gracefully and provide meaningful feedback", + ], + markdown=True, + show_tool_calls=True, +) + +# Example 1: Business Search +print("\n=== Business Search Example ===") +agent.print_response( + "Find me highly rated Indian restaurants in Phoenix, AZ with their contact details", + markdown=True, + stream=True, +) + +# Example 2: Directions +print("\n=== Directions Example ===") +agent.print_response( + """Get driving directions from 'Phoenix Sky Harbor Airport' to 'Desert Botanical Garden', + avoiding highways if possible""", + markdown=True, + stream=True, +) + +# Example 3: Address Validation and Geocoding +print("\n=== Address Validation and Geocoding Example ===") +agent.print_response( + """Please validate and geocode this address: + '1600 Amphitheatre Parkway, Mountain View, CA'""", + markdown=True, + stream=True, +) + +# Example 4: Distance Matrix +print("\n=== Distance Matrix Example ===") +agent.print_response( + """Calculate the travel time and distance between these locations in Phoenix: + Origins: ['Phoenix Sky Harbor Airport', 'Downtown Phoenix'] + Destinations: ['Desert Botanical Garden', 'Phoenix Zoo']""", + markdown=True, + stream=True, +) + +# Example 5: Nearby Places and Details +print("\n=== Nearby Places Example ===") +agent.print_response( + """Find coffee shops near Arizona State University Tempe campus. + Include ratings and opening hours if available.""", + markdown=True, + stream=True, +) + +# Example 6: Reverse Geocoding and Timezone +print("\n=== Reverse Geocoding and Timezone Example ===") +agent.print_response( + """Get the address and timezone information for these coordinates: + Latitude: 33.4484, Longitude: -112.0740 (Phoenix)""", + markdown=True, + stream=True, +) + +# Example 7: Multi-step Route Planning +print("\n=== Multi-step Route Planning Example ===") +agent.print_response( + """Plan a route with multiple stops in Phoenix: + Start: Phoenix Sky Harbor Airport + Stops: + 1. Arizona Science Center + 2. Heard Museum + 3. Desert Botanical Garden + End: Return to Airport + Please include estimated travel times between each stop.""", + markdown=True, + stream=True, +) + +# Example 8: Location Analysis +print("\n=== Location Analysis Example ===") +agent.print_response( + """Analyze this location in Phoenix: + Address: '2301 N Central Ave, Phoenix, AZ 85004' + Please provide: + 1. Exact coordinates + 2. Nearby landmarks + 3. Elevation data + 4. Local timezone""", + markdown=True, + stream=True, +) + +# Example 9: Business Hours and Accessibility +print("\n=== Business Hours and Accessibility Example ===") +agent.print_response( + """Find museums in Phoenix that are: + 1. Open on Mondays + 2. Have wheelchair accessibility + 3. Within 5 miles of downtown + Include their opening hours and contact information.""", + markdown=True, + stream=True, +) + +# Example 10: Transit Options +print("\n=== Transit Options Example ===") +agent.print_response( + """Compare different travel modes from 'Phoenix Convention Center' to 'Phoenix Art Museum': + 1. Driving + 2. Walking + 3. Transit (if available) + Include estimated time and distance for each option.""", + markdown=True, + stream=True, +) diff --git a/libs/agno/agno/tools/google_maps.py b/libs/agno/agno/tools/google_maps.py new file mode 100644 index 0000000000..cad7fcef1e --- /dev/null +++ b/libs/agno/agno/tools/google_maps.py @@ -0,0 +1,277 @@ +""" +This module provides tools for searching business information using the Google Maps API. + +Prerequisites: +- Set the environment variable `GOOGLE_MAPS_API_KEY` with your Google Maps API key. + You can obtain the API key from the Google Cloud Console: + https://console.cloud.google.com/projectselector2/google/maps-apis/credentials + +- You also need to activate the Address Validation API for your . + https://console.developers.google.com/apis/api/addressvalidation.googleapis.com + +""" + +import json +from datetime import datetime +from os import getenv +from typing import List, Optional + +from agno.tools import Toolkit + +try: + import googlemaps +except ImportError: + print("Error importing googlemaps. Please install the package using `pip install googlemaps`.") + + +class GoogleMapTools(Toolkit): + def __init__( + self, + key: Optional[str] = None, + search_places: bool = True, + get_directions: bool = True, + validate_address: bool = True, + geocode_address: bool = True, + reverse_geocode: bool = True, + get_distance_matrix: bool = True, + get_elevation: bool = True, + get_timezone: bool = True, + ): + super().__init__(name="google_maps") + + api_key = key or getenv("GOOGLE_MAPS_API_KEY") + if not api_key: + raise ValueError("GOOGLE_MAPS_API_KEY is not set in the environment variables.") + self.client = googlemaps.Client(key=api_key) + + if search_places: + self.register(self.search_places) + if get_directions: + self.register(self.get_directions) + if validate_address: + self.register(self.validate_address) + if geocode_address: + self.register(self.geocode_address) + if reverse_geocode: + self.register(self.reverse_geocode) + if get_distance_matrix: + self.register(self.get_distance_matrix) + if get_elevation: + self.register(self.get_elevation) + if get_timezone: + self.register(self.get_timezone) + + def search_places(self, query: str) -> str: + """ + Search for places using Google Maps Places API. + This tool takes a search query and returns detailed place information. + + Args: + query (str): The query string to search for using Google Maps Search API. (e.g., "dental clinics in Noida") + + Returns: + Stringified list of dictionaries containing business information like name, address, phone, website, rating, and reviews etc. + """ + try: + # Perform places search + places_result = self.client.places(query) + + if not places_result or "results" not in places_result: + return str([]) + + places = [] + for place in places_result["results"]: + place_info = { + "name": place.get("name", ""), + "address": place.get("formatted_address", ""), + "rating": place.get("rating", 0.0), + "reviews": place.get("user_ratings_total", 0), + "place_id": place.get("place_id", ""), + } + + # Get place details for additional information + if place_info.get("place_id"): + try: + details = self.client.place(place_info["place_id"]) + if details and "result" in details: + result = details["result"] + place_info.update( + { + "phone": result.get("formatted_phone_number", ""), + "website": result.get("website", ""), + "hours": result.get("opening_hours", {}).get("weekday_text", []), + } + ) + except Exception as e: + print(f"Error getting place details: {str(e)}") + # Continue with basic place info if details fetch fails + + places.append(place_info) + + return json.dumps(places) + + except Exception as e: + print(f"Error searching Google Maps: {str(e)}") + return str([]) + + def get_directions( + self, + origin: str, + destination: str, + mode: str = "driving", + departure_time: Optional[datetime] = None, + avoid: Optional[List[str]] = None, + ) -> str: + """ + Get directions between two locations using Google Maps Directions API. + + Args: + origin (str): Starting point address or coordinates + destination (str): Destination address or coordinates + mode (str, optional): Travel mode. Options: "driving", "walking", "bicycling", "transit". Defaults to "driving" + departure_time (datetime, optional): Desired departure time for transit directions + avoid (List[str], optional): Features to avoid: "tolls", "highways", "ferries" + + Returns: + str: Stringified dictionary containing route information including steps, distance, duration, etc. + """ + try: + result = self.client.directions(origin, destination, mode=mode, departure_time=departure_time, avoid=avoid) + return str(result) + except Exception as e: + print(f"Error getting directions: {str(e)}") + return str([]) + + def validate_address( + self, address: str, region_code: str = "US", locality: Optional[str] = None, enable_usps_cass: bool = False + ) -> str: + """ + Validate an address using Google Maps Address Validation API. + + Args: + address (str): The address to validate + region_code (str): The region code (e.g., "US" for United States) + locality (str, optional): The locality (city) to help with validation + enable_usps_cass (bool): Whether to enable USPS CASS validation for US addresses + + Returns: + str: Stringified dictionary containing address validation results + """ + try: + result = self.client.addressvalidation( + [address], regionCode=region_code, locality=locality, enableUspsCass=enable_usps_cass + ) + return str(result) + except Exception as e: + print(f"Error validating address: {str(e)}") + return str({}) + + def geocode_address(self, address: str, region: Optional[str] = None) -> str: + """ + Convert an address into geographic coordinates using Google Maps Geocoding API. + + Args: + address (str): The address to geocode + region (str, optional): The region code to bias results + + Returns: + str: Stringified list of dictionaries containing location information + """ + try: + result = self.client.geocode(address, region=region) + return str(result) + except Exception as e: + print(f"Error geocoding address: {str(e)}") + return str([]) + + def reverse_geocode( + self, lat: float, lng: float, result_type: Optional[List[str]] = None, location_type: Optional[List[str]] = None + ) -> str: + """ + Convert geographic coordinates into an address using Google Maps Reverse Geocoding API. + + Args: + lat (float): Latitude + lng (float): Longitude + result_type (List[str], optional): Array of address types to filter results + location_type (List[str], optional): Array of location types to filter results + + Returns: + str: Stringified list of dictionaries containing address information + """ + try: + result = self.client.reverse_geocode((lat, lng), result_type=result_type, location_type=location_type) + return str(result) + except Exception as e: + print(f"Error reverse geocoding: {str(e)}") + return str([]) + + def get_distance_matrix( + self, + origins: List[str], + destinations: List[str], + mode: str = "driving", + departure_time: Optional[datetime] = None, + avoid: Optional[List[str]] = None, + ) -> str: + """ + Calculate distance and time for a matrix of origins and destinations. + + Args: + origins (List[str]): List of addresses or coordinates + destinations (List[str]): List of addresses or coordinates + mode (str, optional): Travel mode. Options: "driving", "walking", "bicycling", "transit" + departure_time (datetime, optional): Desired departure time + avoid (List[str], optional): Features to avoid: "tolls", "highways", "ferries" + + Returns: + str: Stringified dictionary containing distance and duration information + """ + try: + result = self.client.distance_matrix( + origins, destinations, mode=mode, departure_time=departure_time, avoid=avoid + ) + return str(result) + except Exception as e: + print(f"Error getting distance matrix: {str(e)}") + return str({}) + + def get_elevation(self, lat: float, lng: float) -> str: + """ + Get the elevation for a specific location using Google Maps Elevation API. + + Args: + lat (float): Latitude + lng (float): Longitude + + Returns: + str: Stringified dictionary containing elevation data + """ + try: + result = self.client.elevation((lat, lng)) + return str(result) + except Exception as e: + print(f"Error getting elevation: {str(e)}") + return str([]) + + def get_timezone(self, lat: float, lng: float, timestamp: Optional[datetime] = None) -> str: + """ + Get timezone information for a location using Google Maps Time Zone API. + + Args: + lat (float): Latitude + lng (float): Longitude + timestamp (datetime, optional): The timestamp to use for timezone calculation + + Returns: + str: Stringified dictionary containing timezone information + """ + try: + if timestamp is None: + timestamp = datetime.now() + + result = self.client.timezone(location=(lat, lng), timestamp=timestamp) + return str(result) + except Exception as e: + print(f"Error getting timezone: {str(e)}") + return str({}) diff --git a/libs/agno/pyproject.toml b/libs/agno/pyproject.toml index 0c8360937e..e90dca0cde 100644 --- a/libs/agno/pyproject.toml +++ b/libs/agno/pyproject.toml @@ -202,6 +202,7 @@ module = [ "firecrawl.*", "github.*", "google.*", + "googlemaps.*", "google_auth_oauthlib.*", "googleapiclient.*", "googlesearch.*", diff --git a/libs/agno/tests/unit/tools/test_google_maps.py b/libs/agno/tests/unit/tools/test_google_maps.py new file mode 100644 index 0000000000..7e4633b0e3 --- /dev/null +++ b/libs/agno/tests/unit/tools/test_google_maps.py @@ -0,0 +1,317 @@ +"""Unit tests for Google Maps tools.""" + +import json +from datetime import datetime +from unittest.mock import patch + +import pytest + +from agno.tools.google_maps import GoogleMapTools + +# Mock responses +MOCK_PLACES_RESPONSE = { + "results": [ + { + "name": "Test Business", + "formatted_address": "123 Test St, Test City", + "rating": 4.5, + "user_ratings_total": 100, + "place_id": "test_place_id", + } + ] +} + +MOCK_PLACE_DETAILS = { + "result": { + "formatted_phone_number": "123-456-7890", + "website": "https://test.com", + "opening_hours": {"weekday_text": ["Monday: 9:00 AM – 5:00 PM"]}, + } +} + +MOCK_DIRECTIONS_RESPONSE = [ + { + "legs": [ + { + "distance": {"text": "5 km", "value": 5000}, + "duration": {"text": "10 mins", "value": 600}, + "steps": [], + } + ] + } +] + +MOCK_ADDRESS_VALIDATION_RESPONSE = { + "result": { + "verdict": {"validationGranularity": "PREMISE", "hasInferredComponents": False}, + "address": {"formattedAddress": "123 Test St, Test City, ST 12345"}, + } +} + +MOCK_GEOCODE_RESPONSE = [ + { + "formatted_address": "123 Test St, Test City, ST 12345", + "geometry": {"location": {"lat": 40.7128, "lng": -74.0060}}, + } +] + +MOCK_DISTANCE_MATRIX_RESPONSE = { + "rows": [ + { + "elements": [ + { + "distance": {"text": "5 km", "value": 5000}, + "duration": {"text": "10 mins", "value": 600}, + } + ] + } + ] +} + +MOCK_ELEVATION_RESPONSE = [{"elevation": 100.0}] + +MOCK_TIMEZONE_RESPONSE = { + "timeZoneId": "America/New_York", + "timeZoneName": "Eastern Daylight Time", +} + + +@pytest.fixture +def google_maps_tools(): + """Create a GoogleMapTools instance with a mock API key.""" + with patch.dict("os.environ", {"GOOGLE_MAPS_API_KEY": "AIzaTest"}): + return GoogleMapTools() + + +@pytest.fixture +def mock_client(): + """Create a mock Google Maps client.""" + with patch("googlemaps.Client") as mock: + yield mock + + +def test_search_places(google_maps_tools): + """Test the search_places method.""" + with patch.object(google_maps_tools.client, "places") as mock_places: + with patch.object(google_maps_tools.client, "place") as mock_place: + mock_places.return_value = MOCK_PLACES_RESPONSE + mock_place.return_value = MOCK_PLACE_DETAILS + + result = json.loads(google_maps_tools.search_places("test query")) + + assert len(result) == 1 + assert result[0]["name"] == "Test Business" + assert result[0]["phone"] == "123-456-7890" + assert result[0]["website"] == "https://test.com" + + +def test_get_directions(google_maps_tools): + """Test the get_directions method.""" + with patch.object(google_maps_tools.client, "directions") as mock_directions: + mock_directions.return_value = MOCK_DIRECTIONS_RESPONSE + + result = eval(google_maps_tools.get_directions(origin="Test Origin", destination="Test Destination")) + + assert isinstance(result, list) + assert "legs" in result[0] + assert result[0]["legs"][0]["distance"]["value"] == 5000 + + +def test_validate_address(google_maps_tools): + """Test the validate_address method.""" + with patch.object(google_maps_tools.client, "addressvalidation") as mock_validate: + mock_validate.return_value = MOCK_ADDRESS_VALIDATION_RESPONSE + + result = eval(google_maps_tools.validate_address("123 Test St")) + + assert isinstance(result, dict) + assert "result" in result + assert "verdict" in result["result"] + + +def test_geocode_address(google_maps_tools): + """Test the geocode_address method.""" + with patch.object(google_maps_tools.client, "geocode") as mock_geocode: + mock_geocode.return_value = MOCK_GEOCODE_RESPONSE + + result = eval(google_maps_tools.geocode_address("123 Test St")) + + assert isinstance(result, list) + assert result[0]["formatted_address"] == "123 Test St, Test City, ST 12345" + + +def test_reverse_geocode(google_maps_tools): + """Test the reverse_geocode method.""" + with patch.object(google_maps_tools.client, "reverse_geocode") as mock_reverse: + mock_reverse.return_value = MOCK_GEOCODE_RESPONSE + + result = eval(google_maps_tools.reverse_geocode(40.7128, -74.0060)) + + assert isinstance(result, list) + assert result[0]["formatted_address"] == "123 Test St, Test City, ST 12345" + + +def test_get_distance_matrix(google_maps_tools): + """Test the get_distance_matrix method.""" + with patch.object(google_maps_tools.client, "distance_matrix") as mock_matrix: + mock_matrix.return_value = MOCK_DISTANCE_MATRIX_RESPONSE + + result = eval(google_maps_tools.get_distance_matrix(origins=["Origin"], destinations=["Destination"])) + + assert isinstance(result, dict) + assert "rows" in result + assert result["rows"][0]["elements"][0]["distance"]["value"] == 5000 + + +def test_get_elevation(google_maps_tools): + """Test the get_elevation method.""" + with patch.object(google_maps_tools.client, "elevation") as mock_elevation: + mock_elevation.return_value = MOCK_ELEVATION_RESPONSE + + result = eval(google_maps_tools.get_elevation(40.7128, -74.0060)) + + assert isinstance(result, list) + assert result[0]["elevation"] == 100.0 + + +def test_get_timezone(google_maps_tools): + """Test the get_timezone method.""" + with patch.object(google_maps_tools.client, "timezone") as mock_timezone: + mock_timezone.return_value = MOCK_TIMEZONE_RESPONSE + test_time = datetime(2024, 1, 1, 12, 0) + + result = eval(google_maps_tools.get_timezone(40.7128, -74.0060, test_time)) + + assert isinstance(result, dict) + assert result["timeZoneId"] == "America/New_York" + + +def test_error_handling(google_maps_tools): + """Test error handling in various methods.""" + with patch.object(google_maps_tools.client, "places") as mock_places: + mock_places.side_effect = Exception("API Error") + + result = google_maps_tools.search_places("test query") + assert result == "[]" + + with patch.object(google_maps_tools.client, "directions") as mock_directions: + mock_directions.side_effect = Exception("API Error") + + result = google_maps_tools.get_directions("origin", "destination") + assert result == "[]" + + +def test_initialization_without_api_key(): + """Test initialization without API key.""" + with patch.dict("os.environ", clear=True): + with pytest.raises(ValueError, match="GOOGLE_MAPS_API_KEY is not set"): + GoogleMapTools() + + +def test_initialization_with_selective_tools(): + """Test initialization with only selected tools.""" + with patch.dict("os.environ", {"GOOGLE_MAPS_API_KEY": "AIzaTest"}): + tools = GoogleMapTools( + search_places=True, + get_directions=False, + validate_address=False, + geocode_address=True, + reverse_geocode=False, + get_distance_matrix=False, + get_elevation=False, + get_timezone=False, + ) + + assert "search_places" in [func.name for func in tools.functions.values()] + assert "get_directions" not in [func.name for func in tools.functions.values()] + assert "geocode_address" in [func.name for func in tools.functions.values()] + + +def test_search_places_success(google_maps_tools): + """Test the search_places method with successful response.""" + with patch.object(google_maps_tools.client, "places") as mock_places: + with patch.object(google_maps_tools.client, "place") as mock_place: + mock_places.return_value = MOCK_PLACES_RESPONSE + mock_place.return_value = MOCK_PLACE_DETAILS + + result = json.loads(google_maps_tools.search_places("test query")) + + assert len(result) == 1 + assert result[0]["name"] == "Test Business" + assert result[0]["phone"] == "123-456-7890" + assert result[0]["website"] == "https://test.com" + mock_places.assert_called_once_with("test query") + mock_place.assert_called_once_with("test_place_id") + + +def test_search_places_no_results(google_maps_tools): + """Test search_places when no results are returned.""" + with patch.object(google_maps_tools.client, "places") as mock_places: + mock_places.return_value = {"results": []} + result = json.loads(google_maps_tools.search_places("test query")) + assert result == [] + + +def test_search_places_none_response(google_maps_tools): + """Test search_places when None is returned.""" + with patch.object(google_maps_tools.client, "places") as mock_places: + mock_places.return_value = None + result = json.loads(google_maps_tools.search_places("test query")) + assert result == [] + + +def test_search_places_missing_results_key(google_maps_tools): + """Test search_places when response is missing results key.""" + with patch.object(google_maps_tools.client, "places") as mock_places: + mock_places.return_value = {"status": "OK"} + result = json.loads(google_maps_tools.search_places("test query")) + assert result == [] + + +def test_search_places_missing_place_id(google_maps_tools): + """Test search_places when place_id is missing.""" + with patch.object(google_maps_tools.client, "places") as mock_places: + mock_places.return_value = { + "results": [ + { + "name": "Test Business", + "formatted_address": "123 Test St", + "rating": 4.5, + } + ] + } + result = json.loads(google_maps_tools.search_places("test query")) + assert len(result) == 1 + assert result[0]["name"] == "Test Business" + assert "phone" not in result[0] + assert "website" not in result[0] + + +def test_search_places_invalid_details(google_maps_tools): + """Test search_places when place details are invalid.""" + with patch.object(google_maps_tools.client, "places") as mock_places: + with patch.object(google_maps_tools.client, "place") as mock_place: + mock_places.return_value = MOCK_PLACES_RESPONSE + mock_place.return_value = {"status": "NOT_FOUND"} # Missing 'result' key + + result = json.loads(google_maps_tools.search_places("test query")) + + assert len(result) == 1 + assert result[0]["name"] == "Test Business" + assert "phone" not in result[0] + assert "website" not in result[0] + + +def test_search_places_details_error(google_maps_tools): + """Test search_places when place details call raises an error.""" + with patch.object(google_maps_tools.client, "places") as mock_places: + with patch.object(google_maps_tools.client, "place") as mock_place: + mock_places.return_value = MOCK_PLACES_RESPONSE + mock_place.side_effect = Exception("API Error") + + result = json.loads(google_maps_tools.search_places("test query")) + + assert len(result) == 1 + assert result[0]["name"] == "Test Business" + assert "phone" not in result[0] + assert "website" not in result[0]