NASA-MCP

Instructions

Implementation Reference

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

  Handler function for the 'get_asteroid_lookup' tool. This is the core implementation that fetches asteroid data from NASA's NEO API using the provided asteroid_id, processes the response, and formats it for output. Registered via @mcp.tool() decorator.
  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 get_asteroid_lookup (and other tools) to make HTTP requests to NASA APIs, handle JSON/binary responses, errors, and API key injection.
  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)}