TravelTracker

A small example app that uses Conductor to orchestrate a parallel search for round-trip flights and hotels under a budget. Given a source city, a destination city, outbound/return dates, and per-leg price caps, the workflow:

1. Resolves source/destination city names to IATA airport codes — in parallel. Each side first consults a local SQLite cache (iata_cache.db); on a cache hit the API call is skipped (no flightapi credit burned), on a miss the answer is fetched from flightapi.io and written back to the cache.

2. In parallel, searches:

   • outbound flight fares (flightapi.io one-way trip, src → dest on travelDate),
   • return flight fares (flightapi.io one-way trip, dest → src on returnDate), and
   • hotels (Google Hotels via SerpApi, checking in on travelDate and checking out on returnDate).

   Each of these three "deals" tasks has a Conductor cacheConfig attached so the workflow re-uses prior results for the same route + date within a 1-hour TTL, skipping the external API call (and credit) on a hit.

3. Filters each list to options that fall under the supplied budget and returns the cheapest matches.

 ┌─────────────────┐
 │ iata_fork │
 └────┬───────┬────┘
 │ │
 ┌────────────┘ └─────────────┐
 ▼ ▼
 ┌────────────────────┐ ┌────────────────────┐
 source │ iata_cache_lookup │ destination │ iata_cache_lookup │
 ───► │ (db check) │ ───► │ (db check) │
 └─────────┬──────────┘ └─────────┬──────────┘
 │ {iata, found} │ {iata, found}
 ▼ ▼
 ┌────────────────────┐ ┌────────────────────┐
 │ iata_resolve │ │ iata_resolve │
 │ if cached: skip API│ │ if cached: skip API│
 │ else: flightapi + │ │ else: flightapi + │
 │ write cache │ │ write cache │
 └─────────┬──────────┘ └─────────┬──────────┘
 │ iata │ iata
 └─────────────┬─────────────────────┘
 ▼
 ┌──────────────┐
 │ iata_join │
 └──────┬───────┘
 ▼
 ┌──────────────────┐
 │ search_fork │
 └──┬────────┬────┬─┘
 │ │ │
 ┌─────────────┘ │ └──────────────┐
 ▼ ▼ ▼
 ┌────────────────┐ ┌────────────────┐ ┌────────────────────┐
 │ flight_search_ │ │ flight_search_ │ │ hotel_search_ │
 │ provider │ │ provider │ │ google │
 │ (src→dest, │ │ (dest→src, │ │ (check_in=travel, │
 │ travelDate) │ │ returnDate) │ │ check_out=return) │
 │ filter ≤ maxFP │ │ filter ≤ maxFP │ │ filter ≤ maxHP │
 └────────┬───────┘ └────────┬───────┘ └─────────┬──────────┘
 │ │ │
 └─────────────┬──────┴─────────────────────┘
 ▼
 ┌──────────────────┐
 │ join_search │
 └────────┬─────────┘
 ▼
 flight_deals_outbound + flight_deals_return + hotel_deals

