"""Shared MCP server — tools, resources, and prompts used by both agent clients. Covers: flight status, weather (live OpenMeteo), airport status (live FAA), maintenance flags, hub reference data, and delay explanation prompts. """ from fastmcp import FastMCP from mcp_servers.data.models import HUBS, FlightStatus from mcp_servers.data.real import faa, openmeteo from mcp_servers.data.scenarios.manager import scenario_manager mcp = FastMCP( "united-ops-shared", instructions=( "Shared operational data tools for Stellar Air operations. " "Covers: flight status, weather (live OpenMeteo), airport status " "(live FAA), and maintenance flags. Used by all agent clients." ), ) # ── Tools ────────────────────────────────────────────────────────── @mcp.tool() def get_flight_status(flight_id: str) -> dict: """Lightweight status check for a single flight. Returns: flight_id, status (ON_TIME/DELAYED/CANCELLED/DIVERTED), delay_minutes, gate, last_updated. Use this for quick triage before pulling full operational data. """ for f in scenario_manager.flights: if f.flight_id == flight_id: return { "flight_id": f.flight_id, "status": f.status.value, "delay_minutes": f.delay_minutes, "gate": f.gate, "origin": f.origin, "destination": f.destination, "source": "scenario_mock", } return {"error": f"Flight {flight_id} not found in active scenario"} @mcp.tool() def get_flight_details(flight_id: str) -> dict: """Full operational context for a flight. Returns: scheduled vs actual times, delay cause code, inbound aircraft tail number, gate assignment, crew IDs, passenger count. """ for f in scenario_manager.flights: if f.flight_id == flight_id: return f.model_dump(mode="json") return {"error": f"Flight {flight_id} not found in active scenario"} @mcp.tool() def get_irregular_ops(hub: str) -> list[dict]: """All flights in irregular operations state at a hub. hub: ORD | EWR | IAH | SFO | DEN Returns list of: flight_id, irrop_type, affected_pax_count, delay_cause, delay_minutes. """ results = [] for f in scenario_manager.flights: if f.origin == hub.upper() and f.status != FlightStatus.ON_TIME: results.append({ "flight_id": f.flight_id, "irrop_type": f.status.value, "affected_pax_count": f.passenger_count, "delay_cause": f.delay_cause.value if f.delay_cause else None, "delay_minutes": f.delay_minutes, "origin": f.origin, "destination": f.destination, "gate": f.gate, }) return results @mcp.tool() async def get_route_weather(origin: str, destination: str) -> dict: """Live weather at origin, destination, and en-route waypoints. Source: OpenMeteo API (real-time, no API key). Returns per waypoint: conditions, temperature, wind, visibility, precipitation, significant events. """ return await openmeteo.get_weather_along_route(origin, destination) @mcp.tool() async def get_hub_forecasts() -> dict: """4-hour weather forecast for all United hubs (ORD, EWR, IAH, SFO, DEN). Source: OpenMeteo API. Returns per hub: hourly forecast with conditions, wind, visibility, and a risk_flag if convective activity or low visibility expected. """ return await openmeteo.get_weather_forecast_hubs() @mcp.tool() async def get_airport_status(airport_code: str) -> dict: """Live FAA airport status. Source: FAA NASSTATUS API (real-time, no API key). Returns: delay_type, delay_reason, average_delay_minutes, ground_delay_program active/inactive, ground_stop active/inactive. Falls back to 'status_unavailable' if FAA API is unreachable. """ return await faa.get_airport_status(airport_code) @mcp.tool() async def get_airport_congestion(airport_code: str) -> dict: """Gate availability and taxi times at an airport. Combines: FAA live status (real) + gate/taxi mock data from scenario. Returns: faa_status, gates_available, gates_total, avg_taxi_out_minutes, avg_taxi_in_minutes. """ faa_status = await faa.get_airport_status(airport_code) hub = HUBS.get(airport_code.upper()) total_gates = hub.gates if hub else 100 # Mock congestion based on disrupted flights in scenario disrupted = sum( 1 for f in scenario_manager.flights if f.origin == airport_code.upper() and f.status != FlightStatus.ON_TIME ) gates_occupied = min(total_gates, int(total_gates * 0.7) + disrupted * 3) return { "airport": airport_code.upper(), "faa_status": faa_status, "gates_available": total_gates - gates_occupied, "gates_total": total_gates, "avg_taxi_out_minutes": 12 + disrupted * 4, "avg_taxi_in_minutes": 8 + disrupted * 2, "source": "faa_live+scenario_mock", } @mcp.tool() def get_maintenance_flags(aircraft_tail: str) -> list[dict]: """Active MEL (Minimum Equipment List) items for an aircraft. Returns per item: mel_id, system, description, restriction (route/ops limitations), expiry date. """ items = scenario_manager.maintenance.get(aircraft_tail, []) return [item.model_dump(mode="json") for item in items] # ── Resources ────────────────────────────────────────────────────── @mcp.resource("ops://hubs/{hub_code}") def hub_info(hub_code: str) -> str: """Static reference data for a hub. Returns: full name, codes, coordinates, timezone, terminal/gate/runway count. """ hub = HUBS.get(hub_code.upper()) if not hub: return f"Unknown hub: {hub_code}. Known: {', '.join(HUBS.keys())}" return hub.model_dump_json() @mcp.resource("ops://scenarios/active") def active_scenario() -> str: """Current active scenario metadata.""" import json return json.dumps(scenario_manager.get_metadata()) # ── Prompts ──────────────────────────────────────────────────────── DELAY_TEMPLATES = { ("WEATHER", "passenger"): ( "Explain this flight delay to a passenger. The cause is weather-related. " "Be empathetic and transparent. Include what's happening with the weather, " "what the airline is doing about it, and what the passenger should expect next. " "Do not use aviation jargon." ), ("WEATHER", "ops_manager"): ( "Summarize this weather-caused delay for an ops manager. " "Include: delay cause code, affected systems, GDP/ground stop status if applicable, " "forecast window, and recommended next actions. Be concise." ), ("MAINTENANCE", "passenger"): ( "Explain this maintenance-caused delay to a passenger. " "Be reassuring — emphasize safety. Describe what's being checked " "without technical jargon. Give expected timeline if available." ), ("MAINTENANCE", "ops_manager"): ( "Summarize this maintenance delay for an ops manager. " "Include: MEL item, system affected, restriction details, " "estimated release time, and downstream impact." ), ("CREW", "passenger"): ( "Explain this crew-related delay to a passenger. " "Frame it as 'ensuring your crew is fully rested for safe operation.' " "Do not mention specific regulations or duty limits." ), ("CREW", "ops_manager"): ( "Summarize this crew-caused delay for an ops manager. " "Include: Part 117 status, hours remaining, swap options, " "backup crew availability, and timeline." ), } @mcp.prompt() def delay_explainer(cause_code: str, audience: str) -> str: """Turns a raw delay cause code into a human-readable explanation prompt. cause_code: WEATHER | MAINTENANCE | CREW | ATC | LATE_AIRCRAFT audience: passenger | ops_manager """ key = (cause_code.upper(), audience.lower()) default_key = ("WEATHER", audience.lower()) template = DELAY_TEMPLATES.get(key, DELAY_TEMPLATES.get(default_key, "")) return ( f"{template}\n\n" "Use the operational data provided below. Do not invent details. " "If data is missing for a section, omit that section." )