140 lines
4.8 KiB
Python
140 lines
4.8 KiB
Python
"""Tools for the shared MCP server."""
|
|
|
|
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
|
|
from mcp_servers.shared.server import mcp
|
|
|
|
|
|
@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
|
|
|
|
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]
|