"""Nearby location search tools for MCP."""
import logging
from typing import Any
import httpx
from mcp.types import TextContent, Tool
from jana_mcp.client import APIError, AuthenticationError, JanaClient
from jana_mcp.constants import DEFAULT_LIMIT
from jana_mcp.tools.response import create_error_response, serialize_response
from jana_mcp.tools.validation import validate_coordinates
logger = logging.getLogger(__name__)
NEARBY_TOOL = Tool(
name="find_nearby",
description="""Find environmental monitoring stations and emission sources near a location.
Returns air quality stations and emission facilities within the specified radius
of the given coordinates, with distance information.""",
inputSchema={
"type": "object",
"properties": {
"location_point": {
"type": "array",
"items": {"type": "number"},
"minItems": 2,
"maxItems": 2,
"description": "Point coordinates [longitude, latitude]",
},
"radius_km": {
"type": "number",
"minimum": 0.1,
"maximum": 500,
"description": "Search radius in kilometers",
},
"sources": {
"type": "array",
"items": {
"type": "string",
"enum": ["openaq", "climatetrace", "edgar"],
},
"description": "Data sources to search (default: all)",
},
"limit": {
"type": "integer",
"minimum": 1,
"maximum": 1000,
"default": 100,
"description": "Maximum number of results",
},
},
"required": ["location_point", "radius_km"],
},
)
async def execute_nearby(
client: JanaClient, arguments: dict[str, Any]
) -> list[TextContent]:
"""
Execute the nearby search tool.
Args:
client: Jana API client
arguments: Tool arguments from MCP call
Returns:
List of TextContent with results or error message
"""
logger.info("Executing find_nearby")
# Validate required parameters
location_point = arguments.get("location_point")
radius_km = arguments.get("radius_km")
if not location_point:
error_response = create_error_response(
"location_point is required [longitude, latitude]", "VALIDATION_ERROR"
)
return [TextContent(type="text", text=serialize_response(error_response))]
if not radius_km:
error_response = create_error_response(
"radius_km is required", "VALIDATION_ERROR"
)
return [TextContent(type="text", text=serialize_response(error_response))]
# Validate coordinates
coord_error = validate_coordinates(location_point[0], location_point[1])
if coord_error:
error_response = create_error_response(coord_error, "VALIDATION_ERROR")
return [TextContent(type="text", text=serialize_response(error_response))]
try:
result = await client.find_nearby(
point=location_point,
radius_km=radius_km,
sources=arguments.get("sources"),
limit=arguments.get("limit", DEFAULT_LIMIT),
)
return [
TextContent(
type="text",
text=serialize_response(result),
)
]
except (APIError, AuthenticationError) as e:
logger.exception("API error in find_nearby")
error_response = create_error_response(
f"API error finding nearby data: {e}", "API_ERROR"
)
return [TextContent(type="text", text=serialize_response(error_response))]
except httpx.RequestError as e:
logger.exception("Network error in find_nearby")
error_response = create_error_response(
f"Network error finding nearby data: {e}", "NETWORK_ERROR"
)
return [TextContent(type="text", text=serialize_response(error_response))]
except (KeyError, ValueError, TypeError) as e:
logger.exception("Data parsing error in find_nearby")
error_response = create_error_response(
f"Invalid data format: {e}", "DATA_ERROR"
)
return [TextContent(type="text", text=serialize_response(error_response))]
