"""System and administrative tools for MCP."""
import logging
from typing import Any
import httpx
from mcp.types import TextContent, Tool
from jana_mcp import __version__
from jana_mcp.client import APIError, AuthenticationError, JanaClient
from jana_mcp.tools.response import create_error_response, serialize_response
logger = logging.getLogger(__name__)
DATA_SUMMARY_TOOL = Tool(
name="get_data_summary",
description="""Get a summary of available data in the Jana platform.
Returns information about data coverage, record counts by source,
geographic coverage, and data freshness.""",
inputSchema={
"type": "object",
"properties": {},
"required": [],
},
)
SYSTEM_HEALTH_TOOL = Tool(
name="get_system_health",
description="""Check the health status of the Jana platform.
Returns system availability, backend connectivity, and any active issues.""",
inputSchema={
"type": "object",
"properties": {},
"required": [],
},
)
async def execute_data_summary(client: JanaClient) -> list[TextContent]:
"""
Execute the data summary tool.
Args:
client: Jana API client
Returns:
List of TextContent with summary or error message
"""
logger.info("Executing get_data_summary")
try:
result = await client.get_summary()
return [
TextContent(
type="text",
text=serialize_response(result),
)
]
except (APIError, AuthenticationError):
logger.exception("API error in get_data_summary")
# Return a basic summary with error info
api_fallback: dict[str, Any] = {
"status": "partial",
"error": "API error retrieving summary",
"error_code": "API_ERROR",
"available_sources": [
{
"name": "OpenAQ",
"description": "Global air quality measurements",
"parameters": ["pm25", "pm10", "o3", "no2", "so2", "co"],
},
{
"name": "Climate TRACE",
"description": "Facility-level GHG emissions",
"gases": ["co2", "ch4", "n2o"],
},
{
"name": "EDGAR",
"description": "National emissions inventories",
"gases": ["co2", "ch4", "n2o"],
},
],
}
return [
TextContent(
type="text",
text=serialize_response(api_fallback),
)
]
except httpx.RequestError:
logger.exception("Network error in get_data_summary")
# Return a basic summary with error info
network_fallback: dict[str, Any] = {
"status": "partial",
"error": "Network error retrieving summary",
"error_code": "NETWORK_ERROR",
"available_sources": [
{
"name": "OpenAQ",
"description": "Global air quality measurements",
"parameters": ["pm25", "pm10", "o3", "no2", "so2", "co"],
},
{
"name": "Climate TRACE",
"description": "Facility-level GHG emissions",
"gases": ["co2", "ch4", "n2o"],
},
{
"name": "EDGAR",
"description": "National emissions inventories",
"gases": ["co2", "ch4", "n2o"],
},
],
}
return [
TextContent(
type="text",
text=serialize_response(network_fallback),
)
]
except (KeyError, ValueError, TypeError) as e:
logger.exception("Data parsing error in get_data_summary")
error_response = create_error_response(
f"Invalid data format: {e}", "DATA_ERROR"
)
return [TextContent(type="text", text=serialize_response(error_response))]
async def execute_system_health(client: JanaClient) -> list[TextContent]:
"""
Execute the system health tool.
Args:
client: Jana API client
Returns:
List of TextContent with health status
"""
logger.info("Executing get_system_health")
try:
health = await client.check_health()
# Add MCP server status
health["mcp_server"] = {
"status": "healthy",
"version": __version__,
}
return [
TextContent(
type="text",
text=serialize_response(health),
)
]
except (APIError, AuthenticationError):
logger.exception("API error in get_system_health")
api_error_response: dict[str, Any] = {
"mcp_server": {"status": "healthy", "version": __version__},
"backend": {"status": "error", "error": "API error", "error_code": "API_ERROR"},
}
return [TextContent(type="text", text=serialize_response(api_error_response))]
except httpx.RequestError:
logger.exception("Network error in get_system_health")
network_error_response: dict[str, Any] = {
"mcp_server": {"status": "healthy", "version": __version__},
"backend": {"status": "unreachable", "error": "Network error", "error_code": "NETWORK_ERROR"},
}
return [TextContent(type="text", text=serialize_response(network_error_response))]
except (KeyError, ValueError, TypeError) as e:
logger.exception("Data parsing error in get_system_health")
error_response = create_error_response(
f"Invalid data format: {e}", "DATA_ERROR"
)
return [TextContent(type="text", text=serialize_response(error_response))]
