Rought Flight MCP Server application

app.py

@@ -1,38 +1,363 @@
+import os
+from dotenv import load_dotenv
+import requests
+import re
 import json
+from datetime import date
 import gradio as gr
-from textblob import TextBlob
+from forex_python.converter import CurrencyRates
 
-def sentiment_analysis(text: str) -> str:
+# Load environment variables from .env.local
+load_dotenv(dotenv_path='.env.local')
+
+def get_access_token() -> str|None:
     """
-    Analyze the sentiment of the given text.
+    Get the Access Token required to make requests on the Amadeus API
 
     Args:
-        text (str): The text to analyze
+    
+    Returns:
+        str | None: A string containing the Access Token, or None if an error occurred.
+    """
+    api_key = os.environ.get("AMADEUS_API")
+    api_secret = os.environ.get("AMADEUS_SECRET")
+
+    # Check if env variables loaded
+    if not api_key or not api_secret:
+        print("Error: AMADEUS_API or AMADEUS_SECRET not found in .env.local")
+        return None
+
+    # Getting the Access Token to use for the API
+    token_url = "https://test.api.amadeus.com/v1/security/oauth2/token"
+    headers = {
+        "Content-Type": "application/x-www-form-urlencoded"
+    }
+    data = {
+        "grant_type": "client_credentials",
+        "client_id": api_key,
+        "client_secret": api_secret
+    }
+
+    try:
+        response = requests.post(token_url, headers=headers, data=data)
+        response.raise_for_status()
+        
+        token_data = response.json()
+        
+        # Check the 'state' field in the response JSON
+        if token_data.get('state') != 'approved':
+            print(f"Authorization not approved. State: {token_data.get('state')}")
+            return None
+        
+        print("Authorization approved. Token received.")
+        return token_data.get("access_token")
+
+    except requests.exceptions.RequestException as e:
+        print(f"An error occurred during token request: {e}")
+        return None
+    except KeyError:
+        print("Error: 'access_token' not found in the response.")
+        return None
+
+GLOBAL_TOKEN = get_access_token()
+
+def get_date(year, month, day):
+    """
+    Returns date specified in ISO 8601
 
+    Args:
+        year (int): The year of the date.
+        month (int): The month of the date.
+        day (int): The day of the date.
+
+    Returns:
+        str: A string representing the date in ISO 8601 format.
+    
+    """
+    return date(year, month, day).isoformat()
+
+
+def get_flight_offers(departure_year: int, departure_month: int, departure_day: int, departureLocationCode: str, arrivalLocationCode: str) -> dict|None:
+    """
+    Obtain information about the flights offered between two locations from 
+    various airlines like Frontier Airlines or United Airlines, but critically 
+    does not contain flights from American Airlines or Delta Airlines.
+
+    Args:
+        departure_year (int): The year of the departure date.
+        departure_month (int): The month of the departure date.
+        departure_day (int): The day of the departure date.
+        departureLocationCode (str): The IATA code of the departure location.
+        arrivalLocationCode (str): The IATA code of the arrival location
+    
     Returns:
-        str: A JSON string containing polarity, subjectivity, and assessment
+        dict: A JSON dictionary containing the flight offers and information about them.
     """
-    blob = TextBlob(text)
-    sentiment = blob.sentiment
     
