Aerospace MCP

Instructions

Implementation Reference

• aerospace_mcp/tools/core.py:79-215 (handler)

  The primary handler function for the 'plan_flight' MCP tool. It processes structured input dictionaries for departure, arrival, and aircraft details, resolves airports using internal helpers, computes the great-circle route, generates performance estimates if OpenAP is available, and returns a formatted JSON response.
  def plan_flight( departure: dict, arrival: dict, aircraft: dict | None = None, route_options: dict | None = None, ) -> str: """Plan a flight route between two airports with performance estimates. Args: departure: Dict with departure info (city, country, iata) arrival: Dict with arrival info (city, country, iata) aircraft: Optional aircraft config (ac_type, cruise_alt_ft, route_step_km) route_options: Optional route options Returns: JSON string with flight plan details """ try: # Build request object request_data = { "depart_city": departure["city"], "arrive_city": arrival["city"], } # Add optional fields if departure.get("country"): request_data["depart_country"] = departure["country"] if arrival.get("country"): request_data["arrive_country"] = arrival["country"] if departure.get("iata"): request_data["prefer_depart_iata"] = departure["iata"] if arrival.get("iata"): request_data["prefer_arrive_iata"] = arrival["iata"] # Add aircraft options (ac_type is required by PlanRequest) if aircraft and aircraft.get("ac_type"): request_data["ac_type"] = aircraft["ac_type"] if aircraft.get("cruise_alt_ft"): request_data["cruise_alt_ft"] = aircraft["cruise_alt_ft"] if aircraft.get("route_step_km"): request_data["route_step_km"] = aircraft["route_step_km"] else: # Use a default aircraft type if none provided request_data["ac_type"] = "A320" # Default aircraft type # Create and validate request try: request = PlanRequest(**request_data) except Exception as e: return f"Invalid request: {str(e)}" # Resolve airports try: depart_airport = _resolve_endpoint( request.depart_city, request.depart_country, request.prefer_depart_iata ) arrive_airport = _resolve_endpoint( request.arrive_city, request.arrive_country, request.prefer_arrive_iata ) except AirportResolutionError as e: return f"Airport resolution error: {str(e)}" # Calculate route route_points = great_circle_points( depart_airport.lat, depart_airport.lon, arrive_airport.lat, arrive_airport.lon, step_km=request.route_step_km, ) # Build response response = { "departure": { "airport": { "iata": depart_airport.iata, "icao": depart_airport.icao, "name": depart_airport.name, "city": depart_airport.city, "country": depart_airport.country, "coordinates": { "lat": depart_airport.lat, "lon": depart_airport.lon, }, } }, "arrival": { "airport": { "iata": arrive_airport.iata, "icao": arrive_airport.icao, "name": arrive_airport.name, "city": arrive_airport.city, "country": arrive_airport.country, "coordinates": { "lat": arrive_airport.lat, "lon": arrive_airport.lon, }, } }, "route": { "distance_km": route_points["distance_km"], "distance_nm": route_points["distance_nm"], "initial_bearing_deg": route_points["initial_bearing_deg"], "final_bearing_deg": route_points["final_bearing_deg"], "points": [ {"lat": p[0], "lon": p[1], "distance_km": p[2]} for p in route_points["points"] ], }, } # Add performance estimates if available if request.ac_type and OPENAP_AVAILABLE: try: performance, engine_name = estimates_openap( request.ac_type, request.cruise_alt_ft, request.mass_kg, route_points["distance_km"], ) response["performance"] = performance response["engine"] = engine_name except OpenAPError as e: response["performance_note"] = ( f"Performance estimation failed: {str(e)}" ) elif request.ac_type: response["performance_note"] = ( f"OpenAP not available - no performance estimates for {request.ac_type}" ) return json.dumps(response, indent=2) except Exception as e: logger.error(f"Flight planning error: {str(e)}", exc_info=True) return f"Flight planning error: {str(e)}"
• aerospace_mcp/fastmcp_server.py:80-88 (registration)

  Registration of the 'plan_flight' tool using FastMCP's mcp.tool(plan_flight) as part of the core flight planning tools.
  mcp = FastMCP("aerospace-mcp") # Register all tools # Core flight planning tools mcp.tool(search_airports) mcp.tool(plan_flight) mcp.tool(calculate_distance) mcp.tool(get_aircraft_performance) mcp.tool(get_system_status)
• aerospace_mcp/core.py:37-64 (schema)

  Pydantic schema (PlanRequest) used internally by the plan_flight handler for input validation and structuring the flight planning request.
  class PlanRequest(BaseModel): # You can pass cities, or override with explicit IATA depart_city: str = Field(..., description="e.g., 'San Jose'") arrive_city: str = Field(..., description="e.g., 'Tokyo'") depart_country: str | None = Field( None, description="ISO alpha-2 country code (optional)" ) arrive_country: str | None = Field( None, description="ISO alpha-2 country code (optional)" ) prefer_depart_iata: str | None = Field( None, description="Force a particular departure airport by IATA" ) prefer_arrive_iata: str | None = Field( None, description="Force a particular arrival airport by IATA" ) # Aircraft/performance knobs ac_type: str = Field(..., description="ICAO aircraft type (e.g., 'A320', 'B738')") cruise_alt_ft: int = Field(35000, ge=8000, le=45000) mass_kg: float | None = Field( None, description="If not set, defaults to 85% MTOW when available" ) route_step_km: float = Field( 25.0, gt=1.0, description="Sampling step for polyline points" ) backend: Literal["openap"] = "openap" # Placeholder for future backends
