flight_arrival_departure_schedule

Instructions

Implementation Reference

• src/aviationstack_mcp/server.py:865-871 (registration)

  Decorator registration of the tool 'flight_arrival_departure_schedule' via @mcp.tool(), mapping it to the wrapper function flight_arrival_departure_schedule_tool.

  @mcp.tool(
      name="flight_arrival_departure_schedule",
      description=(
          "Return current-day arrival or departure schedule samples for an airport, "
          "optionally filtered by airline name."
      ),
  )

• src/aviationstack_mcp/server.py:465-514 (handler)

  Core handler function that fetches current-day arrival/departure schedules from the AviationStack API, validates inputs, fetches data, samples flights, and returns JSON.

  def flight_arrival_departure_schedule(
      airport_iata_code: str,
      schedule_type: str,
      airline_name: str,
      number_of_flights: int,
  ) -> str:
      """Get a random sample of current-day arrival/departure schedules for an airport."""
      try:
          _validate_positive_int(number_of_flights, "number_of_flights")
          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}
          if airline_name:
              params["airline_name"] = airline_name
  
          data = fetch_flight_data("timetable", 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_estimated_time": _safe_get(
                          flight, "departure", "estimatedTime"
                      ),
                      "departure_scheduled_time": _safe_get(
                          flight, "departure", "scheduledTime"
                      ),
                      "departure_actual_time": _safe_get(flight, "departure", "actualTime"),
                      "departure_terminal": _safe_get(flight, "departure", "terminal"),
                      "departure_gate": _safe_get(flight, "departure", "gate"),
                      "arrival_estimated_time": _safe_get(flight, "arrival", "estimatedTime"),
                      "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"),
                      "departure_delay": _safe_get(flight, "departure", "delay"),
                  }
              )
          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 schedule", exc)
      except (KeyError, ValueError, TypeError) as exc:
          return _error_response("fetching flight schedule", exc)

• src/aviationstack_mcp/server.py:104-128 (schema)

  Pydantic input schema (FlightArrivalDepartureScheduleInput) defining fields: airport_iata_code, schedule_type, airline_name, number_of_flights.

  class FlightArrivalDepartureScheduleInput(BaseModel):
      """Input schema for flight_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_name: str = Field(
          default="",
          description="Optional airline name filter.",
      )
      number_of_flights: int = Field(
          ...,
          description="Number of random flights to return.",
          gt=0,
      )

• src/aviationstack_mcp/server.py:872-894 (helper)

  Tool wrapper function that validates inputs via the Pydantic schema, then delegates to the core handler function flight_arrival_departure_schedule.

  def flight_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_name: Annotated[str, Field(description="Optional airline name filter.")] = "",
      number_of_flights: Annotated[
          int, Field(description="Number of random flights to return.", gt=0)
      ] = 5,
  ) -> str:
      """Tool wrapper for flight_arrival_departure_schedule."""
      validated_input = FlightArrivalDepartureScheduleInput(
          airport_iata_code=airport_iata_code,
          schedule_type=schedule_type,
          airline_name=airline_name,
          number_of_flights=number_of_flights,
      )
      return flight_arrival_departure_schedule(
          airport_iata_code=validated_input.airport_iata_code,
          schedule_type=validated_input.schedule_type,
          airline_name=validated_input.airline_name,
          number_of_flights=validated_input.number_of_flights,
      )