• Python 3.9+
• A reachable Conductor server. Configured via CONDUCTOR_SERVER_URL in .env. Easiest option is to run it locally with Docker (see Setup step 3); a hosted instance like Orkes Cloud works just as well.
• Docker (only if you're running Conductor locally rather than against a hosted server).
• API keys (both free, no credit card required):
  • flightapi.io — 20 free calls
  • SerpApi — 100 free searches/month

git clone https://github.com/<your-user>/TravelTracker-Conductor.git
cd TravelTracker-Conductor
python3 -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt

cp .env.example .env
# then edit .env and paste in your real API keys

The simplest option is the official standalone image:

docker run --init -p 8080:8080 -p 5000:5000 \
 --name conductor \
 conductoross/conductor-standalone:latest

• Server API: http://localhost:8080/api
• UI: http://localhost:5000

If you point at an Orkes Cloud cluster instead, set CONDUCTOR_SERVER_URL=https://<your-cluster>.orkesconductor.io/api in your .env.

In a separate terminal (with the venv activated):

python main.py

On startup the app will:

• register/overwrite the api_driven_travel_search workflow definition,
• create / open iata_cache.db and pre-warm it with the bundled US-city seed (see Caching), and
• start polling for each of the four task types: iata_cache_lookup, iata_resolve, flight_search_provider, hotel_search_google.

Keep this process running while you trigger workflow executions.

Without burning a single API credit you can confirm everything came up correctly:

# 1. Workflow is registered on the server (version should match workflow_defs.py):
curl -s http://localhost:8080/api/metadata/workflow/api_driven_travel_search \
 | python3 -c "import json,sys; w=json.load(sys.stdin); print(w['name'], 'v', w['version'])"
# 2. Seed loaded into the IATA cache (expect ~58 rows on a fresh setup):
sqlite3 iata_cache.db "SELECT COUNT(*) FROM iata_cache;"
# 3. Worker process is up — look at the terminal running main.py; you should
# have seen four 'Conductor Worker[name=...]' lines on startup, one per
# task type. If those scrolled away, `pgrep -fl "python.*main.py"` should
# show one parent + four child processes.

Then look at the Conductor UI at http://localhost:5000 — the workflow should appear under Workflow Definitions with the right version number.

1. Open http://localhost:5000.
2. Go to Workflow Definitions → api_driven_travel_search → Run Workflow.
3. Provide input, for example:

{
 "source": "London",
 "destination": "New York",
 "travelDate": "2026年07月01日",
 "returnDate": "2026年07月08日",
 "maxFlightPrice": 1500,
 "maxHotelPrice": 400
}

curl -X POST "http://localhost:8080/api/workflow/api_driven_travel_search" \
 -H "Content-Type: application/json" \
 -d '{
 "source": "London",
 "destination": "New York",
 "travelDate": "2026-07-01",
 "returnDate": "2026-07-08",
 "maxFlightPrice": 1500,
 "maxHotelPrice": 400
 }'

The response is the new workflow ID. Inspect progress and output in the UI or poll GET /api/workflow/{workflowId}. The output object has:

maxFlightPrice is applied independently to both legs. Either price cap can be omitted to disable filtering for that leg.

{
 "source": "London",
 "destination": "New York",
 "travelDate": "2026年07月01日",
 "returnDate": "2026年07月08日",
 "maxFlightPrice": 1500,
 "maxHotelPrice": 400
}

A successful execution returns status: "COMPLETED" with an output object shaped like this (a real run picked LHR from cache and resolved JFK via the API the first time):

{
 "status": "COMPLETED",
 "output": {
 "source_iata": "LHR",
 "destination_iata": "JFK",
 "source_iata_from_cache": true,
 "destination_iata_from_cache": false,
 "flight_max_price": 1500,
 "hotel_max_price": 400,
 "flight_deals_outbound": [
 {
 "price": 612.5,
 "currency": "USD",
 "airline": "British Airways",
 "agent": "Expedia",
 "departure": "2026年07月01日T10:30",
 "arrival": "2026年07月01日T13:45",
 "duration_minutes": 495
 },
 {
 "price": 684.0,
 "currency": "USD",
 "airline": "Virgin Atlantic",
 "agent": "Kiwi.com",
 "departure": "2026年07月01日T11:15",
 "arrival": "2026年07月01日T14:25",
 "duration_minutes": 490
 }
 ],
 "flight_deals_return": [
 {
 "price": 598.2,
 "currency": "USD",
 "airline": "American Airlines",
 "agent": "Booking.com",
 "departure": "2026年07月08日T18:20",
 "arrival": "2026年07月09日T06:55",
 "duration_minutes": 455
 },
 {
 "price": 701.9,
 "currency": "USD",
 "airline": "Delta",
 "agent": "Expedia",
 "departure": "2026年07月08日T19:45",
 "arrival": "2026年07月09日T08:10",
 "duration_minutes": 445
 }
 ],
 "hotel_deals": [
 {
 "name": "The Pod 51 Hotel",
 "price_per_night": 189,
 "currency": "USD",
 "rating": 4.1,
 "link": "https://www.google.com/travel/hotels/entity/..."
 },
 {
 "name": "Hampton Inn Manhattan Times Square North",
 "price_per_night": 264,
 "currency": "USD",
 "rating": 4.3,
 "link": "https://www.google.com/travel/hotels/entity/..."
 },
 {
 "name": "The Pierre, A Taj Hotel",
 "price_per_night": 389,
 "currency": "USD",
 "rating": 4.6,
 "link": "https://www.google.com/travel/hotels/entity/..."
 }
 ]
 }
}

