205 lines
7.2 KiB
Python
205 lines
7.2 KiB
Python
"""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()
|
||
def generate_narrative(context: dict) -> str:
|
||
"""Synthesizes aggregated operational context into a structured
|
||
handover brief for ops managers.
|
||
|
||
Uses Claude Sonnet via AWS Bedrock Converse API.
|
||
Output: prioritized, concise, structured by IMMEDIATE / MONITOR / FYI.
|
||
|
||
NOTE: In v1, this returns a structured template from the context data.
|
||
LLM integration will be added when Bedrock is wired up.
|
||
"""
|
||
# V1: structured template — will be replaced with Bedrock call
|
||
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"))
|
||
|
||
header = f"SHIFT HANDOVER BRIEF — {hub} / {shift_time}"
|
||
sections.append(header)
|
||
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."
|
||
)
|