Aerospace MCP

# Architecture Documentation This document provides a comprehensive overview of the Aerospace MCP system design, architecture decisions, and technical considerations. ## Table of Contents - [System Overview](#system-overview) - [Component Architecture](#component-architecture) - [Data Flow](#data-flow) - [Technology Stack](#technology-stack) - [Design Decisions](#design-decisions) - [Performance Considerations](#performance-considerations) - [Security Considerations](#security-considerations) - [Scalability & Future Enhancements](#scalability--future-enhancements) ## System Overview The Aerospace MCP (Model Context Protocol) system is a flight planning service that provides: - **Airport Resolution**: Intelligent mapping from city names to airport codes - **Route Calculation**: Great-circle distance computation with geodesic precision - **Performance Estimation**: Aircraft-specific fuel and time calculations using OpenAP - **API Services**: RESTful endpoints and MCP protocol support ### High-Level Architecture ```mermaid graph TB Client[Client Applications] MCP[MCP Protocol Layer] API[FastAPI Application] Core[Core Services] Data[Data Sources] Client --> MCP Client --> API MCP --> API API --> Core Core --> Data subgraph "Core Services" Airport[Airport Resolution] Route[Route Calculation] Perf[Performance Estimation] end subgraph "Data Sources" AirportDB[Airport Database] OpenAP[OpenAP Models] Geodesic[Geographiclib] end Airport --> AirportDB Route --> Geodesic Perf --> OpenAP ``` ## Component Architecture ### 1. Application Layer (main.py) The main application serves as the orchestration layer, handling: - HTTP request/response processing - Input validation via Pydantic models - Error handling and status code management - Dependency coordination ### 2. Core Service Components #### Airport Resolution Service ```mermaid graph LR Input[City Name + Country] --> Search[City Search Logic] Search --> Filter[Country Filter] Filter --> Rank[Airport Ranking] Rank --> Output[Selected Airport] subgraph "Data Access" Search --> AirportDB[airportsdata.load] Filter --> AirportDB end subgraph "Business Logic" Rank --> IntlPref[Prefer International] Rank --> NameMatch[Name Matching] end ``` **Key Functions**: - `_find_city_airports()`: Fuzzy city name matching - `_resolve_endpoint()`: Airport selection with fallback logic - `_airport_from_iata()`: Direct IATA code lookup **Design Patterns**: - **Strategy Pattern**: Supports city search vs. direct IATA lookup - **Filter Pattern**: Progressive filtering by city, country, airport type - **Fallback Pattern**: Graceful degradation when specific airports unavailable #### Route Calculation Service ```mermaid graph LR Origin[Origin Airport] --> GC[Great Circle Calculation] Dest[Destination Airport] --> GC GC --> Sample[Route Sampling] Sample --> Polyline[Polyline Generation] subgraph "Geodesic Engine" GC --> Geodesic[geographiclib.Geodesic] Sample --> Line[Geodesic.Line] Sample --> Position[Line.Position] end ``` **Key Functions**: - `great_circle_points()`: Generates sampled route coordinates - Configurable sampling resolution via `route_step_km` - WGS84 geodesic calculations for accuracy **Mathematical Foundation**: - Uses Vincenty's formulae for geodesic calculations - Accounts for Earth's ellipsoidal shape (WGS84) - Samples route at configurable intervals for visualization #### Performance Estimation Service ```mermaid graph TB Aircraft[Aircraft Type] --> Props[Aircraft Properties] Route[Route Distance] --> Segments[Flight Segments] Mass[Aircraft Mass] --> FuelFlow[Fuel Flow Models] subgraph "OpenAP Integration" Props --> FlightGen[FlightGenerator] FlightGen --> Climb[Climb Profile] FlightGen --> Cruise[Cruise Profile] FlightGen --> Descent[Descent Profile] end subgraph "Calculations" Segments --> Time[Time Estimation] FuelFlow --> Fuel[Fuel Estimation] Time --> Block[Block Time] Fuel --> Block[Block Fuel] end ``` **Key Functions**: - `estimates_openap()`: OpenAP-based performance calculations - Aircraft property resolution with MTOW defaults - Three-phase flight modeling (climb/cruise/descent) ### 3. Data Models & Validation ```mermaid classDiagram class AirportOut { +string iata +string icao +string name +string city +string country +float lat +float lon +string tz } class PlanRequest { +string depart_city +string arrive_city +string depart_country +string arrive_country +string prefer_depart_iata +string prefer_arrive_iata +string ac_type +int cruise_alt_ft +float mass_kg +float route_step_km +string backend } class PlanResponse { +string engine +AirportOut depart +AirportOut arrive +float distance_km +float distance_nm +List polyline +dict estimates } class SegmentEst { +float time_min +float distance_km +float avg_gs_kts +float fuel_kg } PlanResponse --> AirportOut : contains PlanResponse --> SegmentEst : estimates ``` ## Data Flow ### Flight Planning Request Flow ```mermaid sequenceDiagram participant Client participant FastAPI participant AirportResolver participant RouteCalculator participant PerformanceEstimator participant OpenAP Client->>FastAPI: POST /plan FastAPI->>FastAPI: Validate PlanRequest FastAPI->>AirportResolver: Resolve departure airport AirportResolver->>AirportResolver: Search city/IATA AirportResolver-->>FastAPI: Departure airport FastAPI->>AirportResolver: Resolve arrival airport AirportResolver->>AirportResolver: Search city/IATA AirportResolver-->>FastAPI: Arrival airport FastAPI->>RouteCalculator: Calculate great circle route RouteCalculator->>RouteCalculator: Generate polyline RouteCalculator-->>FastAPI: Route & distance FastAPI->>PerformanceEstimator: Estimate performance PerformanceEstimator->>OpenAP: Get aircraft properties OpenAP-->>PerformanceEstimator: Aircraft data PerformanceEstimator->>OpenAP: Generate flight profiles OpenAP-->>PerformanceEstimator: Climb/cruise/descent PerformanceEstimator->>PerformanceEstimator: Calculate fuel flow PerformanceEstimator-->>FastAPI: Performance estimates FastAPI->>FastAPI: Assemble PlanResponse FastAPI-->>Client: Complete flight plan ``` ### Error Handling Flow ```mermaid graph TD Request[Incoming Request] --> Validate[Input Validation] Validate -->|Invalid| ValidationError[422 Validation Error] Validate -->|Valid| Process[Process Request] Process --> Airport[Airport Resolution] Airport -->|Not Found| NotFoundError[404 Airport Not Found] Airport -->|Found| Route[Route Calculation] Route --> Performance[Performance Estimation] Performance -->|OpenAP Missing| NotImplementedError[501 Backend Unavailable] Performance -->|Aircraft Invalid| BadRequestError[400 Invalid Aircraft] Performance -->|Success| Response[200 Success Response] ValidationError --> ErrorResponse[Error Response] NotFoundError --> ErrorResponse NotImplementedError --> ErrorResponse BadRequestError --> ErrorResponse ``` ## Technology Stack ### Core Technologies | Component | Technology | Version | Purpose | |-----------|------------|---------|---------| | **Web Framework** | FastAPI | Latest | REST API, async support, auto-docs | | **Validation** | Pydantic | V2 | Data validation, serialization | | **Airport Data** | airportsdata | Latest | IATA/ICAO airport database | | **Geodesics** | geographiclib | Latest | Great-circle calculations | | **Performance** | OpenAP | Latest | Aircraft performance modeling | | **ASGI Server** | Uvicorn | Latest | Production ASGI server | ### Development & Testing | Component | Technology | Purpose | |-----------|------------|---------| | **Package Manager** | UV / pip | Dependency management | | **Testing** | pytest | Unit/integration testing | | **Type Checking** | mypy | Static type analysis | | **Formatting** | black, isort | Code formatting | | **Documentation** | FastAPI + OpenAPI | Auto-generated API docs | ### Data Sources ```mermaid graph LR subgraph "Airport Data" AirportData[airportsdata Package] IATA[IATA Database] OpenFlights[OpenFlights Data] end subgraph "Aircraft Performance" OpenAP[OpenAP Database] BADA[BADA Models] Manufacturers[OEM Data] end subgraph "Geographic" WGS84[WGS84 Geodesic] GeographicLib[GeographicLib] end AirportData --> IATA AirportData --> OpenFlights OpenAP --> BADA OpenAP --> Manufacturers GeographicLib --> WGS84 ``` ## Design Decisions ### 1. Monolithic vs. Microservices **Decision**: Monolithic architecture in single `main.py` **Rationale**: - **Simplicity**: Single deployment unit, minimal operational complexity - **Performance**: No inter-service communication overhead - **Development Speed**: Rapid prototyping and iteration - **Data Locality**: All operations on shared in-memory data **Trade-offs**: - ✅ Faster development, easier debugging, atomic deployments - ❌ Harder to scale individual components, technology lock-in ### 2. In-Memory vs. Database Storage **Decision**: In-memory airport data loading at startup **Rationale**: - **Performance**: Sub-millisecond airport lookups - **Simplicity**: No database configuration or management - **Data Size**: ~28k airports fit comfortably in memory (~50MB) - **Update Frequency**: Airport data changes infrequently **Implementation**: ```python _AIRPORTS_IATA = airportsdata.load("IATA") # Loaded once at startup ``` **Trade-offs**: - ✅ Extremely fast lookups, no external dependencies - ❌ Higher memory usage, startup time, no real-time updates ### 3. Synchronous vs. Asynchronous Processing **Decision**: Synchronous processing with FastAPI async framework **Rationale**: - **CPU-bound Operations**: Flight calculations are computational, not I/O bound - **Third-party Libraries**: OpenAP and geographiclib are synchronous - **Complexity**: Async adds complexity without clear benefit for current use case **Future Consideration**: Could add async for: - External API calls (weather, NOTAMs) - Concurrent flight planning requests - Database operations ### 4. Backend Abstraction Layer **Decision**: Pluggable backend system with single OpenAP implementation **Architecture**: ```python class PlanRequest(BaseModel): backend: Literal["openap"] = "openap" # Extensible enum def plan(req: PlanRequest): if req.backend == "openap": return estimates_openap(...) # Future: elif req.backend == "eurocontrol": # return estimates_eurocontrol(...) ``` **Rationale**: - **Extensibility**: Easy to add new performance estimation engines - **A/B Testing**: Compare different estimation methods - **Fallback**: Graceful degradation if primary backend unavailable ### 5. Error Handling Strategy **Decision**: HTTPException-based error handling with detailed messages **Pattern**: ```python def _resolve_endpoint(city, country, prefer_iata, role): if prefer_iata: ap = _airport_from_iata(prefer_iata) if not ap: raise HTTPException( status_code=400, detail=f"{role}: IATA '{prefer_iata}' not found." ) # ... more validation ``` **Benefits**: - Clear error messages for debugging - Consistent error format across endpoints - Proper HTTP status codes for different error types ## Performance Considerations ### Memory Usage ```mermaid pie title Memory Usage Distribution "Airport Database" : 45 "OpenAP Models" : 25 "Application Code" : 15 "FastAPI Framework" : 10 "Python Runtime" : 5 ``` **Airport Database**: ~50MB for 28k airports with geographic data **OpenAP Models**: Loaded lazily per aircraft type, ~20-30MB per aircraft **Route Polylines**: Minimal memory (temporary generation) ### CPU Performance **Bottlenecks** (in order of impact): 1. **OpenAP Calculations**: Flight profile generation (200-500ms) 2. **Geodesic Calculations**: Route sampling (10-50ms) 3. **Airport Search**: City name fuzzy matching (1-5ms) 4. **JSON Serialization**: Pydantic model conversion (1-2ms) **Optimization Strategies**: - Cache OpenAP aircraft properties - Optimize route sampling resolution - Pre-compute common city-airport mappings - Use more efficient JSON serialization ### Scalability Metrics Current single-threaded performance estimates: - **Health checks**: 10,000+ req/sec - **Airport searches**: 1,000+ req/sec - **Flight planning**: 5-10 req/sec (OpenAP-limited) **Scaling Strategies**: ```mermaid graph TB Current[Single Process] --> Horizontal[Horizontal Scaling] Current --> Vertical[Vertical Scaling] Current --> Async[Async Processing] Horizontal --> LoadBalancer[Load Balancer] Horizontal --> MultiInstance[Multiple Instances] Vertical --> MoreCPU[More CPU Cores] Vertical --> MoreRAM[More RAM] Async --> Celery[Celery Workers] Async --> AsyncFramework[Async Framework] ``` ## Security Considerations ### Current Security Posture **Authentication**: None (suitable for development/internal use) **Authorization**: None (open access to all endpoints) **Input Validation**: Pydantic models with type checking **Rate Limiting**: None (could lead to resource exhaustion) ### Production Security Enhancements #### 1. Authentication & Authorization ```python # Example JWT implementation from fastapi.security import HTTPBearer from jose import jwt security = HTTPBearer() @app.dependency async def get_current_user(token: str = Depends(security)): payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"]) return payload["sub"] @app.post("/plan", dependencies=[Depends(get_current_user)]) def plan(req: PlanRequest): # Protected endpoint pass ``` #### 2. Input Sanitization ```python class PlanRequest(BaseModel): depart_city: str = Field(..., min_length=2, max_length=100, regex=r'^[a-zA-Z\s]+$') ac_type: str = Field(..., regex=r'^[A-Z0-9]{3,4}$') # ICAO format ``` #### 3. Rate Limiting ```python from slowapi import Limiter limiter = Limiter(key_func=get_remote_address) @app.post("/plan") @limiter.limit("10/minute") def plan(request: Request, req: PlanRequest): pass ``` #### 4. HTTPS/TLS ```bash # Production deployment with TLS uvicorn main:app --host 0.0.0.0 --port 443 \ --ssl-keyfile key.pem --ssl-certfile cert.pem ``` ### Data Privacy **No PII Collection**: System doesn't collect personal information **Audit Logging**: Consider logging for flight planning requests **Data Retention**: No persistent storage of user requests ## Scalability & Future Enhancements ### Horizontal Scaling Architecture ```mermaid graph TB LB[Load Balancer] --> App1[App Instance 1] LB --> App2[App Instance 2] LB --> AppN[App Instance N] App1 --> Cache[Redis Cache] App2 --> Cache AppN --> Cache App1 --> DB[(Airport Database)] App2 --> DB AppN --> DB subgraph "Background Workers" Worker1[Performance Worker 1] Worker2[Performance Worker 2] WorkerN[Performance Worker N] end App1 --> Queue[Task Queue] App2 --> Queue AppN --> Queue Queue --> Worker1 Queue --> Worker2 Queue --> WorkerN ``` ### Database Migration Strategy **Phase 1**: Add optional database support alongside in-memory ```python if DATABASE_URL: airports = load_from_database() else: airports = airportsdata.load("IATA") # Fallback ``` **Phase 2**: Implement caching layer ```python @cache(ttl=3600) # 1-hour cache def find_city_airports(city: str, country: Optional[str]): return db.query_airports(city, country) ``` **Phase 3**: Full database migration with real-time updates ### Enhanced Features Roadmap #### 1. Weather Integration ```python class PlanRequest(BaseModel): include_weather: bool = False weather_source: Literal["noaa", "openweather"] = "noaa" # Weather-adjusted performance calculations def estimates_with_weather(ac_type, route, weather_data): # Adjust for winds, temperature, turbulence pass ``` #### 2. NOTAM Integration ```python @app.get("/route/{flight_id}/notams") def get_route_notams(flight_id: str): # Return NOTAMs affecting the route pass ``` #### 3. Multi-leg Flight Planning ```python class MultiLegRequest(BaseModel): waypoints: List[str] # City names or IATA codes aircraft: str stopovers: List[StopoverConfig] = [] class StopoverConfig(BaseModel): min_ground_time: int = 45 # minutes fuel_stop: bool = False passenger_stop: bool = True ``` #### 4. Real-time Flight Tracking ```python @app.websocket("/ws/track/{flight_id}") async def track_flight(websocket: WebSocket, flight_id: str): # Real-time position updates pass ``` ### Performance Optimization Roadmap #### 1. Caching Strategy - **L1 Cache**: In-memory LRU cache for common requests - **L2 Cache**: Redis for shared cache across instances - **L3 Cache**: CDN for static airport data #### 2. Database Optimization - **Indexing**: Spatial indexes for airport coordinates - **Partitioning**: Partition by region/country - **Read Replicas**: Separate read/write workloads #### 3. Async Processing ```python @app.post("/plan/async") async def plan_async(req: PlanRequest): task_id = await queue_flight_plan.delay(req) return {"task_id": task_id} @app.get("/plan/status/{task_id}") async def get_plan_status(task_id: str): return await get_task_status(task_id) ``` ### Monitoring & Observability ```mermaid graph LR App[Application] --> Metrics[Metrics Collector] App --> Logs[Log Aggregator] App --> Traces[Distributed Tracing] Metrics --> Prometheus[Prometheus] Logs --> ELK[ELK Stack] Traces --> Jaeger[Jaeger] Prometheus --> Grafana[Grafana Dashboard] ELK --> Kibana[Kibana Dashboard] Jaeger --> UI[Jaeger UI] ``` **Key Metrics to Monitor**: - Request latency (p50, p95, p99) - Error rates by endpoint - OpenAP calculation time - Memory usage trends - Cache hit rates **Alerting Thresholds**: - Response time > 5 seconds - Error rate > 5% - Memory usage > 80% - OpenAP unavailable This architecture provides a solid foundation for the current requirements while maintaining flexibility for future enhancements and scaling needs.