Flights MCP

Instructions

Implementation Reference

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

  The main asynchronous handler function implementing the search_flights tool logic. It prepares the request body, generates the API signature, initiates the search via POST, polls the results endpoint every 5 seconds for up to 90 seconds, aggregates and parses flight proposals, caches them, reports progress, and returns a summary description of the results.

  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:60-82 (schema)

  Pydantic models defining the input schema for the search_flights tool, including segment details and overall search parameters like passengers, class, currency, and locale.

  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")
  
  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:88-95 (registration)

  The @mcp.tool decorator that registers the search_flights function as an MCP tool, including its 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/main.py:52-59 (helper)

  Helper function to generate the MD5 signature required for authenticating the API request according to Travelpayouts specification. Used in search_flights.

  def _generate_signature(token: str, body_without_sig: Dict[str, Any]) -> str:
      """Compute MD5 signature per Travelpayouts Flight Search spec."""
  
      ordered_values = _collect_values_sorted(body_without_sig)
      base_string = ":".join([token] + ordered_values)
      print(f"Signature base string: {base_string}", file=sys.stderr)
      return hashlib.md5(base_string.encode()).hexdigest()

• src/flights-mcp/proposal.py:745-759 (helper)

  Helper function to parse the raw API response into a structured ProposalsBatchModel, setting the batch reference on each proposal. Called repeatedly in search_flights to aggregate results.

  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