nova/mcp_servers/shared/server.py

"""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(
    "stellar-ops-shared",
    instructions=(
        "Shared operational data tools for Stellar Air NOVA 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."
    )