flight_arrival_departure_schedule
Retrieve flight arrival and departure schedules for airports using IATA codes, airline names, and schedule types to monitor air traffic.
Instructions
MCP tool to get flight arrival and departure schedule.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| airport_iata_code | Yes | ||
| schedule_type | Yes | ||
| airline_name | Yes | ||
| number_of_flights | Yes |
Implementation Reference
- src/aviationstack_mcp/server.py:62-104 (handler)Handler function for 'flight_arrival_departure_schedule' tool, including @mcp.tool() decorator for registration. Fetches timetable data from AviationStack API, processes and returns JSON of flight schedules.@mcp.tool() def flight_arrival_departure_schedule( airport_iata_code: str, schedule_type: str, airline_name: str, number_of_flights: int ) -> str: """MCP tool to get flight arrival and departure schedule.""" try: data = fetch_flight_data( 'http://api.aviationstack.com/v1/timetable', {'iataCode': airport_iata_code, 'type': schedule_type, 'airline_name': airline_name} ) data_list = data.get('data', []) number_of_flights = min(number_of_flights, len(data_list)) # Sample random flights from the data list sampled_flights = random.sample(data_list, number_of_flights) filtered_flights = [] for flight in sampled_flights: filtered_flights.append({ 'airline': flight.get('airline').get('name'), 'flight_number': flight.get('flight').get('iataNumber'), 'departure_estimated_time': flight.get('departure').get('estimatedTime'), 'departure_scheduled_time': flight.get('departure').get('scheduledTime'), 'departure_actual_time': flight.get('departure').get('actualTime'), 'departure_terminal': flight.get('departure').get('terminal'), 'departure_gate': flight.get('departure').get('gate'), 'arrival_estimated_time': flight.get('arrival').get('estimatedTime'), 'arrival_scheduled_time': flight.get('arrival').get('scheduledTime'), 'arrival_airport_code': flight.get('arrival').get('iataCode'), 'arrival_terminal': flight.get('arrival').get('terminal'), 'arrival_gate': flight.get('arrival').get('gate'), 'departure_delay': flight.get('departure').get('delay'), }) return json.dumps(filtered_flights) if filtered_flights else ( f"No flights found for iata code '{airport_iata_code}'." ) except requests.RequestException as e: return f"Request error: {str(e)}" except (KeyError, ValueError, TypeError) as e: return f"Error fetching flight schedule: {str(e)}"
- Helper function used by the tool to fetch data from the AviationStack API, adding API key and handling requests.def fetch_flight_data(url: str, params: dict) -> dict: """Fetch flight data from the AviationStack API.""" api_key = os.getenv('AVIATION_STACK_API_KEY') if not api_key: raise ValueError("AVIATION_STACK_API_KEY not set in environment.") params = {'access_key': api_key, **params} response = requests.get(url, params=params, timeout=10) response.raise_for_status() return response.json()