Upload tool

requirements.txt

@@ -1,2 +1,2 @@
 smolagents
-requests
+aiohttp

tool.py

@@ -1,8 +1,9 @@
 from typing import Any, Optional
 from smolagents.tools import Tool
-import json
-import requests
+import asyncio
 import logging
+import json
+import aiohttp
 
 class UserLocationTool(Tool):
     name = "get_user_location"
@@ -10,88 +11,65 @@ class UserLocationTool(Tool):
     inputs = {}
     output_type = "string"
 
-    def __init__(self):
-        # Calls the constructor of the parent class `Tool`
+    def __init__(self): # Initialize the logger
         super().__init__()
-        self.dependencies = ["requests"]
-
-    def _validate_location_data(self, data: dict) -> str | None:
-        """Validates the location data and returns error message if invalid."""
-
-        required_fields = ['city', 'region', 'country']
-
-        if not isinstance(data, dict):
-            return f"Expected dictionary data, got {type(data)}"
-
-        missing_fields = [field for field in required_fields if field not in data]
-        if missing_fields:
-            return f"Missing required location fields: {', '.join(missing_fields)}"
-
-        empty_fields = [field for field in required_fields if not data.get(field)]
-        if empty_fields:
-            return f"Empty values for fields: {', '.join(empty_fields)}"
-
-        return None
-
-    def forward(self) -> str:
-        # Import inside method as per SmolAgents requirements
-        import requests
-        import json
         import logging
+        self.logger = logging.getLogger(__name__)
+        self.logger.setLevel(logging.INFO)
 
-        logger = logging.getLogger(__name__)
-        logger.setLevel(logging.INFO)
-
-        try:
-            logger.info("Initiating request to ipinfo.io")
-            response = requests.get("https://ipinfo.io/", timeout=5)
-
-            # Handle rate limiting
-            if response.status_code == 429:
-                error_msg = "Rate limit exceeded for IP geolocation service"
-                logger.error(error_msg)
-                return f"Error getting user location: {error_msg}. Please try again later."
-
-            # Handle other HTTP errors
-            if response.status_code != 200:
-                error_msg = f"HTTP {response.status_code}: {response.reason}"
-                logger.error(f"ipinfo.io request failed: {error_msg}")
-                return f"Error getting user location: Server returned {error_msg}"
+    def _validate_location_data(self, response: Any) -> tuple[bool, str]:
+        # Validates the API response, JSON parsing, and location data. Returns (is_valid, error_message or location_string)
+        import json
 
-            # check if the response is valid JSON
-            try:
-                data = response.json()
-            except json.JSONDecodeError as e:
-                error_msg = f"Invalid JSON response: {str(e)}"
-                logger.error(error_msg)
-                return f"Error getting user location: {error_msg}"
+        if response.status != 200: # Check HTTP status
+            if response.status == 429:
+                return False, "Rate limit exceeded for IP geolocation service. Please try again later."
+            return False, f"Server returned HTTP {response.status}"
 
-            # check if the data contains: city, region, and country
-            validation_error = self._validate_location_data(data)
-            if validation_error:
-                logger.error(f"Data validation failed: {validation_error}")
-                return f"Error getting user location: {validation_error}"
+        try: # Parse and validate JSON
+            response_data = response.json()
+        except json.JSONDecodeError:
+            return False, "Invalid response from server"
 
-            location = f"According to your IP address, the user's location is {data['city']}, {data['region']}, {data['country']}."
-            logger.info(f"Successfully retrieved location: {location}")
-            return location
+        if not isinstance(response_data, dict): # Validate response data type
+            return False, f"Expected dictionary data, got {type(response_data)}"
 
-        except requests.exceptions.Timeout:
-            error_msg = "Request timed out after 5 seconds"
-            logger.error(error_msg)
-            return f"Error getting user location: {error_msg}. Please try again."
+        required_fields = ['city', 'region', 'country'] # Validate required fields
+        missing_fields = [field for field in required_fields if not response_data.get(field)]
+        if missing_fields:
+            return False, f"Missing or empty location fields: {', '.join(missing_fields)}"
 
-        except requests.exceptions.ConnectionError as e:
-            error_msg = f"Connection failed: {str(e)}"
-            logger.error(error_msg)
-            return f"Error getting user location: Unable to connect to location service. Please check your internet connection."
+        location = f"According to your IP address, the user's location is {response_data['city']}, {response_data['region']}, {response_data['country']}."
+        return True, location
 
-        except requests.exceptions.RequestException as e:
-            error_msg = f"Network error: {str(e)}"
-            logger.error(error_msg)
-            return f"Error getting user location: {error_msg}"
+    async def _call_api(self) -> tuple[bool, Any]:
+        # Makes the API call to ipinfo.io. Returns (success, response_object_or_error_message)
+        import aiohttp
+        import asyncio
 
+        try:
+            async with aiohttp.ClientSession() as session:
+                async with session.get("https://ipinfo.io/", timeout=5) as response:
+                    return True, response
+        except asyncio.TimeoutError:
+            return False, "Request timed out after 5 seconds. Please try again."
+        except aiohttp.ClientError as e:
+            return False, "Unable to connect to location service. Please check your internet connection."
         except Exception as e:
-            error_msg = f"Unexpected error: {str(e)}"
-            logger.error(error_msg)
-            return f"Error getting user location: An unexpected error occurred. Please try again later."
+            return False, f"An unexpected error occurred: {str(e)}"
+
+    async def forward(self) -> str:
+        # Send the request to the API
+        success, response = await self._call_api()
+        if not success:
+            self.logger.error(response)
+            return f"Error getting user location: {response}"
+
+        # Validate response format
+        is_valid, result = self._validate_location_data(response)
+        if not is_valid:
+            self.logger.error(f"Data validation failed: {result}")
+            return f"Error getting user location: {result}"
+
+        self.logger.info(f"Successfully retrieved location: {result}")
+        return result