• aerospace_mcp/core.py:169-182 (helper)

  Helper function to compute great-circle polyline points and total distance between two lat/lon coordinates, used by the plan_flight handler.
  def great_circle_points( lat1: float, lon1: float, lat2: float, lon2: float, step_km: float ) -> tuple[list[tuple[float, float]], float]: g = Geodesic.WGS84.Inverse(lat1, lon1, lat2, lon2) dist_m = g["s12"] line = Geodesic.WGS84.Line(lat1, lon1, g["azi1"]) n = max(1, int(math.ceil((dist_m / 1000.0) / step_km))) pts = [] for i in range(n + 1): s = min(dist_m, (dist_m * i) / n) p = line.Position(s) pts.append((p["lat2"], p["lon2"])) return pts, dist_m / 1000.0
• aerospace_mcp/core.py:193-303 (helper)

  Helper function for generating flight performance estimates (time, fuel, etc.) using the OpenAP library, called within the plan_flight handler.
  def estimates_openap( ac_type: str, cruise_alt_ft: int, mass_kg: float | None, route_dist_km: float ) -> tuple[dict, str]: if not OPENAP_AVAILABLE: raise OpenAPError("OpenAP backend unavailable. Please `pip install openap`.") # Resolve mass default from aircraft properties (fallback to 85% MTOW if present) mass = mass_kg engine_note = "openap" try: ac_props = prop.aircraft(ac_type, use_synonym=True) mtow = (ac_props.get("limits") or {}).get("MTOW") or ac_props.get("mtow") if mass is None and mtow: mass = 0.85 * float(mtow) # conservative default elif mass is None: # fallback generic narrowbody guess mass = 60_000.0 except Exception: mass = mass or 60_000.0 fgen = FlightGenerator(ac=ac_type) dt = 10 # seconds # Generate climb & descent segments at requested cruise altitude (if allowed) try: climb = fgen.climb(dt=dt, alt_cr=cruise_alt_ft) except TypeError: climb = fgen.climb(dt=dt) try: descent = fgen.descent(dt=dt, alt_cr=cruise_alt_ft) except TypeError: descent = fgen.descent(dt=dt) # Cruise segment baseline cruise_seg = fgen.cruise(dt=dt) def seg(df): t_s = float(df["t"].iloc[-1]) dist_km = float(df["s"].iloc[-1]) / 1000.0 # 's' is meters in OpenAP docs alt_ft = float(df["altitude"].mean()) gs_kts = float(df["groundspeed"].mean()) vs_fpm = float(df["vertical_rate"].mean()) return t_s, dist_km, alt_ft, gs_kts, vs_fpm t_climb, d_climb, a_climb, gs_climb, vs_climb = seg(climb) t_des, d_des, a_des, gs_des, vs_des = seg(descent) _, _, a_cru, gs_cru, _ = seg(cruise_seg) # How much cruise distance remains after climb+descent? d_remaining = max(0.0, route_dist_km - (d_climb + d_des)) # Guard for super-short hops cruise_time_s = ( 0.0 if gs_cru <= 1e-6 else (d_remaining * KM_PER_NM) / (gs_cru / 3600.0 / 1.852) ) # but simpler to compute by kts: # Convert properly: kts = nm/hour → km/s = (kts * NM_PER_KM) / 3600 cruise_time_s = ( 0.0 if gs_cru <= 1e-6 else (d_remaining / ((gs_cru * NM_PER_KM) / 3600.0)) ) fuelflow = FuelFlow(ac=ac_type) def fuel_from( avg_gs_kts: float, avg_alt_ft: float, vs_fpm: float, time_s: float ) -> float: # TAS ~ GS (zero-wind assumption for baseline) try: ff_kg_s = float( fuelflow.enroute(mass=mass, tas=avg_gs_kts, alt=avg_alt_ft, vs=vs_fpm) ) except Exception: ff_kg_s = 0.0 return ff_kg_s * time_s fuel_climb = fuel_from(gs_climb, a_climb, vs_climb, t_climb) fuel_des = fuel_from(gs_des, a_des, vs_des, t_des) fuel_cru = fuel_from(gs_cru, cruise_alt_ft, 0.0, cruise_time_s) # Build segments climb_out = SegmentEst( time_min=t_climb / 60.0, distance_km=d_climb, avg_gs_kts=gs_climb, fuel_kg=fuel_climb, ) cruise_out = SegmentEst( time_min=cruise_time_s / 60.0, distance_km=d_remaining, avg_gs_kts=gs_cru, fuel_kg=fuel_cru, ) des_out = SegmentEst( time_min=t_des / 60.0, distance_km=d_des, avg_gs_kts=gs_des, fuel_kg=fuel_des ) block_time_min = climb_out.time_min + cruise_out.time_min + des_out.time_min block_fuel_kg = climb_out.fuel_kg + cruise_out.fuel_kg + des_out.fuel_kg estimates = { "block": {"time_min": block_time_min, "fuel_kg": block_fuel_kg}, "climb": climb_out.model_dump(), "cruise": cruise_out.model_dump(), "descent": des_out.model_dump(), "assumptions": { "zero_wind": True, "mass_kg": mass, "cruise_alt_ft": cruise_alt_ft, }, } return estimates, engine_note