Flights MCP

Instructions

Input Schema

Input Schema (JSON Schema)

Implementation Reference

• src/flights-mcp/main.py:96-158 (handler)

  The core handler function implementing the search_flights tool. It constructs the API request to Travelpayouts, generates signature, polls for results, parses them using proposal parser, caches, and returns a description.
  async def search_flights( request: SearchRequestModel, ctx: Context ) -> Dict[str, Any]: """Search for flights using Travelpayouts Flight Search API.""" request_body = request.model_dump() request_body["token"] = API_TOKEN request_body["marker"] = MARKER request_body["passengers"] = { "adults": request.adults, "children": request.children, "infants": request.infants } del request_body["adults"] del request_body["children"] del request_body["infants"] signature = _generate_signature(API_TOKEN, request_body) request_body["signature"] = signature async with httpx.AsyncClient(timeout=40) as client: init_resp = await client.post(SEARCH_URL, json=request_body) if init_resp.status_code != 200: raise ToolError(f"Aviasales API returned non-200 status code: {init_resp.status_code}, raw text: {init_resp.text}") init_data = init_resp.json() search_id = init_data["search_id"] set_currency_rates(init_data["currency_rates"]) deadline = datetime.now() + timedelta(seconds=90) batch_proposals = None while datetime.now() < deadline: await asyncio.sleep(5) res_r = await client.get(f"{RESULTS_URL}?uuid={search_id}") res_r.raise_for_status() res_json = res_r.json() # Defensive: ensure we got a *list* per the API spec. if not isinstance(res_json, list): raise ToolError("Unexpected response format: expected a list of results") # Aggregate proposals from every object that contains them. for obj in res_json: if isinstance(obj, dict) and obj.get("proposals"): try: if not batch_proposals: batch_proposals = parse_proposals_batch(obj) else: batch_proposals = batch_proposals.combine_with(parse_proposals_batch(obj)) except Exception as e: print(f"Error parsing proposals: \n {json.dumps(obj, indent=2)}", file=sys.stderr) raise ctx.report_progress(progress=len(batch_proposals.proposals), total=None, message=f"Found {len(batch_proposals.proposals)} options so far...") if set(obj.keys()) == {"search_id"}: search_results_cache[search_id] = batch_proposals return batch_proposals.get_description() search_results_cache[search_id] = batch_proposals return batch_proposals.get_description() if batch_proposals else "No proposals found until the search timed out."
• src/flights-mcp/main.py:65-82 (schema)

  Pydantic input schema model for the search_flights tool defining the structure of flight search parameters including segments, passengers, class, currency, etc.
  class SearchRequestModel(BaseModel): """Search request model for Travelpayouts Flight Search API.""" segments: List[SearchRequestSegmentModel] = Field(..., description='''List of CONNECTED flight segments for the same journey. Each segment represents one leg of a multi-city trip or round trip. IMPORTANT: Do NOT use multiple segments for alternative dates of the same route. For flexible dates, perform separate searches. Examples: - One way: [{'origin': 'SFO', 'destination': 'LAX', 'date': '2023-10-01'}] - Round trip: [{'origin': 'SFO', 'destination': 'LAX', 'date': '2023-10-01'}, {'origin': 'LAX', 'destination': 'SFO', 'date': '2023-10-15'}] - Multi-city: [{'origin': 'SFO', 'destination': 'LAX', 'date': '2023-10-01'}, {'origin': 'LAX', 'destination': 'JFK', 'date': '2023-10-05'}] For alternative dates (e.g., 'July 13 OR July 14'), use separate calls of this tool.''') adults: int = Field(1, ge=1, le=9, description="Number of adult passengers (12 years old and older)") children: int = Field(0, ge=0, le=6, description="Number of children (2-11 years old)") infants: int = Field(0, ge=0, le=6, description="Number of infants (under 2 years old)") trip_class: str = Field("Y", description="Trip class - single letter: Y for economy, C for business. Default is Y (economy class)") currency: str = Field("USD", description="Currency code (default is USD)") locale: str = Field("en", description="Locale for the response (default is en). These are the supported locales: en-us, en-gb, ru, de, es, fr, pl")
• src/flights-mcp/main.py:60-64 (schema)

  Pydantic schema for individual flight segments used within SearchRequestModel.
  class SearchRequestSegmentModel(BaseModel): origin: str = Field(..., description='Origin IATA (this can be airport IATA or in case city has multiple airports better to use city IATA). The IATA code is shown in uppercase letters LON or MOW') destination: str = Field(..., description='Destination IATA (this can be airport IATA or in case city has multiple airports better to use city IATA). The IATA code is shown in uppercase letters LON or MOW') date: str = Field(..., description="Departure date in YYYY-MM-DD format")
• src/flights-mcp/main.py:88-95 (registration)

  FastMCP tool registration decorator for the search_flights tool, including detailed description.
  @mcp.tool( description="Search for flights using the Aviasales Flight Search API. " \ "This tool performs search based on the provided flight segments, number of passengers, trip class, currency, and locale. " \ "It provides search_id and description of search results and saves found options internally." \ "After receiving the result client can use `get_flight_options` tool to retrieve the found options with more granular filters." \ "IMPORTANT: All times are local to departure/arrival locations and use HH:MM 24-hour format." \ "IMPORTANT: Call this tool as many times as needed to find the best flight options." )
• src/flights-mcp/proposal.py:745-759 (helper)

  Helper function to parse the raw API response into the ProposalsBatchModel used by search_flights to process and aggregate flight proposals.
  def parse_proposals_batch(api_response: dict) -> ProposalsBatchModel: """ Parse API response dictionary into ProposalsBatchModel. Args: api_response: Dictionary containing the API response Returns: ProposalsBatchModel instance with flattened flight data """ batch = ProposalsBatchModel(**api_response) for i in range(len(batch.proposals)): batch.proposals[i].batch_ref = batch return batch