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.

Workflow

                          ┌─────────────────┐
                          │    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

Prerequisites

• 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

Setup

1. Clone & install Python deps

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

2. Configure environment

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

3. Run Conductor locally (Docker)

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.

4. Start the workers

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.

5. Verify the setup

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.

Triggering a workflow

From the Conductor UI

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
}

From the API

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:

Key	Source
source_iata / destination_iata	resolved IATA codes
source_iata_from_cache / destination_iata_from_cache	true if served from iata_cache.db (no flightapi credit used)
flight_deals_outbound	up to 5 cheapest outbound fares ≤ maxFlightPrice
flight_deals_return	up to 5 cheapest return fares ≤ maxFlightPrice
hotel_deals	up to 5 cheapest hotels for the stay ≤ maxHotelPrice

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

Sample input

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

Sample output

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-01T10:30",
        "arrival": "2026-07-01T13:45",
        "duration_minutes": 495
      },
      {
        "price": 684.0,
        "currency": "USD",
        "airline": "Virgin Atlantic",
        "agent": "Kiwi.com",
        "departure": "2026-07-01T11:15",
        "arrival": "2026-07-01T14:25",
        "duration_minutes": 490
      }
    ],
    "flight_deals_return": [
      {
        "price": 598.2,
        "currency": "USD",
        "airline": "American Airlines",
        "agent": "Booking.com",
        "departure": "2026-07-08T18:20",
        "arrival": "2026-07-09T06:55",
        "duration_minutes": 455
      },
      {
        "price": 701.9,
        "currency": "USD",
        "airline": "Delta",
        "agent": "Expedia",
        "departure": "2026-07-08T19:45",
        "arrival": "2026-07-09T08: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/..."
      }
    ]
  }
}

What the worker terminal shows during a run

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.

Caching

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:

Task ref	Cache key	TTL
flights_out	${source_iata}_${dest_iata}_${date}	3600 s
flights_ret	${source_iata}_${dest_iata}_${date}	3600 s
hotels	${location}_${date}_${check_out_date}	3600 s

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.

Extending the workflow

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.

Troubleshooting

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.