NASA-MCP

Instructions

Implementation Reference

• src/nasa_mcp/server.py:810-876 (handler)

  The handler function decorated with @mcp.tool(), implementing the core logic for querying NASA's planetary/earth/assets API to retrieve available Earth imagery assets for given latitude, longitude, date, and dimension parameters. Handles API requests, errors, and formats the response.
  @mcp.tool() async def get_earth_assets(lat: float, lon: float, date: str, dim: float = 0.025) -> str: """Get information about available imagery assets for a specific location and date. Args: lat: Latitude. lon: Longitude. date: Date in YYYY-MM-DD format. dim: Width and height of the image in degrees (0.025 degrees is approximately 2.7 km). """ params = { "lat": lat, "lon": lon, "date": date, "dim": dim } url = f"{NASA_API_BASE}/planetary/earth/assets" # This endpoint is expected to return JSON based on docs data = await make_nasa_request(url, params) if not data: return "Could not retrieve Earth asset data due to a connection error." # Check for error response (must be a dictionary) if isinstance(data, dict) and "error" in data: # Handle specific case where API returns error message if isinstance(data.get("details"), str) and "No Landsat 8 assets found" in data["details"]: return f"No Landsat 8 assets found for the specified location and date range (Lat: {lat}, Lon: {lon}, Date: {date})." return f"API Error: {data.get('error')} - Details: {data.get('details', 'N/A')}" if isinstance(data, dict) and data.get("binary_content"): return f"Received unexpected binary content from Earth Assets API. URL: {data.get('url')}" try: # The structure might have changed, docs mention 'results' array now? # Let's assume the old structure for now based on previous code # If 'results' is the key for a list: assets = data.get('results', []) # If it's a single result: assets = [data] if data else [] # Based on the docs example, it seems to return a single object with 'id' and 'dataset' # Let's refine based on the example query structure asset_id = data.get('id') asset_dataset = data.get('dataset') asset_date = data.get('date', 'Unknown').split('T')[0] # Extract date part if not asset_id: # Check if response is like {'count': 0, 'results': []} if data.get('count') == 0 and isinstance(data.get('results'), list): return f"No Landsat 8 assets found for the specified location and date range (Lat: {lat}, Lon: {lon}, Date: {date})." else: logger.warning(f"Unexpected JSON structure from Earth Assets API: {data}") return "No asset ID found in the response." result = [ f"Asset found for coordinates: Lat {lat}, Lon {lon}", f"Date (closest): {asset_date}", f"ID: {asset_id}", f"Dataset: {asset_dataset}" # The API doesn't seem to return a direct URL here anymore based on docs ] return "n".join(result) except Exception as e: logger.error(f"Error processing Earth Assets data: {str(e)}") return f"Error processing asset data: {str(e)}"
• src/nasa_mcp/server.py:28-82 (helper)

  Shared helper function used by get_earth_assets (and other tools) to make HTTP requests to NASA APIs, handling JSON, binary responses, errors, and adding the API key.
  async def make_nasa_request(url: str, params: dict = None) -> Union[dict[str, Any], List[Any], None]: """Make a request to the NASA API with proper error handling. Handles both JSON and binary (image) responses. """ logger.info(f"Making request to: {url} with params: {params}") if params is None: params = {} # Ensure API key is included in parameters if "api_key" not in params: params["api_key"] = API_KEY async with httpx.AsyncClient() as client: try: response = await client.get(url, params=params, timeout=30.0, follow_redirects=True) response.raise_for_status() # Raise an exception for bad status codes (4xx or 5xx) content_type = response.headers.get("Content-Type", "").lower() if "application/json" in content_type: try: return response.json() except json.JSONDecodeError as json_err: logger.error(f"JSON decode error for URL {response.url}: {json_err}") logger.error(f"Response text: {response.text[:500]}") # Log beginning of text return {"error": "Failed to decode JSON response", "details": str(json_err)} elif content_type.startswith("image/"): logger.info(f"Received binary image content ({content_type}) from {response.url}") # Return a dictionary indicating binary content was received return { "binary_content": True, "content_type": content_type, "url": str(response.url) # Return the final URL after redirects } else: # Handle other unexpected content types logger.warning(f"Unexpected content type '{content_type}' received from {response.url}") return {"error": f"Unexpected content type: {content_type}", "content": response.text[:500]} except httpx.HTTPStatusError as http_err: logger.error(f"HTTP error occurred: {http_err} - {http_err.response.status_code} for URL {http_err.request.url}") try: # Try to get more details from response body if possible error_details = http_err.response.json() except Exception: error_details = http_err.response.text[:500] return {"error": f"HTTP error: {http_err.response.status_code}", "details": error_details} except httpx.RequestError as req_err: logger.error(f"Request error occurred: {req_err} for URL {req_err.request.url}") return {"error": "Request failed", "details": str(req_err)} except Exception as e: logger.error(f"An unexpected error occurred: {str(e)}") return {"error": "An unexpected error occurred", "details": str(e)}