flight_arrival_departure_schedule
Retrieve real-time flight arrival and departure schedules by specifying the airport IATA code, schedule type, airline name, and number of flights using the Aviationstack MCP Server.
Instructions
MCP tool to get flight arrival and departure schedule.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| airline_name | Yes | ||
| airport_iata_code | Yes | ||
| number_of_flights | Yes | ||
| schedule_type | Yes |
Implementation Reference
- src/aviationstack_mcp/server.py:62-104 (handler)Handler function decorated with @mcp.tool() implementing the flight_arrival_departure_schedule tool. Fetches timetable data from AviationStack API, samples random flights, extracts schedule details, and returns JSON.@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 to fetch data from AviationStack API using the API key from environment, adds access_key to params, makes GET request, and returns JSON response. Used by the tool handler.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()
- src/aviationstack_mcp/server.py:11-11 (registration)Initialization of the FastMCP server instance. Tool registration occurs via @mcp.tool() decorators on handler functions.mcp = FastMCP("Aviationstack MCP")