"""Ops MCP server — tools, resources, and prompts for the Handover agent only. Covers: crew duty status, crew notes, pending rebookings, ops narrative, crew roster, last handover brief, and handover brief prompt template. """ import json from datetime import datetime, timezone from fastmcp import FastMCP from mcp_servers.data.scenarios.manager import scenario_manager mcp = FastMCP( "stellar-ops-internal", instructions=( "Internal operations tools — crew duty status, pending rebookings, " "and ops-audience narrative generation. Restricted to ops-facing clients." ), ) # In-memory store for the last generated handover brief _last_handover: dict | None = None def store_handover_brief(brief: dict) -> None: """Called by the handover agent after generating a brief.""" global _last_handover _last_handover = { **brief, "stored_at": datetime.now(timezone.utc).isoformat(), } # ── Tools ────────────────────────────────────────────────────────── @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. """ # Filter rebookings by flights originating from this hub 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) # ── Resources ────────────────────────────────────────────────────── @mcp.resource("ops://crew/roster") def crew_roster() -> str: """Full crew roster for the current scenario. Returns per crew: crew_id, name, role, base_hub, duty status summary. """ roster = [] for c in scenario_manager.crew: hours_remaining = c.duty_hours_limit - c.duty_hours_elapsed roster.append({ "crew_id": c.crew_id, "name": c.name, "role": c.role.value, "base_hub": c.base_hub, "duty_status": "AT_RISK" if hours_remaining <= 2.0 else "OK", "hours_until_limit": round(hours_remaining, 2), "next_flight": c.next_scheduled_flight, }) return json.dumps(roster) @mcp.resource("ops://handover/latest") def latest_handover() -> str: """The most recently generated shift handover brief. Returns null if no brief has been generated yet. """ if _last_handover is None: return json.dumps(None) return json.dumps(_last_handover) # ── Prompts ──────────────────────────────────────────────────────── @mcp.prompt() def handover_brief(hub: str = "ALL", shift_time: str = "21:00 CT") -> str: """Template for generating a shift handover brief. Defines the structured format: IMMEDIATE / MONITOR / FYI sections, priority scoring expectations, and detail level per item. """ return ( f"Generate a shift handover brief for {hub} at {shift_time}. " "Structure the output as follows:\n\n" "HEADER: Hub, time, outgoing/incoming manager names.\n\n" "IMMEDIATE ACTION: Items requiring action within the next 2 hours. " "Each item: flight/issue ID, what's happening, time until critical, " "specific action needed, and any pre-staged resources.\n\n" "MONITOR: Items that could escalate. Each item: what to watch, " "trigger conditions for escalation, current trajectory.\n\n" "FYI: Resolved items or low-risk situations the incoming shift should know about.\n\n" "Be concise. Ops managers scan, they don't read paragraphs. " "Use the data provided — do not invent details." )