""" Hiking trail search tool for MCP server. """ from smolagents import Tool from typing import Optional, Dict, Any from services.hiking_service import HikingRecommendationServer class HikingSearchTool(Tool): """Tool for searching hiking trails using enhanced hiking service.""" def __init__(self): self.name = "hiking_search" self.description = "Search for hiking trails and outdoor activities in a location" self.input_type = "object" self.output_type = "object" self.inputs = { "location": { "type": "string", "description": "Location to search for hiking trails" }, "difficulty": { "type": "string", "description": "Difficulty level: Easy, Moderate, Hard, Very Hard", "optional": True, "nullable": True }, "max_distance": { "type": "integer", "description": "Maximum distance in miles", "optional": True, "nullable": True } } self.outputs = { "location": { "type": "string", "description": "Search location" }, "total_found": { "type": "integer", "description": "Total number of trails found" }, "trails": { "type": "array", "description": "List of hiking trails found" }, "stats": { "type": "object", "description": "Statistics about the trails found" } } self.required_inputs = ["location"] self.is_initialized = True def forward(self, location: str, difficulty: Optional[str] = None, max_distance: Optional[int] = None) -> dict: """Search for hiking trails using enhanced hiking service.""" try: # Set defaults if difficulty is None: difficulty = "All" if max_distance is None: max_distance = 50 # Normalize difficulty parameter to handle lowercase inputs from client if difficulty and difficulty.lower() != "all": difficulty_mapping = { "easy": "Easy", "moderate": "Moderate", "hard": "Hard", "difficult": "Hard", "challenging": "Hard", "very hard": "Very Hard", "very_hard": "Very Hard", "extreme": "Very Hard" } normalized_difficulty = difficulty_mapping.get(difficulty.lower(), difficulty) print(f"🏔️ Tool: Normalized difficulty '{difficulty}' → '{normalized_difficulty}'") difficulty = normalized_difficulty # Initialize hiking service hiking_server = HikingRecommendationServer() # Get structured recommendations results = hiking_server.get_hiking_recommendations_structured(location, max_distance, difficulty) # Handle error cases if not results.get("success", False): return { "location": location, "total_found": 0, "trails": [], "error": results.get("message", "Failed to find hiking trails") } # Convert service results to tool output format formatted_trails = [] for hike in results.get('hikes', []): trail_info = { "name": hike.get('name', 'Unnamed Trail'), "difficulty": hike.get('difficulty_level', 'Unknown'), "distance": f"{hike.get('distance', 0)} miles", "elevation_gain": f"{hike.get('elevation_gain', 0)} ft", "type": hike.get('type', 'Trail'), "rating": hike.get('rating', 'Not rated'), "reviews": hike.get('reviews', 'No reviews'), "features": hike.get('features', []), "description": hike.get('description', 'A beautiful trail') } # Add additional details if available if 'duration' in hike: trail_info['duration'] = hike['duration'] if 'popularity' in hike: trail_info['popularity'] = hike['popularity'] if 'best_time' in hike: trail_info['best_time'] = hike['best_time'] formatted_trails.append(trail_info) # Include statistics stats = results.get('stats', {}) return { "location": location, "search_radius": f"{max_distance} miles", "difficulty_filter": difficulty, "total_found": results.get('total_hikes', len(formatted_trails)), "trails": formatted_trails, "stats": { "total_hikes": stats.get('total_hikes', 0), "avg_distance": f"{stats.get('avg_distance', 0)} miles", "avg_elevation": f"{stats.get('avg_elevation', 0)} ft", "difficulty_breakdown": stats.get('difficulty_breakdown', {}) } } except Exception as e: return { "location": location, "total_found": 0, "trails": [], "error": f"Error finding hiking trails: {str(e)}" }