NASA-MCP

Instructions

Implementation Reference

• src/nasa_mcp/server.py:496-528 (handler)

  The core handler function for the 'get_solar_energetic_particle' tool. It makes an API request to NASA's DONKI/SEP endpoint, handles errors, and uses the format_donki_results helper to produce a formatted string output of solar energetic particle events.
  async def get_solar_energetic_particle(start_date: str = None, end_date: str = None) -> str: """Get solar energetic particle (SEP) 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/SEP" data = await make_nasa_request(url, params) if not data: return "Could not retrieve SEP 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 SEP API. URL: {data.get('url')}" try: # Ensure data is a list for format_donki_results if not isinstance(data, list): logger.error(f"Unexpected non-list response from SEP API: {data}") return "Received unexpected data format from SEP API." return format_donki_results(data, "Solar Energetic Particle Events", "sepID") except Exception as e: logger.error(f"Error processing SEP data: {str(e)}") return f"Error processing solar energetic particle data: {str(e)}"
• src/nasa_mcp/server.py:318-365 (helper)

  Supporting helper function format_donki_results that formats the list of DONKI events (including SEP events) into a readable string output, limiting display to 10 entries and handling common fields like linked events.
  def format_donki_results(data: list, title_prefix: str, id_key: str) -> str: if not data: return f"No {title_prefix.lower()} data for the specified period." result = [f"{title_prefix} found: {len(data)}"] display_limit = 10 count = 0 for item in data: if count >= display_limit: result.append(f"n... and {len(data) - display_limit} more entries.") break result.append(f"nID: {item.get(id_key, 'Unknown')}") # Add common fields if they exist if 'startTime' in item: result.append(f"Start Time: {item.get('startTime', 'Unknown')}") if 'eventTime' in item: result.append(f"Event Time: {item.get('eventTime', 'Unknown')}") if 'sourceLocation' in item: result.append(f"Source Location: {item.get('sourceLocation', 'Unknown')}") if 'note' in item: result.append(f"Note: {item.get('note', 'N/A')}") if 'link' in item: result.append(f"Link: {item.get('link', 'N/A')}") # Specific fields for different DONKI types can be added here if needed # Example for CME: if id_key == 'activityID' and 'cmeAnalyses' in item: analyses = item.get('cmeAnalyses', []) if analyses: result.append(" Analyses:") for analysis in analyses[:2]: # Limit analyses shown result.append(f" - Time: {analysis.get('time21_5', 'N/A')}, Speed: {analysis.get('speed', 'N/A')} km/s, Type: {analysis.get('type', 'N/A')}") # Example for GST: if id_key == 'gstID' and 'allKpIndex' in item: kp_indices = item.get('allKpIndex', []) if kp_indices: result.append(" Kp Indices (first 2):") for kp in kp_indices[:2]: result.append(f" - Time: {kp.get('observedTime', 'N/A')}, Index: {kp.get('kpIndex', 'N/A')}") # Linked Events linked_events = item.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)