flights_with_airline

Instructions

Implementation Reference

• src/aviationstack_mcp/server.py:58-72 (schema)

  Pydantic input model for flights_with_airline tool with airline_name (str) and number_of_flights (int) fields.

  class FlightsWithAirlineInput(BaseModel):
      """Input schema for flights_with_airline tool."""
  
      model_config = ConfigDict(extra="forbid")
  
      airline_name: str = Field(
          ...,
          description="Airline name to filter flights (for example: Delta Air Lines).",
          min_length=1,
      )
      number_of_flights: int = Field(
          ...,
          description="Number of random flights to return.",
          gt=0,
      )

• src/aviationstack_mcp/server.py:801-821 (registration)

  MCP tool registration for flights_with_airline via @mcp.tool decorator with name and description.

  @mcp.tool(
      name="flights_with_airline",
      description="Return a random sample of live flights filtered by airline name.",
  )
  def flights_with_airline_tool(
      airline_name: Annotated[
          str, Field(description="Airline name to filter flights (for example: Delta Air Lines).")
      ],
      number_of_flights: Annotated[
          int, Field(description="Number of random flights to return.", gt=0)
      ],
  ) -> str:
      """Tool wrapper for flights_with_airline."""
      validated_input = FlightsWithAirlineInput(
          airline_name=airline_name,
          number_of_flights=number_of_flights,
      )
      return flights_with_airline(
          airline_name=validated_input.airline_name,
          number_of_flights=validated_input.number_of_flights,
      )

• src/aviationstack_mcp/server.py:385-417 (handler)

  Core handler function flights_with_airline that fetches live flights filtered by airline name, samples results, and returns a JSON string of filtered flight data.

  def flights_with_airline(airline_name: str, number_of_flights: int) -> str:
      """Get a random sample of real-time flights for an airline."""
      try:
          _validate_positive_int(number_of_flights, "number_of_flights")
          data = fetch_flight_data(
              "flights", {"airline_name": airline_name, "limit": number_of_flights}
          )
          sampled_flights = _sample_data(data.get("data", []), number_of_flights)
  
          filtered_flights = []
          for flight in sampled_flights:
              filtered_flights.append(
                  {
                      "flight_number": _safe_get(flight, "flight", "iata"),
                      "airline": _safe_get(flight, "airline", "name"),
                      "departure_airport": _safe_get(flight, "departure", "airport"),
                      "departure_timezone": _safe_get(flight, "departure", "timezone"),
                      "departure_time": _safe_get(flight, "departure", "scheduled"),
                      "arrival_airport": _safe_get(flight, "arrival", "airport"),
                      "arrival_timezone": _safe_get(flight, "arrival", "timezone"),
                      "flight_status": flight.get("flight_status"),
                      "departure_delay": _safe_get(flight, "departure", "delay"),
                      "departure_terminal": _safe_get(flight, "departure", "terminal"),
                      "departure_gate": _safe_get(flight, "departure", "gate"),
                  }
              )
          if not filtered_flights:
              return f"No flights found for airline '{airline_name}'."
          return json.dumps(filtered_flights)
      except requests.RequestException as exc:
          return _error_response("fetching flights", exc)
      except (KeyError, ValueError, TypeError) as exc:
          return _error_response("fetching flights", exc)

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

  fetch_flight_data helper used by the handler to call the Aviationstack API with params.

  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

• src/aviationstack_mcp/server.py:335-338 (helper)

  _sample_data helper used to randomly sample flights from API results.

  def _sample_data(items: list[dict[str, Any]], count: int) -> list[dict[str, Any]]:
      _validate_positive_int(count, "count")
      number_to_fetch = min(count, len(items))
      return random.sample(items, number_to_fetch)