-    result = {
-        "polarity": round(sentiment.polarity, 2),  # -1 (negative) to 1 (positive)
-        "subjectivity": round(sentiment.subjectivity, 2),  # 0 (objective) to 1 (subjective)
-        "assessment": "positive" if sentiment.polarity > 0 else "negative" if sentiment.polarity < 0 else "neutral"
+    if not GLOBAL_TOKEN:
+        print("Cannot get flight offers without a valid access token.")
+        return
+
+    offer_url = "https://test.api.amadeus.com/v2/shopping/flight-offers" # Note: v2 for flight offers
+    headers = {
+        "Authorization": f"Bearer {GLOBAL_TOKEN}" # Correctly format the Authorization header
     }
+    # For GET requests, parameters are passed in 'params', not 'data'
+    params = {
+        "originLocationCode":departureLocationCode,
+        "destinationLocationCode":arrivalLocationCode,
+        "departureDate":get_date(departure_year, departure_month, departure_day),
+        "adults":1
+    }
+
+    try:
+        response = requests.get(offer_url, headers=headers, params=params)
+        response.raise_for_status()
+
+        # if write_to_file:
+        #     with open("api_output.json", "w") as file:
+        #         json.dump(response.json(), file, indent=2)
+
+        return response.json()
+    except requests.exceptions.RequestException as e:
+        print(f"An error occurred during flight offer request: {e}")
+
+
+def path_is_direct(flight_path: dict) -> bool:
+    """
+    Determines if a path is a direct flight
+    
+    Args:
+        flight_path (dict): A dictionary representing the flight path.
+    
+    Returns:
+        bool: True if the path is a direct flight, False otherwise.
+    """
+    return len(list(flight_path.keys())) == 1
+
+
+def parse_duration_to_minutes(duration_str: str) -> int:
+    """
+    Parses an ISO 8601 duration string (e.g., 'PT2H8M') and returns the total duration in minutes.
+    
+    Args:
+        duration_str (str): The ISO 8601 duration string.
+
+    Returns:
+        int: The total duration in minutes.
+    """
+    if not duration_str.startswith('PT'):
+        return 0
+
+    hours = 0
+    minutes = 0
+
+    hours_match = re.search(r'(\d+)H', duration_str)
+    if hours_match:
+        hours = int(hours_match.group(1))
+
+    minutes_match = re.search(r'(\d+)M', duration_str)
+    if minutes_match:
+        minutes = int(minutes_match.group(1))
+
+    return (hours * 60) + minutes
+
+
+def get_carrier_name(carrier_dict: dict[str, str], carrier_code: str) -> str:
+    """
+    From an IATA carrier code, returns the full name of the airline carrier
+
+    Args:
+        carrier_dict (dict[str, str]): A dictionary with keys of the IATA code 
+            and values of the full airline carrier name.
+        carrier_code (str): The IATA code of the airline carrier.
+
+    Returns:
+        str: The full name of the airline carrier.
+    """
+    return carrier_dict.get(carrier_code, "Unknown")
+
+
+def get_flight_paths(offers_json: dict) -> list[dict]:
+    """
+    Parses flight offers and returns a list of dictionaries,
+    each representing the path of a flight schedule and 
+    info about the flight path.
+
+    Args:
+        offers_json (dict): A JSON dictionary containing the flight offers.
+
+    Returns:
+        list[dict]: A list of dictionaries representing the flight paths and info about the flights. 
+        Each dictionary contains info on if the flight is a direct flight, 
+        the flight price, currency, duration in minutes, airline carrier, 
+        and flight path, all separated by id.
+    """
+    if not offers_json or 'data' not in offers_json:
+        return []
+
+    all_paths = []
+    for offer in offers_json['data']:
+        # Itineraries are a list, we'll process the first one.
+        if not offer.get('itineraries') or not offer.get('id'):
+            continue
+        itinerary = offer['itineraries'][0]
+        path = {}
+        path['id'] = offer['id']
+        path['price'] = offer['price']['total']
+        path['currency'] = offer['price']['currency']
+        path['duration_in_minutes'] = parse_duration_to_minutes(itinerary.get('duration', ''))
+        path['airline_carrier'] = get_carrier_name(
+            offers_json["dictionaries"]["carriers"], 
+            offer['validatingAirlineCodes'][0]
+        )
+        path['flight_paths'] = {}
+
+        for segment in itinerary['segments']:
+            departure_code = segment['departure']['iataCode']
+            arrival_code = segment['arrival']['iataCode']
+            path['flight_paths'][departure_code] = arrival_code
+
+        path['is_direct_flight'] = path_is_direct(path['flight_paths'])
+        all_paths.append(path)
+    return all_paths
+
+def sort_flight_offers(unsorted_flight_offers_json: str, sort_by: str) -> list[dict]:
+    """
+    Sorts a list of flight offer dictionaries by a specified key.
+
+    Args:
+        flight_offers_json (str): A JSON string representing the list of flight offers from get_flight_paths().
+        sort_by (str): The string of the key to sort by. Can be 'price' or 'duration_in_minutes'.
+
+    Returns:
+        list[dict]: A new list of flight offers sorted by the specified key.
+
+    Raises:
+        ValueError: If sort_by is not 'price' or 'duration_in_minutes'.
+    """
+    flight_offers = json.loads(unsorted_flight_offers_json)
+    if sort_by not in ['price', 'duration_in_minutes']:
+        raise ValueError("Can only sort by 'price' or 'duration_in_minutes'.")
+
+    return sorted(flight_offers, key=lambda x: x[sort_by])
+
+def remove_non_direct(all_flight_offers_json: str) -> list[dict]:
+    """
+    Returns a list of the flight offers that are direct flights.
+
+    Args:
+        all_flight_offers_json (str): A JSON string representing the list of flight offers from get_flight_paths().
+
+    Returns:
+        list[dict]: A new list of flight offers that are all direct flights.
+    """
+    if all_flight_offers_json is None:
+        return []
+    all_flight_offers = json.loads(all_flight_offers_json)
+    return [offer for offer in all_flight_offers if offer.get('is_direct_flight')]
+
+def convert_eur_to_usd(price_in_eur: float) -> float:
+    """
+    Converts a price from Euros to US Dollars using the current exchange rate.
+
+    Args:
+        price_in_eur (float): The price in Euros.
+
+    Returns:
+        float: The price in US Dollars.
+    """
+    c = CurrencyRates()
+    exchange_rate = c.get_rate('EUR', 'USD')
+    price_in_usd = price_in_eur * exchange_rate
+    return price_in_usd
+
+def print_paths():
+    global GLOBAL_TOKEN
+    # GLOBAL_TOKEN = get_access_token()
+    GLOBAL_TOKEN = None
+    if GLOBAL_TOKEN:
+        offers_json = get_flight_offers(2025, 9, 18, "CLT", "ORD")
+    else:
+        print("Using saved info")
+        with open("api_output.json", "r") as f:
+            offers_json = json.loads(f.read())
+
+    if offers_json:
+        flight_paths = get_flight_paths(offers_json)
+        for i, path in enumerate(flight_paths, 1):
+            print(f"Offer {i} Path: {path}")
+
+def get_flight_offers_and_paths(
+    departure_year: int, 
+    departure_month: int, 
+    departure_day: int, 
+    departureLocationCode: str, 
+    arrivalLocationCode: str
+) -> list[dict] | None:
+    """
+    A wrapper function that first gets flight offers and then immediately
+    parses them to get the flight paths.
+
+    Args:
+        departure_year (int): The year of the departure date.
+        departure_month (int): The month of the departure date.
+        departure_day (int): The day of the departure date.
+        departureLocationCode (str): The IATA code of the departure location.
+        arrivalLocationCode (str): The IATA code of the arrival location
+
+    Returns:
+        list[dict]: A list of dictionaries representing the flight paths and info about the flights. 
+        Each dictionary contains info on if the flight is a direct flight, 
+        the flight price, currency, duration in minutes, airline carrier, 
+        and flight path, all separated by id.
+    """
 
-    return json.dumps(result)
+    offers_json = get_flight_offers(
+        departure_year, departure_month, departure_day, 
+        departureLocationCode, arrivalLocationCode
+    )
+    if offers_json:
+        return get_flight_paths(offers_json)
+    return None
 
-# Create the Gradio interface
-demo = gr.Interface(
-    fn=sentiment_analysis,
-    inputs=gr.Textbox(placeholder="Enter text to analyze..."),
-    outputs=gr.Textbox(),  # Changed from gr.JSON() to gr.Textbox()
-    title="Text Sentiment Analysis",
-    description="Analyze the sentiment of text using TextBlob",
-    api_name="sentiment_analysis"
-)
+with gr.Blocks() as demo:
+    gr.Markdown("# Amadeus Flight Offers API")
+    with gr.Tab("Get Flight Offers"):
+        with gr.Row():
+            dep_year = gr.Number(label="Departure Year", value=2025)
+            dep_month = gr.Number(label="Departure Month", value=9)
+            dep_day = gr.Number(label="Departure Day", value=18)
+        with gr.Row():
+            dep_loc = gr.Textbox(label="Departure Location Code", value="CLT")
+            arr_loc = gr.Textbox(label="Arrival Location Code", value="ORD")
+        get_offers_btn = gr.Button("Get Offers")
+        offers_output = gr.JSON(label="Flight Offers")
+        get_offers_btn.click(
+            fn=get_flight_offers_and_paths,
+            inputs=[dep_year, dep_month, dep_day, dep_loc, arr_loc],
+            outputs=offers_output,
+            api_name="get_flight_offers_and_paths"
+        )
+    with gr.Tab("Sort Flight Offers"):
+        sort_input_offers = gr.Textbox(label="Flight Offers (JSON from 'Get Flight Offers' tab)")
+        sort_by = gr.Radio(['price', 'duration_in_minutes'], label="Sort By", value='price')
+        sort_button = gr.Button("Sort Offers")
+        sorted_offers_output = gr.JSON(label="Sorted Flight Offers")
+        sort_button.click(fn=sort_flight_offers, inputs=[sort_input_offers, sort_by], outputs=sorted_offers_output, api_name="sort_flight_offers")
+    with gr.Tab("Filter Direct Flight Offers"):
+        filter_input_offers = gr.Textbox(label="Flight Offers (JSON from 'Get Flight Offers' tab)")
+        filter_button = gr.Button("Filter Offers")
+        filter_offers_output = gr.JSON(label="Filtered Flight Offers")
+        sort_button.click(fn=remove_non_direct, inputs=[filter_input_offers], outputs=filter_offers_output, api_name="remove_non_direct")
+    with gr.Tab("Convert EUR to USD"):
+        eur_input = gr.Number(label="Price in EUR")
+        convert_button = gr.Button("Convert to USD")
+        usd_output = gr.Number(label="Price in USD")
+        convert_button.click(
+            fn=convert_eur_to_usd, inputs=[eur_input], outputs=usd_output, api_name="convert_eur_to_usd"
+        )
 
 # Launch the interface and MCP server
 if __name__ == "__main__":
-    demo.launch(mcp_server=True)
+    demo.launch(mcp_server=True)

requirements.txt

@@ -1,2 +1,3 @@
 gradio[mcp]
-textblob
+textblob
+forex-python