NASA-MCP

Instructions

Implementation Reference

• src/nasa_mcp/server.py:212-271 (handler)

  The handler function for the 'get_asteroid_lookup' tool. It takes an asteroid_id parameter, queries the NASA NEO API, processes the response data including details like name, diameter, hazardous status, close approaches, and orbital data, and returns a formatted string summary.

  async def get_asteroid_lookup(asteroid_id: str) -> str:
      """Look up a specific asteroid based on its NASA JPL ID.
      
      Args:
          asteroid_id: Asteroid ID in the NASA JPL small body (SPK-ID) system.
      """
      url = f"{NASA_API_BASE}/neo/rest/v1/neo/{asteroid_id}"
      data = await make_nasa_request(url)
      
      if not data:
          return f"Could not retrieve data for asteroid ID {asteroid_id} due to a connection error."
      
      # Check for error response (must be a dictionary)
      if isinstance(data, dict) and "error" in data:
          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 Asteroid Lookup API. URL: {data.get('url')}"
  
      try:
          result = [
              f"ID: {data.get('id', 'Unknown')}",
              f"Name: {data.get('name', 'Unknown')}",
              f"Designation: {data.get('designation', 'Unknown')}",
              f"NASA JPL URL: {data.get('nasa_jpl_url', 'Not available')}",
              f"Absolute magnitude: {data.get('absolute_magnitude_h', 'Unknown')}",
              f"nEstimated diameter:",
              f"  Minimum (km): {data.get('estimated_diameter', {}).get('kilometers', {}).get('estimated_diameter_min', 'Unknown')}",
              f"  Maximum (km): {data.get('estimated_diameter', {}).get('kilometers', {}).get('estimated_diameter_max', 'Unknown')}",
              f"nPotentially hazardous: {'Yes' if data.get('is_potentially_hazardous_asteroid', False) else 'No'}",
              f"Sentry Object: {'Yes' if data.get('is_sentry_object', False) else 'No'}" # Corrected field name based on common API patterns
          ]
          
          # Approach information
          close_approaches = data.get('close_approach_data', [])
          if close_approaches:
              result.append("nClose approaches:")
              for i, approach in enumerate(close_approaches[:5], 1):  # Limit to first 5 to avoid overload
                  result.append(f"n  Approach {i}:")
                  result.append(f"  Date: {approach.get('close_approach_date_full', 'Unknown')}")
                  result.append(f"  Orbiting body: {approach.get('orbiting_body', 'Unknown')}")
                  result.append(f"  Distance (km): {approach.get('miss_distance', {}).get('kilometers', 'Unknown')}")
                  result.append(f"  Relative velocity (km/h): {approach.get('relative_velocity', {}).get('kilometers_per_hour', 'Unknown')}")
              
              if len(close_approaches) > 5:
                  result.append(f"n  ... and {len(close_approaches) - 5} more approaches.")
          
          # Orbital data
          orbital_data = data.get('orbital_data', {})
          if orbital_data:
              result.append("nOrbital data:")
              result.append(f"  Orbit determination date: {orbital_data.get('orbit_determination_date', 'Unknown')}") # Corrected field name
              result.append(f"  Semi-major axis: {orbital_data.get('semi_major_axis', 'Unknown')} AU")
              result.append(f"  Eccentricity: {orbital_data.get('eccentricity', 'Unknown')}")
              result.append(f"  Inclination: {orbital_data.get('inclination', 'Unknown')} degrees")
              result.append(f"  Orbital period: {orbital_data.get('orbital_period', 'Unknown')} days")
          
          return "n".join(result)
      except Exception as e:
          logger.error(f"Error processing Asteroid Lookup data: {str(e)}")
          return f"Error processing asteroid data: {str(e)}"

• src/nasa_mcp/server.py:28-83 (helper)

  Helper function used by the get_asteroid_lookup tool (and others) to perform HTTP GET requests to NASA APIs, handle authentication with API key, parse JSON or detect binary content, and manage various error conditions with detailed logging.

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

• src/nasa_mcp/server.py:212-212 (registration)

  The @mcp.tool() decorator registers the get_asteroid_lookup function as an MCP tool.

  async def get_asteroid_lookup(asteroid_id: str) -> str:

• src/nasa_mcp/server.py:213-217 (schema)

  The function signature and docstring define the input schema (asteroid_id: str) and output (str) for the tool, used by MCP for validation.

  """Look up a specific asteroid based on its NASA JPL ID.
  
  Args:
      asteroid_id: Asteroid ID in the NASA JPL small body (SPK-ID) system.
  """