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)}