"""Emissions data 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__)
EMISSIONS_TOOL = Tool(
name="get_emissions",
description="""Query greenhouse gas emissions from Climate TRACE and EDGAR databases.
Returns facility-level and national emissions data including CO2, CH4, and N2O
with sector classifications, timestamps, and source information.
At least one location filter is required (bbox, point+radius, or country_codes).""",
inputSchema={
"type": "object",
"properties": {
"sources": {
"type": "array",
"items": {
"type": "string",
"enum": ["climatetrace", "edgar"],
},
"description": "Data sources to query (default: all)",
},
"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', 'CHN', 'DEU'])",
},
"sectors": {
"type": "array",
"items": {"type": "string"},
"description": "Sector filters (e.g., ['power', 'steel', 'cement', 'oil-and-gas'])",
},
"gases": {
"type": "array",
"items": {
"type": "string",
"enum": ["co2", "ch4", "n2o"],
},
"description": "Greenhouse gas types to filter",
},
"date_from": {
"type": "string",
"format": "date",
"description": "Start date (ISO 8601)",
},
"date_to": {
"type": "string",
"format": "date",
"description": "End date (ISO 8601)",
},
"limit": {
"type": "integer",
"minimum": 1,
"maximum": 1000,
"default": 100,
"description": "Maximum number of results",
},
},
"required": [],
},
)
async def execute_emissions(
client: JanaClient, arguments: dict[str, Any]
) -> list[TextContent]:
"""
Execute the emissions tool.
Args:
client: Jana API client
arguments: Tool arguments from MCP call
Returns:
List of TextContent with results or error message
"""
logger.info("Executing get_emissions")
# 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_emissions(
sources=arguments.get("sources"),
bbox=arguments.get("location_bbox"),
point=arguments.get("location_point"),
radius_km=arguments.get("radius_km"),
country_codes=arguments.get("country_codes"),
sectors=arguments.get("sectors"),
gases=arguments.get("gases"),
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_emissions")
error_response = create_error_response(
f"API error querying emissions data: {e}", "API_ERROR"
)
return [TextContent(type="text", text=serialize_response(error_response))]
except httpx.RequestError as e:
logger.exception("Network error in get_emissions")
error_response = create_error_response(
f"Network error querying emissions 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_emissions")
error_response = create_error_response(
f"Invalid data format: {e}", "DATA_ERROR"
)
return [TextContent(type="text", text=serialize_response(error_response))]