For each execution main.py's stdout adds diagnostic lines — useful for seeing how the two cache layers behave:

[iata] cache HIT 'London' -> LHR
[iata] resolve SKIP (cached) 'London' -> LHR
[iata] cache MISS 'New York' -> None
[iata] resolve API 'New York' -> JFK (stored)
[flights] LHR->JFK 2026年07月01日 keys=['itineraries', 'legs', 'segments', 'places', 'carriers', 'agents']
[flights] parsed 47 fares before filtering
[flights] JFK->LHR 2026年07月08日 keys=['itineraries', 'legs', 'segments', 'places', 'carriers', 'agents']
[flights] parsed 51 fares before filtering
[hotels] New York 2026年07月01日->2026年07月08日 keys=['search_metadata', 'search_parameters', 'properties'] properties=18 error=None
[hotels] parsed 18 priced hotels before filtering

A re-trigger of the same workflow within the 1-hour cacheConfig TTL won't print anything from [flights] or [hotels] — those workers aren't called at all because Conductor served the cached output directly.

This project uses two layers of caching, each tuned to what makes sense for that data:

1. IATA codes — local SQLite (iata_cache.db), permanent. iata_cache_lookup and iata_resolve together form a two-step pattern runnable in parallel for source and destination:

2. Flight / hotel deals — Conductor cacheConfig, 1-hour TTL. Each of the three "deals" tasks in the search fork has a cacheConfig on the workflow definition:

Within the TTL window, a second workflow execution with the same route and dates returns the cached task output directly without calling the worker — zero flightapi or SerpApi credits used. Note max_price is deliberately not part of the key, so if you trigger the same route+date with a different maxFlightPrice / maxHotelPrice inside the hour you'll receive the previously filtered list, not a freshly filtered one. Wait out the TTL (or include max_price in the cacheConfig key template) if you need budget-sensitive freshness.

Ideas that fit naturally into what's already here. Each one is a small addition rather than a rewrite — that's the point of having Conductor in the middle.

More providers (new tasks in the existing fork):

• Flights: Wire in Amadeus, Duffel, or Kiwi as additional @worker_task workers and add their references as new branches in search_fork. The join already collects whatever branches you define.
• Hotels: Add a Booking.com / Agoda worker alongside the SerpApi one and the cheapest-across-providers logic becomes a downstream task.

New steps after the search (new tasks added to the workflow):

• Trip total. A pure-compute SIMPLE task that picks the single cheapest outbound + cheapest return + cheapest hotel and reports the total trip cost in the workflow output.
• Sentiment-rank hotels. An LLM_TASK (or HTTP_TASK against a model API) that reads the hotel review snippets SerpApi returns and ranks results by sentiment, not just price.
• Slack / email alert. An HTTP_TASK at the end that posts the deals to a channel if the cheapest total falls under a notifyBelow input.
• Human approval before booking. A HUMAN task that waits for someone to pick a flight + hotel combination in the UI; downstream a SIMPLE task records the choice.

Most of the additions above touch only workflow_defs.py and workers.py — the orchestration shape stays the same and the UI keeps showing the same kind of timeline.

If the workers don't seem to run at all on a re-trigger, that's the Conductor cacheConfig 1-hour TTL doing its job — the deals were served from the workflow-level cache. Trigger the workflow with a different travelDate or returnDate to bypass.

FLIGHTAPI_KEY is not set in the environment when running workflows. The workers load_dotenv() once at import time. If you edit .env after starting main.py, restart the process so the new value is picked up.