future_flights_arrival_departure_schedule

Instructions

Implementation Reference

• src/aviationstack_mcp/server.py:517-565 (handler)

  Core handler function that fetches future flight arrival/departure schedule data from the Aviationstack 'flightsFuture' API endpoint. Validates inputs (positive int, ISO date, schedule type), calls fetch_flight_data, samples results, and returns a JSON string of filtered flight details (airline, flight number, scheduled times, arrival airport/terminal/gate, aircraft).

  def future_flights_arrival_departure_schedule(
      airport_iata_code: str,
      schedule_type: str,
      airline_iata: str,
      date: str,
      number_of_flights: int,
  ) -> str:
      """Get a random sample of future flights for an airport and date."""
      try:
          _validate_positive_int(number_of_flights, "number_of_flights")
          _validate_iso_date(date, "date")
          normalized_schedule_type = schedule_type.lower()
          if normalized_schedule_type not in {"arrival", "departure"}:
              raise ValueError("'schedule_type' must be either 'arrival' or 'departure'.")
  
          params: dict[str, Any] = {
              "iataCode": airport_iata_code,
              "type": normalized_schedule_type,
              "date": date,
          }
          if airline_iata:
              params["airline_iata"] = airline_iata
  
          data = fetch_flight_data("flightsFuture", params)
          sampled_flights = _sample_data(data.get("data", []), number_of_flights)
  
          filtered_flights = []
          for flight in sampled_flights:
              filtered_flights.append(
                  {
                      "airline": _safe_get(flight, "airline", "name"),
                      "flight_number": _safe_get(flight, "flight", "iataNumber"),
                      "departure_scheduled_time": _safe_get(
                          flight, "departure", "scheduledTime"
                      ),
                      "arrival_scheduled_time": _safe_get(flight, "arrival", "scheduledTime"),
                      "arrival_airport_code": _safe_get(flight, "arrival", "iataCode"),
                      "arrival_terminal": _safe_get(flight, "arrival", "terminal"),
                      "arrival_gate": _safe_get(flight, "arrival", "gate"),
                      "aircraft": _safe_get(flight, "aircraft", "modelText"),
                  }
              )
          if not filtered_flights:
              return f"No flights found for iata code '{airport_iata_code}'."
          return json.dumps(filtered_flights)
      except requests.RequestException as exc:
          return _error_response("fetching flight future schedule", exc)
      except (KeyError, ValueError, TypeError) as exc:
          return _error_response("fetching flight future schedule", exc)

• src/aviationstack_mcp/server.py:130-158 (schema)

  Pydantic input schema (FutureFlightsArrivalDepartureScheduleInput) for the tool. Defines fields: airport_iata_code (required), schedule_type (required, 'arrival'/'departure'), airline_iata (optional), date (required YYYY-MM-DD), number_of_flights (required, >0). Uses extra='forbid'.

  class FutureFlightsArrivalDepartureScheduleInput(BaseModel):
      """Input schema for future_flights_arrival_departure_schedule tool."""
  
      model_config = ConfigDict(extra="forbid")
  
      airport_iata_code: str = Field(
          ...,
          description="Airport IATA code (for example: SFO).",
          min_length=1,
      )
      schedule_type: str = Field(
          ...,
          description="Schedule type: arrival or departure.",
          examples=["arrival", "departure"],
      )
      airline_iata: str = Field(
          default="",
          description="Optional airline IATA code filter (for example: UA).",
      )
      date: str = Field(
          ...,
          description="Future date in YYYY-MM-DD format.",
          examples=["2026-03-01"],
      )
      number_of_flights: int = Field(
          ...,
          description="Number of random flights to return.",
          gt=0,
      )

• src/aviationstack_mcp/server.py:897-931 (registration)

  MCP tool registration via @mcp.tool decorator. Wraps the handler with FastMCP tool metadata (name='future_flights_arrival_departure_schedule', description). The wrapper function validates inputs through the Pydantic schema, then delegates to the core handler.

  @mcp.tool(
      name="future_flights_arrival_departure_schedule",
      description="Return future arrival or departure schedule samples for an airport and date.",
  )
  def future_flights_arrival_departure_schedule_tool(
      airport_iata_code: Annotated[
          str, Field(description="Airport IATA code (for example: SFO).", min_length=1)
      ],
      schedule_type: Annotated[str, Field(description="Schedule type: arrival or departure.")],
      airline_iata: Annotated[
          str, Field(description="Optional airline IATA code filter (for example: UA).")
      ] = "",
      date: Annotated[
          str,
          Field(description="Future date in YYYY-MM-DD format.", examples=["2026-03-01"]),
      ] = "",
      number_of_flights: Annotated[
          int, Field(description="Number of random flights to return.", gt=0)
      ] = 5,
  ) -> str:
      """Tool wrapper for future_flights_arrival_departure_schedule."""
      validated_input = FutureFlightsArrivalDepartureScheduleInput(
          airport_iata_code=airport_iata_code,
          schedule_type=schedule_type,
          airline_iata=airline_iata,
          date=date,
          number_of_flights=number_of_flights,
      )
      return future_flights_arrival_departure_schedule(
          airport_iata_code=validated_input.airport_iata_code,
          schedule_type=validated_input.schedule_type,
          airline_iata=validated_input.airline_iata,
          date=validated_input.date,
          number_of_flights=validated_input.number_of_flights,
      )

• src/aviationstack_mcp/server.py:355-373 (helper)

  fetch_flight_data helper used by the handler to call the Aviationstack API. Builds request with API key, makes GET request, handles errors, and returns parsed JSON payload.

  def fetch_flight_data(endpoint: str, params: dict[str, Any]) -> dict[str, Any]:
      """Fetch data from the Aviationstack API."""
      api_key = _get_api_key()
      request_params = {"access_key": api_key, **params}
      response = requests.get(f"{API_BASE_URL}/{endpoint}", params=request_params, timeout=15)
      try:
          payload = response.json()
      except ValueError:
          payload = {}
  
      if payload.get("error"):
          error = payload["error"]
          code = error.get("code", "api_error")
          error_type = error.get("type", "api_error")
          message = error.get("message", "Unknown API error")
          raise ValueError(f"{error_type} ({code}): {message}")
  
      response.raise_for_status()
      return payload