Flights MCP

Instructions

Input Schema

Input Schema (JSON Schema)

Implementation Reference

• src/flights-mcp/main.py:159-191 (handler)

  The handler function decorated with @mcp.tool, implementing the core logic for retrieving, filtering, paginating, and formatting flight options from a cached search result.
  @mcp.tool( description="Get flight options from the previously performed search. " \ "This tool allows you to filter the found flight options by price, departure and arrival times, and airlines. " \ "It returns a paginated list of flight options that match the specified filters and sorting option." \ "IMPORTANT: This is very cheap operation, so you can call it as many times as needed to find the best flight options." ) def get_flight_options( search_id: str, filters: FiltersModel, page: int = Field(0, description="Page number for pagination. Default is 0."), page_size: int = Field(10, description="Number of results per page. Default is 10.") ): batch = search_results_cache.get(search_id) if not batch: raise ToolError(f"No search results found for search_id: {search_id}. " \ "It may have expired after 10 minutes or not been performed yet. " \ "Please perform a search first using the `search_flights` tool.") filtered_batch = batch.apply_filters(filters) if not filtered_batch.proposals: raise ToolError(f"No flight options found for search_id: {search_id} with the specified filters.") total_results = len(filtered_batch.proposals) start_index = page * page_size end_index = start_index + page_size paginated_results = filtered_batch.proposals[start_index:end_index] result = f'Retrieved {len(paginated_results)} flight options for search_id: {search_id} (Page {page}/{(total_results // page_size) + 1})\n\n' for i, proposal in enumerate(paginated_results): result += proposal.get_short_description() if i < len(paginated_results) - 1: result += "\n---\n" return result
• src/flights-mcp/proposal.py:33-70 (schema)

  Pydantic models defining the input schema for the tool, including SortingMethod enum, TimeRange, SegmentTimeFilter, and the main FiltersModel used in get_flight_options.
  class SortingMethod(str, Enum): CHEAP_FIRST = "cheap_first" EARLY_DEPARTURE_FIRST = "early_departure_first" EARLY_ARRIVAL_FIRST = "early_arrival_first" MINIMAL_DURATION_FIRST = "minimal_duration_first" class TimeRange(BaseModel): """Time range for filtering departure/arrival times""" start_time: Optional[str] = Field(None, description="Start time in HH:MM format (e.g., '08:00')") end_time: Optional[str] = Field(None, description="End time in HH:MM format (e.g., '22:00')") @field_validator('start_time', 'end_time') def validate_time_format(cls, v): if v is None: return v try: datetime.strptime(v, '%H:%M') return v except ValueError: raise ValueError(f"Time must be in HH:MM format, got: {v}") class SegmentTimeFilter(BaseModel): """Time filters for a specific segment""" departure_time_range: Optional[TimeRange] = None arrival_time_range: Optional[TimeRange] = None class FiltersModel(BaseModel): """Model for filtering flight proposals""" max_total_duration: Optional[int] = Field(None, description="Maximum total duration of whole trip in minutes") max_price: Optional[int] = Field(None, description="Maximum price filter") allowed_airlines: Optional[List[str]] = Field(None, description="List of allowed airline IATA codes") segment_time_filters: Optional[List[SegmentTimeFilter]] = Field(None, description="Time filters for each segment") max_stops: Optional[int] = Field(None, description="Maximum number of stops allowed") sorting: Optional[SortingMethod] = Field(SortingMethod.CHEAP_FIRST, description="Sorting method")
• src/flights-mcp/proposal.py:517-553 (helper)

  Key helper method on ProposalsBatchModel that applies the filters, performs sorting, and returns a filtered batch, directly called from the tool handler.
  def apply_filters(self, filters: FiltersModel) -> 'ProposalsBatchModel': """ Apply filters to the proposals and return a new ProposalsBatchModel with filtered results. Args: filters: FiltersModel containing all filter criteria Returns: New ProposalsBatchModel with filtered proposals """ filtered_proposals = [] for proposal in self.proposals: if self._proposal_matches_filters(proposal, filters): filtered_proposals.append(proposal) # Apply sorting if filters.sorting: filtered_proposals = self._sort_proposals(filtered_proposals, filters.sorting) # Create new instance with filtered proposals return ProposalsBatchModel( proposals=filtered_proposals, airports=self.airports, search_id=self.search_id, chunk_id=self.chunk_id, meta=self.meta, airlines=self.airlines, gates_info=self.gates_info, flight_info=self.flight_info, segments=self.segments, market=self.market, clean_marker=self.clean_marker, open_jaw=self.open_jaw, currency=self.currency, initiated_at=self.initiated_at )