"""Air quality measurement 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, POINT_COORDINATES_REQUIRED
from jana_mcp.tools.response import create_error_response, serialize_response
from jana_mcp.tools.validation import require_location_filter, validate_bbox, validate_coordinates, validate_date_format
logger = logging.getLogger(__name__)
AIR_QUALITY_TOOL = Tool(
name="get_air_quality",
description="""Query air quality measurements from OpenAQ data.
Returns pollutant concentrations (PM2.5, PM10, O3, NO2, SO2, CO) with location,
timestamp, and measurement details.
At least one location filter is required (bbox, point+radius, or country_codes).""",
inputSchema={
"type": "object",
"properties": {
"location_bbox": {
"type": "array",
"items": {"type": "number"},
"minItems": 4,
"maxItems": 4,
"description": "Bounding box [min_lon, min_lat, max_lon, max_lat]",
},
"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 (requires location_point)",
},
"country_codes": {
"type": "array",
"items": {"type": "string"},
"description": "ISO-3 country codes (e.g., ['USA', 'GBR'])",
},
"parameters": {
"type": "array",
"items": {
"type": "string",
"enum": ["pm25", "pm10", "o3", "no2", "so2", "co"],
},
"description": "Pollutant parameters to filter",
},
"date_from": {
"type": "string",
"format": "date",
"description": "Start date (ISO 8601, e.g., '2024-01-01')",
},
"date_to": {
"type": "string",
"format": "date",
"description": "End date (ISO 8601, e.g., '2024-12-31')",
},
"limit": {
"type": "integer",
"minimum": 1,
"maximum": 1000,
"default": 100,
"description": "Maximum number of results",
},
},
"required": [],
},
)
async def execute_air_quality(
client: JanaClient, arguments: dict[str, Any]
) -> list[TextContent]:
"""
Execute the air quality tool.
Args:
client: Jana API client
arguments: Tool arguments from MCP call
Returns:
List of TextContent with results or error message
"""
# Log tool execution without sensitive data
logger.info("Executing get_air_quality")
# Validate location filter
location_error = require_location_filter(arguments)
if location_error:
error_response = create_error_response(location_error, "VALIDATION_ERROR")
return [
TextContent(
type="text",
text=serialize_response(error_response),
)
]
# Validate coordinates and dates
bbox = arguments.get("location_bbox")
if bbox:
bbox_error = validate_bbox(bbox)
if bbox_error:
error_response = create_error_response(bbox_error, "VALIDATION_ERROR")
return [TextContent(type="text", text=serialize_response(error_response))]
point = arguments.get("location_point")
if point and len(point) >= POINT_COORDINATES_REQUIRED:
coord_error = validate_coordinates(point[0], point[1])
if coord_error:
error_response = create_error_response(coord_error, "VALIDATION_ERROR")
return [TextContent(type="text", text=serialize_response(error_response))]
date_from = arguments.get("date_from")
if date_from:
date_error = validate_date_format(date_from)
if date_error:
error_response = create_error_response(date_error, "VALIDATION_ERROR")
return [TextContent(type="text", text=serialize_response(error_response))]
date_to = arguments.get("date_to")
if date_to:
date_error = validate_date_format(date_to)
if date_error:
error_response = create_error_response(date_error, "VALIDATION_ERROR")
return [TextContent(type="text", text=serialize_response(error_response))]
try:
result = await client.get_air_quality(
bbox=arguments.get("location_bbox"),
point=arguments.get("location_point"),
radius_km=arguments.get("radius_km"),
country_codes=arguments.get("country_codes"),
parameters=arguments.get("parameters"),
date_from=arguments.get("date_from"),
date_to=arguments.get("date_to"),
limit=arguments.get("limit", DEFAULT_LIMIT),
)
return [
TextContent(
type="text",
text=serialize_response(result),
)
]
except (APIError, AuthenticationError) as e:
logger.exception("API error in get_air_quality")
error_response = create_error_response(
f"API error querying air quality data: {e}", "API_ERROR"
)
return [TextContent(type="text", text=serialize_response(error_response))]
except httpx.RequestError as e:
logger.exception("Network error in get_air_quality")
error_response = create_error_response(
f"Network error querying air quality 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 get_air_quality")
error_response = create_error_response(
f"Invalid data format: {e}", "DATA_ERROR"
)
return [TextContent(type="text", text=serialize_response(error_response))]
