NASA-MCP

Instructions

Implementation Reference

• src/nasa_mcp/server.py:367-400 (handler)

  The handler function decorated with @mcp.tool() that implements the get_coronal_mass_ejection tool. It queries the NASA DONKI/CME endpoint, handles responses, and formats output using the format_donki_results helper.
  @mcp.tool() async def get_coronal_mass_ejection(start_date: str = None, end_date: str = None) -> str: """Get coronal mass ejection (CME) 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/CME" data = await make_nasa_request(url, params) if not data: return "Could not retrieve CME 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 CME 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 CME API: {data}") return "Received unexpected data format from CME API." return format_donki_results(data, "Coronal Mass Ejections", "activityID") except Exception as e: logger.error(f"Error processing CME data: {str(e)}") return f"Error processing coronal mass ejection data: {str(e)}"
• src/nasa_mcp/server.py:318-365 (helper)

  Supporting utility function format_donki_results used by get_coronal_mass_ejection to consistently format lists of DONKI event data, including CME-specific handling.
  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)
• src/nasa_mcp/server.py:367-367 (registration)

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