NASA-MCP

Instructions

Implementation Reference

• src/nasa_mcp/server.py:438-493 (handler)

  The handler function implementing the 'get_solar_flare' MCP tool. It queries NASA's DONKI/FLR API for solar flare data within specified date range, processes the JSON response, and formats recent flares with details like ID, times, class, location, and linked events.
  async def get_solar_flare(start_date: str = None, end_date: str = None) -> str: """Get solar flare (FLR) data. Args: start_date: Start date in YYYY-MM-DD format. Defaults to 30 days before current date. end_date: End date in YYYY-MM-DD format. Defaults to current date. """ params = {} if start_date: params["startDate"] = start_date if end_date: params["endDate"] = end_date url = f"{NASA_API_BASE}/DONKI/FLR" data = await make_nasa_request(url, params) if not data: return "Could not retrieve FLR data 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 FLR API. URL: {data.get('url')}" try: # Ensure data is a list if not isinstance(data, list): logger.error(f"Unexpected non-list response from FLR API: {data}") return "Received unexpected data format from FLR API." # Add specific FLR fields to the formatter if needed, or format here if not data: return "No solar flare data for the specified period." result = [f"Solar Flares found: {len(data)}"] display_limit = 10 count = 0 for flare in data: if count >= display_limit: result.append(f"n... and {len(data) - display_limit} more entries.") break result.append(f"nID: {flare.get('flrID', 'Unknown')}") result.append(f"Begin Time: {flare.get('beginTime', 'Unknown')}") result.append(f"Peak Time: {flare.get('peakTime', 'Unknown')}") result.append(f"End Time: {flare.get('endTime', 'Unknown')}") result.append(f"Class: {flare.get('classType', 'Unknown')}") result.append(f"Source Location: {flare.get('sourceLocation', 'Unknown')}") linked_events = flare.get('linkedEvents', []) if linked_events: result.append(" Related event IDs (first 5):") result.append(" " + ", ".join([le.get('activityID', 'N/A') for le in linked_events[:5]])) result.append("-" * 40) count += 1 return "n".join(result) except Exception as e: logger.error(f"Error processing FLR data: {str(e)}") return f"Error processing solar flare data: {str(e)}"
• src/nasa_mcp/server.py:28-83 (helper)

  Helper function used by get_solar_flare (and other tools) to make HTTP requests to NASA APIs, handling JSON, binary responses, errors, and adding 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)}