Skip to content

async_mcp_server

Async FastMCP server for KMA Weather APIs.

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

main() async

Initialize and run the async MCP server with API key validation.

Source code in python/src/kma_mcp/async_mcp_server.py 
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
async def main() -> None:
    """Initialize and run the async MCP server with API key validation."""
    logger.info('Starting KMA Async 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 await 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) async

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/async_mcp_server.py 
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
async 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)
        async with AsyncAWSClient(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 = await 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