67 lines
2.7 KiB
Python
67 lines
2.7 KiB
Python
"""Tools for the passenger MCP server."""
|
|
|
|
import json
|
|
|
|
from mcp_servers.passenger.server import mcp
|
|
|
|
|
|
@mcp.tool()
|
|
async def generate_notification(context: dict) -> str:
|
|
"""Synthesizes flight disruption context into an empathetic,
|
|
actionable passenger notification.
|
|
|
|
Uses Claude via Anthropic SDK (or Bedrock when USE_BEDROCK=true).
|
|
Output: clear, human, no jargon, includes gate/time/status.
|
|
Falls back to template if no API key is configured.
|
|
"""
|
|
try:
|
|
from mcp_servers.shared_llm import generate, _get_provider
|
|
|
|
system_prompt = (
|
|
"You are a passenger notification system for Stellar Air. "
|
|
"Write a clear, empathetic notification about this flight disruption. "
|
|
"Explain WHY the delay or cancellation happened using the operational data provided. "
|
|
"Tell the passenger what's happening next: new boarding time, gate, status. "
|
|
"Be human and reassuring. No aviation jargon. No speculation. "
|
|
"If data is missing for a section, omit it — don't make things up."
|
|
)
|
|
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_notification(context), "provider": "template"})
|
|
|
|
|
|
def _template_notification(context: dict) -> str:
|
|
"""Structured template fallback when LLM is unavailable."""
|
|
flight_id = context.get("flight_id", "")
|
|
origin = context.get("origin", "")
|
|
destination = context.get("destination", "")
|
|
status = context.get("status", "DELAYED")
|
|
delay_minutes = context.get("delay_minutes", 0)
|
|
delay_cause = context.get("delay_cause", "")
|
|
gate = context.get("gate", "")
|
|
weather_summary = context.get("weather_summary", "")
|
|
|
|
lines = [
|
|
f"{flight_id} — {origin} → {destination}",
|
|
f"Status: {status}" + (f" {delay_minutes} minutes" if delay_minutes else ""),
|
|
"",
|
|
]
|
|
|
|
if delay_cause:
|
|
cause_text = {
|
|
"WEATHER": "weather conditions along your route",
|
|
"MAINTENANCE": "a routine maintenance check on your aircraft",
|
|
"CREW": "ensuring your crew is fully rested for safe operation",
|
|
"ATC": "air traffic control restrictions",
|
|
"LATE_AIRCRAFT": "the late arrival of your inbound aircraft",
|
|
}.get(delay_cause, f"{delay_cause.lower()}")
|
|
lines.append(f"Your flight is delayed due to {cause_text}.")
|
|
if weather_summary:
|
|
lines.append(f"Current conditions: {weather_summary}.")
|
|
|
|
if gate:
|
|
lines.append(f"\nGate {gate} — no gate change expected.")
|
|
|
|
return "\n".join(lines)
|