Multi-Agent Travel Planner

# travel_planner.py — Vol. 23 multi-agent capstone (educational) # Planner / researcher / booker. Mock tools. Human approve. No payments. from enum import Enum from fastapi import FastAPI, HTTPException from pydantic import BaseModel, Field app = FastAPI(title="Vol23 Travel Planner") FLIGHTS = [ {"id": "F1", "from": "SFO", "to": "JFK", "date": "2026-09-10", "usd_band": "mid"}, {"id": "F2", "from": "SFO", "to": "JFK", "date": "2026-09-10", "usd_band": "low"}, ] HOTELS = [ {"id": "H1", "city": "NYC", "nights": 3, "usd_band": "mid"}, {"id": "H2", "city": "NYC", "nights": 3, "usd_band": "low"}, ] class Brief(BaseModel): trip_id: str origin: str dest: str start_date: str nights: int = Field(ge=1, le=14) budget_band: str = Field(pattern="^(low|mid|high)$") notes: str = "" class TripState(BaseModel): brief: Brief plan: list[str] = [] research: dict = {} itinerary: dict | None = None approved: bool = False booking_ids: list[str] = [] trace: list[str] = [] TRIPS: dict[str, TripState] = {} def mock_search_flights(origin: str, dest: str, date: str, band: str) -> list[dict]: return [f for f in FLIGHTS if f["from"] == origin and f["to"] == dest and f["usd_band"] == band] def mock_search_hotels(city: str, nights: int, band: str) -> list[dict]: return [h for h in HOTELS if h["city"] == city and h["nights"] == nights and h["usd_band"] == band] def mock_hold(itinerary: dict) -> str: # Never call a payment API. Confirmation is a fake id. return "HOLD-" + itinerary.get("flight_id", "x") + "-" + itinerary.get("hotel_id", "x") class Role(str, Enum): planner = "planner" researcher = "researcher" booker = "booker" def planner_node(state: TripState) -> TripState: b = state.brief state.plan = [ f"Search flights {b.origin}→{b.dest} on {b.start_date} band={b.budget_band}", f"Search hotel near {b.dest} nights={b.nights} band={b.budget_band}", "Draft itinerary; wait for human approve before booker", ] state.trace.append("planner: emitted research checklist") return state def researcher_node(state: TripState) -> TripState: b = state.brief city = "NYC" if b.dest.upper() in {"JFK", "NYC", "EWR"} else b.dest flights = mock_search_flights(b.origin, b.dest, b.start_date, b.budget_band) hotels = mock_search_hotels(city, b.nights, b.budget_band) state.research = {"flights": flights, "hotels": hotels} if flights and hotels: state.itinerary = {"flight_id": flights[0]["id"], "hotel_id": hotels[0]["id"], "band": b.budget_band} state.trace.append("researcher: mock search only") return state @app.post("/v1/trips/plan") def plan(brief: Brief): state = TripState(brief=brief) state = planner_node(state) state = researcher_node(state) TRIPS[brief.trip_id] = state return state.model_dump() @app.post("/v1/trips/{trip_id}/approve") def approve(trip_id: str, ok: bool = True): state = TRIPS.get(trip_id) if not state: raise HTTPException(404, "unknown trip") state.approved = bool(ok and state.itinerary) state.trace.append(f"hitl: approved={state.approved}") return {"trip_id": trip_id, "approved": state.approved} @app.post("/v1/trips/{trip_id}/book") def book(trip_id: str): state = TRIPS.get(trip_id) if not state: raise HTTPException(404, "unknown trip") if not state.approved: raise HTTPException(403, "Human approval required before booker runs.") # Booker role: mock hold only. No payment fields accepted. hold_id = mock_hold(state.itinerary or {}) state.booking_ids.append(hold_id) state.trace.append(f"booker: mock hold {hold_id}") return {"trip_id": trip_id, "booking_ids": state.booking_ids, "payment": "none"}