Flight Ticket MCP Server

Instructions

Implementation Reference

• flight_ticket_mcp_server/main.py:279-284 (handler)

  MCP tool handler for 'trackMultipleFlights', decorated with @mcp.tool() for automatic registration. Logs the invocation and delegates execution to the core helper function in simple_opensky_tools.

  @mcp.tool()
  def trackMultipleFlights(flight_numbers: list, date: str = None):
      """批量航班跟踪 - 同时查询多个航班的实时状态。flight_numbers为航班呼号列表，如['CCA1234','CSN5678']"""
      logger.debug(f"调用批量航班跟踪工具: flight_numbers={flight_numbers}, date={date}")
      return simple_opensky_tools.trackMultipleFlights(flight_numbers, date)

• flight_ticket_mcp_server/tools/simple_opensky_tools.py:354-387 (helper)

  Core helper function implementing the tracking logic: loops through each flight number, retrieves status via getFlightStatus, applies rate limiting, aggregates results into a structured dictionary with success metrics.

  def trackMultipleFlights(flight_numbers: List[str], date: str = None) -> Dict[str, Any]:
      """
      批量跟踪多个航班状态
      
      Args:
          flight_numbers: 航班号列表
          date: 日期参数（OpenSky仅支持实时数据）
          
      Returns:
          包含所有航班状态的字典
      """
      if date:
          logger.info("OpenSky仅支持实时数据，忽略date参数")
      
      logger.info(f"批量查询航班状态: {flight_numbers}")
      
      results = []
      for flight_number in flight_numbers:
          result = getFlightStatus(flight_number)
          results.append(result)
          time.sleep(1)  # 避免API频率限制
      
      successful_count = sum(1 for r in results if r.get("status") == "success" and r.get("flight_count", 0) > 0)
      
      return {
          "status": "success",
          "message": f"批量查询完成，共查询{len(flight_numbers)}个航班，找到{successful_count}个",
          "flight_count": len(flight_numbers),
          "found_count": successful_count,
          "results": results,
          "query_time": datetime.now().isoformat(),
          "data_source": "opensky_network_rest",
          "note": "OpenSky仅提供实时数据，无法查询历史航班信息"
      }