Skip to content

mcp_server

FastMCP server for KMA Weather APIs.

This module provides an MCP (Model Context Protocol) server that exposes KMA weather observation data through standardized tools, including: - Surface observations (ASOS, AWS, Climate, Dust, UV, Snow, etc.) - Weather forecasts and warnings

main()

Initialize and run the MCP server with API key validation.

Source code in python/src/kma_mcp/mcp_server.py
def main() -> None:
    """Initialize and run the MCP server with API key validation."""
    logger.info('Starting KMA MCP server...')

    # Validate API key on startup
    if not API_KEY:
        logger.warning('KMA_API_KEY environment variable not set')
        logger.warning('Server will start but API calls will fail')
    else:
        logger.info('Validating API key...')
        if validate_api_key(API_KEY):
            logger.info('API key is valid and working')
        else:
            logger.error('API key validation failed - API calls may not work properly')
            logger.error('Please check your API key at https://apihub.kma.go.kr/')

    # Initialize and run the server
    logger.info('Server initialized successfully')
    mcp.run(transport='stdio')

validate_api_key(api_key)

Validate API key by making a simple API call.

Uses AWS minutely data API as a lightweight validation endpoint.

Parameters:

  • api_key (str) –

    KMA API key to validate

Returns:

  • bool

    True if API key is valid, False otherwise

Source code in python/src/kma_mcp/mcp_server.py
def validate_api_key(api_key: str) -> bool:
    """Validate API key by making a simple API call.

    Uses AWS minutely data API as a lightweight validation endpoint.

    Args:
        api_key: KMA API key to validate

    Returns:
        True if API key is valid, False otherwise
    """
    if not api_key:
        logger.error('API key is empty')
        return False

    try:
        # Use AWS minutely data for validation (lightweight endpoint)
        with AWSClient(api_key) as client:
            # Get data from 10 minutes ago to ensure data availability
            test_time = datetime.now(UTC) - timedelta(minutes=10)
            # Test with a single station (104 = Bukgangneung)
            result = client.get_minutely_data(tm2=test_time, stn=104)

            # Check if we got valid data back
            if result and not isinstance(result, str):
                logger.info('API key validation successful')
                return True
            else:
                logger.error('API key validation failed: %s', result)
                return False

    except Exception as e:  # noqa: BLE001
        logger.error('API key validation error: %s', e)
        return False