134 lines
4.7 KiB
Python
134 lines
4.7 KiB
Python
"""Tools for the ops MCP server."""
|
||
|
||
import json
|
||
from datetime import datetime, timezone
|
||
|
||
from mcp_servers.data.scenarios.manager import scenario_manager
|
||
from mcp_servers.ops.server import mcp
|
||
|
||
|
||
@mcp.tool()
|
||
def get_crew_notes(flight_id: str) -> list[str]:
|
||
"""Free-text notes logged by crew or ops team for a flight.
|
||
|
||
Includes: maintenance write-ups, catering delays, gate conflicts,
|
||
passenger incidents, operational remarks.
|
||
"""
|
||
return scenario_manager.crew_notes.get(flight_id, [])
|
||
|
||
|
||
@mcp.tool()
|
||
def get_crew_duty_status(crew_ids: list[str]) -> list[dict]:
|
||
"""Duty hours and rest state for each crew member.
|
||
|
||
Returns per crew: crew_id, name, role, duty_hours_elapsed,
|
||
duty_hours_limit, hours_until_limit, rest_hours_since_last,
|
||
at_risk (true if within 2h of FAA Part 117 limit).
|
||
"""
|
||
results = []
|
||
crew_map = {c.crew_id: c for c in scenario_manager.crew}
|
||
|
||
for crew_id in crew_ids:
|
||
crew = crew_map.get(crew_id)
|
||
if not crew:
|
||
results.append({"crew_id": crew_id, "error": "not_found"})
|
||
continue
|
||
|
||
hours_until_limit = crew.duty_hours_limit - crew.duty_hours_elapsed
|
||
results.append({
|
||
"crew_id": crew.crew_id,
|
||
"name": crew.name,
|
||
"role": crew.role.value,
|
||
"duty_hours_elapsed": crew.duty_hours_elapsed,
|
||
"duty_hours_limit": crew.duty_hours_limit,
|
||
"hours_until_limit": round(hours_until_limit, 2),
|
||
"rest_hours_since_last": crew.rest_hours_since_last,
|
||
"at_risk": hours_until_limit <= 2.0,
|
||
"base_hub": crew.base_hub,
|
||
"next_scheduled_flight": crew.next_scheduled_flight,
|
||
})
|
||
|
||
return results
|
||
|
||
|
||
@mcp.tool()
|
||
def get_pending_rebookings(hub: str, limit: int = 20) -> list[dict]:
|
||
"""Passengers needing rebooking at a hub, sorted by urgency.
|
||
|
||
Returns: pax_id, name, mileage_plus_status, original_flight,
|
||
destination, next_available_option, urgency.
|
||
"""
|
||
hub_flights = {f.flight_id for f in scenario_manager.flights if f.origin == hub.upper()}
|
||
rebookings = [
|
||
r for r in scenario_manager.rebookings
|
||
if r.original_flight in hub_flights
|
||
]
|
||
|
||
urgency_order = {"HIGH": 0, "MEDIUM": 1, "LOW": 2}
|
||
rebookings.sort(key=lambda r: urgency_order.get(r.urgency, 3))
|
||
|
||
return [r.model_dump(mode="json") for r in rebookings[:limit]]
|
||
|
||
|
||
@mcp.tool()
|
||
async def generate_narrative(context: dict) -> str:
|
||
"""Synthesizes aggregated operational context into a structured
|
||
handover brief for ops managers.
|
||
|
||
Uses Claude via Anthropic SDK (or Bedrock when USE_BEDROCK=true).
|
||
Output: prioritized, concise, structured by IMMEDIATE / MONITOR / FYI.
|
||
Falls back to template if no API key is configured.
|
||
"""
|
||
try:
|
||
from mcp_servers.shared_llm import generate, _get_provider
|
||
|
||
hub = context.get("hub", "ALL")
|
||
shift_time = context.get("shift_time", datetime.now(timezone.utc).strftime("%H:%M UTC"))
|
||
|
||
system_prompt = (
|
||
f"You are an airline operations shift handover briefing system. "
|
||
f"Generate a concise handover brief for {hub} at {shift_time}. "
|
||
f"Structure as: HEADER, then IMMEDIATE ACTION (items needing action within 2h), "
|
||
f"MONITOR (items that could escalate), FYI (resolved or low-risk). "
|
||
f"Be concise — ops managers scan, they don't read paragraphs. "
|
||
f"Use the data provided. Do not invent details."
|
||
)
|
||
text = await generate(system_prompt, json.dumps(context, indent=2))
|
||
return json.dumps({"text": text, "provider": _get_provider()})
|
||
except Exception:
|
||
return json.dumps({"text": _template_narrative(context), "provider": "template"})
|
||
|
||
|
||
def _template_narrative(context: dict) -> str:
|
||
"""Structured template fallback when LLM is unavailable."""
|
||
sections = []
|
||
immediate = context.get("immediate", [])
|
||
monitor = context.get("monitor", [])
|
||
fyi = context.get("fyi", [])
|
||
|
||
hub = context.get("hub", "ALL")
|
||
shift_time = context.get("shift_time", datetime.now(timezone.utc).strftime("%H:%M UTC"))
|
||
|
||
sections.append(f"SHIFT HANDOVER BRIEF — {hub} / {shift_time}")
|
||
sections.append(f"Generated: {datetime.now(timezone.utc).strftime('%H:%M UTC')}")
|
||
sections.append("")
|
||
|
||
if immediate:
|
||
sections.append("━━━ IMMEDIATE ACTION ━━━")
|
||
for item in immediate:
|
||
sections.append(f"▶ {item}")
|
||
sections.append("")
|
||
|
||
if monitor:
|
||
sections.append("━━━ MONITOR ━━━")
|
||
for item in monitor:
|
||
sections.append(f"⚠ {item}")
|
||
sections.append("")
|
||
|
||
if fyi:
|
||
sections.append("━━━ FYI ━━━")
|
||
for item in fyi:
|
||
sections.append(f"ℹ {item}")
|
||
|
||
return "\n".join(sections)
|