get_air_quality
Retrieve air quality data for any city including pollutant levels and health advisories to assess environmental conditions.
Instructions
Get air quality information for a specified city including PM2.5, PM10, ozone, nitrogen dioxide, carbon monoxide, and other pollutants. Provides health advisories based on current air quality levels.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| city | Yes | The name of the city to fetch air quality information for, PLEASE NOTE English name only, if the parameter city isn't English please translate to English before invoking this function. | |
| variables | No | Air quality variables to retrieve. If not specified, defaults to pm10, pm2_5, ozone, nitrogen_dioxide, and carbon_monoxide. |
Implementation Reference
- The run_tool method in GetAirQualityToolHandler that implements the core execution logic for the 'get_air_quality' tool: validates input, geocodes city using WeatherService, fetches air quality data via AirQualityService, computes current index, formats comprehensive response.async def run_tool(self, args: dict) -> Sequence[TextContent | ImageContent | EmbeddedResource]: """ Execute the air quality tool. """ try: self.validate_required_args(args, ["city"]) city = args["city"] variables = args.get("variables", [ "pm10", "pm2_5", "ozone", "nitrogen_dioxide", "carbon_monoxide" ]) logger.info(f"Getting air quality for: {city} with variables: {variables}") # Get coordinates for the city latitude, longitude = await self.weather_service.get_coordinates(city) # Get air quality data aq_data = await self.air_quality_service.get_air_quality( latitude, longitude, variables ) # Get current air quality values current_aq = self.air_quality_service.get_current_air_quality_index(aq_data) # Build comprehensive response data for AI comprehension response_data = { "city": city, "latitude": latitude, "longitude": longitude, "current_air_quality": current_aq, "full_data": aq_data } # Format the response with comprehensive field descriptions formatted_response = self.air_quality_service.format_air_quality_comprehensive( response_data ) return [ TextContent( type="text", text=formatted_response ) ] except ValueError as e: logger.error(f"Air quality service error: {str(e)}") return [ TextContent( type="text", text=f"Error: {str(e)}" ) ] except Exception as e: logger.exception(f"Unexpected error in get_air_quality: {str(e)}") return [ TextContent( type="text", text=f"Unexpected error occurred: {str(e)}" ) ]
- The get_tool_description method defining the Tool object for 'get_air_quality', including detailed inputSchema with city (required) and optional variables enum.def get_tool_description(self) -> Tool: """ Return the tool description for air quality lookup. """ return Tool( name=self.name, description="""Get air quality information for a specified city including PM2.5, PM10, ozone, nitrogen dioxide, carbon monoxide, and other pollutants. Provides health advisories based on current air quality levels.""", inputSchema={ "type": "object", "properties": { "city": { "type": "string", "description": "The name of the city to fetch air quality information for, PLEASE NOTE English name only, if the parameter city isn't English please translate to English before invoking this function." }, "variables": { "type": "array", "items": { "type": "string", "enum": [ "pm10", "pm2_5", "carbon_monoxide", "nitrogen_dioxide", "ozone", "sulphur_dioxide", "ammonia", "dust", "aerosol_optical_depth" ] }, "description": "Air quality variables to retrieve. If not specified, defaults to pm10, pm2_5, ozone, nitrogen_dioxide, and carbon_monoxide." } }, "required": ["city"] } )
- src/mcp_weather_server/server.py:95-97 (registration)Registration of the GetAirQualityToolHandler in the server's register_all_tools function.# Air quality tools add_tool_handler(GetAirQualityToolHandler()) add_tool_handler(GetAirQualityDetailsToolHandler())
- AirQualityService.get_air_quality: core helper method that makes HTTP request to Open-Meteo air quality API to fetch pollutant data for given lat/lon and variables.async def get_air_quality( self, latitude: float, longitude: float, hourly_vars: List[str] = None ) -> Dict[str, Any]: """ Get air quality data for given coordinates. Args: latitude: Latitude coordinate longitude: Longitude coordinate hourly_vars: List of hourly variables to retrieve Returns: Air quality data dictionary Raises: ValueError: If air quality data cannot be retrieved """ if hourly_vars is None: hourly_vars = ["pm10", "pm2_5", "ozone", "nitrogen_dioxide", "carbon_monoxide"] hourly_str = ",".join(hourly_vars) url = ( f"{self.BASE_AIR_QUALITY_URL}" f"?latitude={latitude}&longitude={longitude}" f"&hourly={hourly_str}" f"&timezone=GMT" ) logger.info(f"Fetching air quality data from: {url}") try: async with httpx.AsyncClient() as client: response = await client.get(url) if response.status_code != 200: raise ValueError(f"Air Quality API returned status {response.status_code}") return response.json() except httpx.RequestError as e: raise ValueError(f"Network error while fetching air quality data: {str(e)}") except (KeyError, IndexError) as e: raise ValueError(f"Invalid response format from air quality API: {str(e)}") def get_current_air_quality_index(self, aq_data: Dict[str, Any]) -> Dict[str, Any]: