doomsweek_mcp_server.py

# -*- coding: utf-8 -*-
 
from mcp.server.fastmcp import FastMCP
import gradio as gr
import io
import json
import os
import re
import sys

import requests

sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', errors='replace')
sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding='utf-8', errors='replace')

mcp = FastMCP("doomsweek_mcp_server")

def calc_asteroid_factor(data, weight=1.0):
    """
    Calculates the asteroid doom probability.

    Args:
        data (object): The data structure returned from NASA
        weight (float): The weight to apply to the hazard factor.
    
    Returns:
        Weighted factor for Near Earth Object impact for the next seven days.    
    """

    objects = [obj for sublist in data["near_earth_objects"].values() for obj in sublist] 

    num_objects = len(objects)

    hazardous_objects = [obj for sublist in data["near_earth_objects"].values() for obj in sublist if obj["is_potentially_hazardous_asteroid"] == True]

    num_hazardous_objects = len(hazardous_objects)

    return (num_hazardous_objects/num_objects * weight) # hazard factor 


def fetch_asteroid_data(url, api_key):
    """
    Fetches data from the NASA Near Earth Object Web Service.
    
    Returns:
        Near Earth Objects for the next seven days.    
    """

    request_data = requests.get(
        url,
        params={"api_key": api_key},  
    )

    if request_data.status_code != 200:    
        return None
    else:    
        return json.loads(request_data.content)
    
@mcp.tool()
async def calc_doom_probability():
    """
    Calculates the overall doom probability.
    
    Returns:
       The overall doom probability for the next seven days.    
    """

    nasa_url = "https://api.nasa.gov/neo/rest/v1/feed" 
    
    if not nasa_api_key:
        return "ERROR: NASA_API_KEY not set."
    else: 
        nasa_data = fetch_asteroid_data(nasa_url, nasa_api_key)
        if not nasa_data:
            return "ERROR: Unable to fetch data from NASA API."
        else:
            return calc_asteroid_factor(nasa_data, weight=1.0)
        
# demo = gr.Interface(fn=calc_doom_probability, inputs=None, outputs="text")
# demo.launch(mcp_server=True)

if __name__ == "__main__":
    # NOTE the NASA API is provided by the client via env server parameters
    nasa_api_key = os.getenv("NASA_API_KEY")
    mcp.run(transport='stdio')