test_mcp_server / tools /hiking_tool.py
SrikanthNagelli's picture
initial commit
b0979b9
"""
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)}"
}