🏖️ Capstone Travel Planner#

Capstone Travel Assistant Agent brings together every concept from the AWS Strands module into a single cohesive workflow: custom domain tools, default parameter values, prebuilt strands_tools, cross-tool data dependencies, and system-prompt workflow guidance.

Given a natural-language trip brief like "I'm going to Goa for 3 days with 2 friends. My budget is 20000 rupees. What should I pack, and does it fit my budget?", the agent coordinates four tools in sequence:

1. get_weather_forecast(city="Goa", days=3) — retrieves daytime high (32°C), night low (26°C), and conditions ("humid, occasional showers").
2. suggest_packing_list(high_c=32, low_c=26, days=3, conditions=...) — feeds the forecast outputs into rule-based packing logic (adding cotton clothing, sunscreen, umbrella, and quick-dry footwear).
3. estimate_trip_cost(city="Goa", days=3, travellers=3) — computes stay, food, and local transport in INR (inferring travellers=3 from "with 2 friends").
4. calculator — checks the itemized total against the ₹20,000 budget so the final itinerary gives an exact verdict.

1. Soft Workflow Steering via System Prompt

Rather than hardcoding a rigid Python pipeline that always calls all three functions in fixed code, the system prompt provides a high-level policy: "When asked about a trip: check the weather first, then suggest what to pack based on that weather, then estimate the cost. Always say whether the trip fits the user's budget." This keeps the agent flexible enough to handle partial questions (e.g., "Just check the weather in Manali for 4 days") while following a structured plan for full trip requests.

2. Semantic Reasoning Over Pragmatic Phrasing ("with 2 friends" → 3 travellers)

Notice the subtle natural-language reasoning required: the user says "I'm going to Goa... with 2 friends". A naive keyword extractor would see the digit 2 and pass travellers=2. The LLM understands that the speaker plus two friends equals 3 people and passes travellers=3 to estimate_trip_cost().

3. Combining Rule Engines with LLMs

suggest_packing_list() uses deterministic Python thresholds (high_c >= 30, low_c <= 15, "rain" in conditions) to build a guaranteed baseline checklist, and the LLM weaves that structured list into a tailored, readable travel brief.

4. Full Workflow Traceability (workflow_steps)

Every toolUse block in agent.messages is captured in order and returned in workflow_steps, letting you inspect the exact multi-turn call chain and parameter values passed between tools.

1. Core AI, Custom Tools & Prebuilt Tool Components in 12_travel_assistant.py

• Pydantic Contracts (TravelAssistantRequest & TravelAssistantResponse): Define the default trip brief (3 days in Goa with 2 friends on a ₹20,000 budget) and the response structure (query: str, response: str, and workflow_steps: list[str]).
• Step ② — Model Selection (model_id): Resolves the target Bedrock model ID, defaulting to "amazon.nova-lite-v1:0".
• Custom Tool 1 — @tool def get_weather_forecast(city: str, days: int) -> dict: Looks up city.lower() in the forecasts dictionary (covering "goa", "bangalore", "jaipur", and "manali", with a moderate fallback of 28°C / 20°C for unknown cities) and returns {"city": city, "days": days, "high_c": ..., "low_c": ..., "conditions": ...}.
• Custom Tool 2 — @tool def suggest_packing_list(high_c: int, low_c: int, days: int, conditions: str) -> list: Starts with base items (f"{days + 1} sets of clothes", "toiletries", "phone charger") and applies deterministic threshold rules:
  • high_c >= 30 → adds light cotton clothing, sunscreen, sunglasses, reusable water bottle.
  • low_c <= 15 → adds warm jacket, thermal layer; elif low_c <= 22 → adds light jacket for evenings.
  • "rain" or "shower" in conditions.lower() → adds compact umbrella, quick-dry footwear.
  • "snow" in conditions.lower() → adds gloves, woollen cap, waterproof boots.
• Custom Tool 3 — @tool def estimate_trip_cost(city: str, days: int, travellers: int = 1) -> dict: Computes accommodation (per_night.get(city.lower(), 3000) * days), food (1200 * days * travellers), and local transport (800 * days) in Indian rupees, returning the itemized breakdown and total_inr.
• Prebuilt Tool 4 — calculator (from strands_tools): Available to verify budget differences, per-person splits, or extra arithmetic.
• run_travel_assistant_agent(query: str) -> dict:
  • Step ③: Instantiates Agent with all four tools and a standing system_prompt instructing the agent to check weather first, suggest packing second, estimate cost third, and state whether the trip fits the user's budget.
  • Step ④: Runs response = agent(query) across multiple Bedrock Converse turns.
  • Step ⑤: Extracts all "toolUse" blocks from agent.messages into workflow_steps and returns {"query": query, "response": str(response), "workflow_steps": tool_calls}.

2. End-to-End AI Execution Walkthrough

1. run_travel_assistant_agent(query) creates a Strands Agent with [get_weather_forecast, suggest_packing_list, estimate_trip_cost, calculator] and the travel-planning system_prompt.
2. Following the system prompt for the Goa 3-day trip query, Bedrock starts by calling get_weather_forecast(city="Goa", days=3), which returns {"city": "Goa", "days": 3, "high_c": 32, "low_c": 26, "conditions": "humid, occasional showers"}.
3. On the next turn, Bedrock pipes those exact weather outputs into suggest_packing_list(high_c=32, low_c=26, days=3, conditions="humid, occasional showers"), receiving the 9-item checklist including cotton clothes, sunscreen, and a compact umbrella.
4. Bedrock also infers from "I'm going... with 2 friends" that travellers=3 and calls estimate_trip_cost(city="Goa", days=3, travellers=3), which returns stay_inr=10500, food_inr=10800, local_travel_inr=2400, and total_inr=23700 (optionally invoking calculator to compare ₹23,700 against the ₹20,000 budget).
5. Bedrock synthesizes the complete weather summary, packing checklist, itemized cost breakdown, and budget verdict (₹3,700 over the ₹20,000 budget), and Step ⑤ returns {"query": query, "response": str(response), "workflow_steps": tool_calls}.

Capstone Travel Assistant Agent Service.

"""Capstone Travel Assistant Agent Service.

Demonstrates comprehensive multi-step autonomous planning:
1. Custom domain tools plus one prebuilt tool:
   - `get_weather_forecast(city, days)`
   - `suggest_packing_list(high_c, low_c, days, conditions)`
   - `estimate_trip_cost(city, days, travellers)`
   - `calculator` (prebuilt from `strands_tools`)
2. The agentic loop feeds the weather forecast into packing recommendations,
   estimates stay, food and travel costs, and checks them against the traveller's budget.
"""

import os

from dotenv import find_dotenv, load_dotenv
from pydantic import BaseModel
from strands import Agent, tool
from strands.models.bedrock import BedrockModel
from strands_tools import calculator

# ① Load AWS credentials and Bedrock configuration from .env
load_dotenv(find_dotenv())


class TravelAssistantRequest(BaseModel):
    """Input payload for Travel Assistant Agent."""

    query: str = (
        "I'm going to Goa for 3 days with 2 friends. My budget is 20000 rupees. "
        "What should I pack, and does it fit my budget?"
    )


class TravelAssistantResponse(BaseModel):
    """Output payload from Travel Assistant Agent."""

    query: str
    response: str
    workflow_steps: list[str]


# ② Resolve the target Bedrock model identifier
model_id = os.getenv("BEDROCK_MODEL_ID", "amazon.nova-lite-v1:0")


@tool
def get_weather_forecast(city: str, days: int) -> dict:
    """Get the weather forecast for a city over a number of days.

    Args:
        city: Destination city name (e.g., "Goa", "Bangalore")
        days: Number of days in the trip
    """
    # Look up destination weather forecast in the mock catalog (or fall back to moderate defaults)
    forecasts = {
        "goa":       {"high_c": 32, "low_c": 26, "conditions": "humid, occasional showers"},
        "bangalore": {"high_c": 27, "low_c": 18, "conditions": "mild, light evening rain"},
        "jaipur":    {"high_c": 38, "low_c": 25, "conditions": "hot and dry"},
        "manali":    {"high_c": 14, "low_c": 3,  "conditions": "cold, chance of snow"},
    }
    data = forecasts.get(city.lower(), {"high_c": 28, "low_c": 20, "conditions": "moderate"})
    return {"city": city, "days": days, **data}


@tool
def suggest_packing_list(high_c: int, low_c: int, days: int, conditions: str) -> list:
    """Suggest what to pack based on temperatures, trip length and conditions.

    Args:
        high_c: Daytime high in Celsius
        low_c: Night-time low in Celsius
        days: Number of days in the trip
        conditions: Short description of expected weather
    """
    # Start with base clothing, toiletries, and electronics for the trip duration
    items = [f"{days + 1} sets of clothes", "toiletries", "phone charger"]

    # Append temperature- and weather-condition-specific gear
    if high_c >= 30:
        items += ["light cotton clothing", "sunscreen", "sunglasses", "reusable water bottle"]
    if low_c <= 15:
        items += ["warm jacket", "thermal layer"]
    elif low_c <= 22:
        items += ["light jacket for evenings"]
    if "rain" in conditions.lower() or "shower" in conditions.lower():
        items += ["compact umbrella", "quick-dry footwear"]
    if "snow" in conditions.lower():
        items += ["gloves", "woollen cap", "waterproof boots"]

    return items


@tool
def estimate_trip_cost(city: str, days: int, travellers: int = 1) -> dict:
    """Estimate the cost of a trip in Indian rupees.

    Args:
        city: Destination city
        days: Number of days
        travellers: Number of people travelling (default: 1)
    """
    # Calculate accommodation, food, and local transport costs in INR
    per_night = {"goa": 3500, "bangalore": 3000, "jaipur": 2500, "manali": 2800}
    stay = per_night.get(city.lower(), 3000) * days
    food = 1200 * days * travellers
    local_travel = 800 * days
    total = stay + food + local_travel

    # Return itemized and total estimated trip cost breakdown
    return {
        "city": city,
        "days": days,
        "travellers": travellers,
        "stay_inr": stay,
        "food_inr": food,
        "local_travel_inr": local_travel,
        "total_inr": total,
    }


def run_travel_assistant_agent(query: str) -> dict:
    """Run the Capstone Travel Assistant Agent."""
    # ③ Initialize the Travel Assistant Agent with custom tools, calculator, and system prompt
    agent = Agent(
        model=BedrockModel(model_id=model_id),
        tools=[get_weather_forecast, suggest_packing_list, estimate_trip_cost, calculator],
        system_prompt=(
            "You are a practical travel assistant. "
            "When asked about a trip: check the weather first, then suggest what to pack "
            "based on that weather, then estimate the cost. "
            "Always say whether the trip fits the user's budget, and keep advice concise."
        ),
    )

    # ④ Run the multi-step planning loop (weather -> packing -> cost -> budget check)
    response = agent(query)

    # ⑤ Collect the tool calls in the order the agent made them
    tool_calls = [
        f"{block['toolUse']['name']}({block['toolUse']['input']})"
        for message in agent.messages
        for block in message["content"]
        if "toolUse" in block
    ]

    return {"query": query, "response": str(response), "workflow_steps": tool_calls}

Flask Blueprint server and route handler (travel_assistant_route) in src/python/app.py.

"""Flask server for 05-ai-aws-strands."""

import asyncio
import importlib.util
import os
import sys
from pathlib import Path

from dotenv import find_dotenv, load_dotenv
from flask import Blueprint, Flask, jsonify, request
from flask_cors import CORS
from rate_limiter import check_rate_limit

load_dotenv(find_dotenv())

PATH_PREFIX = os.environ.get("PATH_PREFIX", "")
STATIC_DIR = Path(__file__).resolve().parents[1]
app = Flask(__name__, static_folder=str(STATIC_DIR))
CORS(app)

bp = Blueprint("main", __name__)

_CUR_DIR = Path(__file__).resolve().parent


def _load_service(filename: str, alias: str):
    # ① Resolve the numbered service file path and create its import spec
    file_path = _CUR_DIR / filename
    spec = importlib.util.spec_from_file_location(f"services.{alias}", file_path)
    if spec is None or spec.loader is None:
        raise ImportError(f"Cannot load module from {file_path}")
    # ② Register and execute the module so its functions can be called by routes
    mod = importlib.util.module_from_spec(spec)
    sys.modules[f"services.{alias}"] = mod
    spec.loader.exec_module(mod)
    return mod


s_hello = _load_service("01_hello_world_agent.py", "s_hello")
s_langgraph = _load_service("02_hello_world_langgraph.py", "s_langgraph")
s_fn_tool = _load_service("03_function_to_tool.py", "s_fn_tool")
s_tip = _load_service("04_tip_calculator.py", "s_tip")
s_sales = _load_service("05_multi_tool_sales.py", "s_sales")
s_inv = _load_service("06_custom_tool_inventory.py", "s_inv")
s_calc = _load_service("07_prebuilt_tools.py", "s_calc")
s_multi_pre = _load_service("08_multi_prebuilt_tools.py", "s_multi_pre")
s_aws = _load_service("09_use_aws.py", "s_aws")
s_cls_tool = _load_service("10_class_based_tools.py", "s_cls_tool")
s_async_tool = _load_service("11_async_tools.py", "s_async_tool")
s_travel = _load_service("12_travel_assistant.py", "s_travel")


@bp.before_request
def enforce_rate_limit():
    # ① Only rate-limit mutating POST requests
    if request.method == "POST":
        # ② Check whether this client has exceeded 30 requests per hour
        blocked, msg, retry_after = check_rate_limit(
            request, max_requests=30, window_seconds=3600
        )
        if blocked:
            # ③ Return HTTP 429 with Retry-After header when rate-limited
            resp = jsonify({"error": msg})
            resp.status_code = 429
            resp.headers["Retry-After"] = str(retry_after)
            return resp


@bp.route("/")
def index():
    # ① Read index.html and inject the runtime PATH_PREFIX into data-api-base
    with open(os.path.join(app.static_folder, "index.html"), encoding="utf-8") as f:
        html = f.read()
    html = html.replace('data-api-base=""', f'data-api-base="{PATH_PREFIX}"')
    # ② Return the patched HTML page
    return app.response_class(html, mimetype="text/html")


@bp.route("/css/<path:filename>")
def css(filename):
    return app.send_static_file(os.path.join("css", filename))


@bp.route("/js/<path:filename>")
def js(filename):
    return app.send_static_file(os.path.join("js", filename))


@bp.route("/explanation/<path:filename>")
@bp.route("/info/<path:filename>")
def explanation_pages(filename):
    return app.send_static_file(os.path.join("explanation", filename))


@bp.route("/theory/<path:filename>")
def theory_pages(filename):
    """Serve the theory documentation pages from src/theory."""
    return app.send_static_file(os.path.join("theory", filename))


@bp.route("/api/hello-world-agent", methods=["POST"])
def hello_world_route():
    # ① Parse prompt from JSON payload and reject empty values
    data = request.get_json(force=True, silent=True) or {}
    prompt = (data.get("prompt") or data.get("message") or "").strip()
    if not prompt:
        return jsonify({"error": "Prompt cannot be empty"}), 400
    try:
        # ② Invoke the Strands Hello World Agent in 01_hello_world_agent.py
        res = s_hello.run_hello_world_agent(prompt)
        res["result"] = res.get("response") or res.get("answer") or ""
        return jsonify(res)
    except Exception as exc:
        return jsonify({"error": str(exc)}), 500


@bp.route("/api/langgraph-agent", methods=["POST"])
def langgraph_route():
    # ① Parse prompt from JSON payload and reject empty values
    data = request.get_json(force=True, silent=True) or {}
    prompt = (data.get("prompt") or data.get("message") or "").strip()
    if not prompt:
        return jsonify({"error": "Prompt cannot be empty"}), 400
    try:
        # ② Invoke the LangGraph + Bedrock Converse agent in 02_hello_world_langgraph.py
        res = s_langgraph.run_langgraph_agent(prompt)
        res["result"] = res.get("response") or res.get("answer") or ""
        return jsonify(res)
    except Exception as exc:
        return jsonify({"error": str(exc)}), 500


@bp.route("/api/function-to-tool", methods=["POST"])
def function_to_tool_route():
    # ① Parse question from JSON payload and reject empty values
    data = request.get_json(force=True, silent=True) or {}
    question = (data.get("question") or data.get("message") or "").strip()
    if not question:
        return jsonify({"error": "Question cannot be empty"}), 400
    try:
        # ② Run the HTTP server health-check tool agent in 03_function_to_tool.py
        res = s_fn_tool.run_function_to_tool_agent(question)
        res["result"] = res.get("response") or res.get("answer") or ""
        return jsonify(res)
    except Exception as exc:
        return jsonify({"error": str(exc)}), 500


@bp.route("/api/tip-calculator", methods=["POST"])
def tip_calculator_route():
    # ① Parse query from JSON payload and reject empty values
    data = request.get_json(force=True, silent=True) or {}
    query = (data.get("query") or data.get("message") or "").strip()
    if not query:
        return jsonify({"error": "Query cannot be empty"}), 400
    try:
        # ② Run the multi-parameter tip calculator agent in 04_tip_calculator.py
        res = s_tip.run_tip_calculator_agent(query)
        res["result"] = res.get("response") or res.get("answer") or ""
        return jsonify(res)
    except Exception as exc:
        return jsonify({"error": str(exc)}), 500


@bp.route("/api/multi-tool-sales", methods=["POST"])
def multi_tool_sales_route():
    # ① Parse task from JSON payload and reject empty values
    data = request.get_json(force=True, silent=True) or {}
    task = (data.get("task") or data.get("message") or "").strip()
    if not task:
        return jsonify({"error": "Task cannot be empty"}), 400
    try:
        # ② Run the 3-tool sales pipeline agent in 05_multi_tool_sales.py
        res = s_sales.run_multi_tool_sales_agent(task)
        res["result"] = res.get("response") or res.get("answer") or ""
        return jsonify(res)
    except Exception as exc:
        return jsonify({"error": str(exc)}), 500


@bp.route("/api/custom-inventory", methods=["POST"])
def custom_inventory_route():
    # ① Parse question from JSON payload and reject empty values
    data = request.get_json(force=True, silent=True) or {}
    question = (data.get("question") or data.get("message") or "").strip()
    if not question:
        return jsonify({"error": "Question cannot be empty"}), 400
    try:
        # ② Run the custom inventory lookup agent in 06_custom_tool_inventory.py
        res = s_inv.run_custom_inventory_agent(question)
        res["result"] = res.get("response") or res.get("answer") or ""
        return jsonify(res)
    except Exception as exc:
        return jsonify({"error": str(exc)}), 500


@bp.route("/api/prebuilt-calculator", methods=["POST"])
def prebuilt_calculator_route():
    # ① Parse math query from JSON payload and reject empty values
    data = request.get_json(force=True, silent=True) or {}
    query = (data.get("query") or data.get("message") or "").strip()
    if not query:
        return jsonify({"error": "Query cannot be empty"}), 400
    try:
        # ② Run the prebuilt calculator tool agent in 07_prebuilt_tools.py
        res = s_calc.run_prebuilt_calculator_agent(query)
        res["result"] = res.get("response") or res.get("answer") or ""
        return jsonify(res)
    except Exception as exc:
        return jsonify({"error": str(exc)}), 500


@bp.route("/api/multi-prebuilt-tools", methods=["POST"])
def multi_prebuilt_tools_route():
    # ① Parse task from JSON payload and reject empty values
    data = request.get_json(force=True, silent=True) or {}
    task = (data.get("task") or data.get("message") or "").strip()
    if not task:
        return jsonify({"error": "Task cannot be empty"}), 400
    try:
        # ② Run the http_request + calculator agent in 08_multi_prebuilt_tools.py
        res = s_multi_pre.run_multi_prebuilt_tools_agent(task)
        res["result"] = res.get("response") or res.get("answer") or ""
        return jsonify(res)
    except Exception as exc:
        return jsonify({"error": str(exc)}), 500


@bp.route("/api/use-aws", methods=["POST"])
def use_aws_route():
    # ① Parse natural-language AWS prompt from JSON payload and reject empty values
    data = request.get_json(force=True, silent=True) or {}
    prompt = (data.get("prompt") or data.get("message") or "").strip()
    if not prompt:
        return jsonify({"error": "Prompt cannot be empty"}), 400
    try:
        # ② Run the prebuilt use_aws (boto3) agent in 09_use_aws.py
        res = s_aws.run_use_aws_agent(prompt)
        res["result"] = res.get("response") or res.get("answer") or ""
        return jsonify(res)
    except Exception as exc:
        return jsonify({"error": str(exc)}), 500


@bp.route("/api/class-based-tools", methods=["POST"])
def class_based_tools_route():
    # ① Parse query from JSON payload and reject empty values
    data = request.get_json(force=True, silent=True) or {}
    query = (data.get("query") or data.get("message") or "").strip()
    if not query:
        return jsonify({"error": "Query cannot be empty"}), 400
    try:
        # ② Run the stateful InventoryTools class agent in 10_class_based_tools.py
        res = s_cls_tool.run_class_based_tools_agent(query)
        res["result"] = res.get("response") or res.get("answer") or ""
        return jsonify(res)
    except Exception as exc:
        return jsonify({"error": str(exc)}), 500


@bp.route("/api/async-tools", methods=["POST"])
def async_tools_route():
    # ① Parse query from JSON payload and reject empty values
    data = request.get_json(force=True, silent=True) or {}
    query = (data.get("query") or data.get("message") or "").strip()
    if not query:
        return jsonify({"error": "Query cannot be empty"}), 400
    try:
        # ② Run the async parallel warehouse tools agent in 11_async_tools.py
        res = asyncio.run(s_async_tool.run_async_tools_agent(query))
        res["result"] = res.get("response") or res.get("answer") or ""
        return jsonify(res)
    except Exception as exc:
        return jsonify({"error": str(exc)}), 500


@bp.route("/api/travel-assistant", methods=["POST"])
def travel_assistant_route():
    # ① Parse travel query from JSON payload and reject empty values
    data = request.get_json(force=True, silent=True) or {}
    query = (data.get("query") or data.get("message") or "").strip()
    if not query:
        return jsonify({"error": "Query cannot be empty"}), 400
    try:
        # ② Run the 4-tool travel assistant agent in 12_travel_assistant.py
        res = s_travel.run_travel_assistant_agent(query)
        res["result"] = res.get("response") or res.get("answer") or ""
        return jsonify(res)
    except Exception as exc:
        return jsonify({"error": str(exc)}), 500


app.register_blueprint(bp, url_prefix=PATH_PREFIX)

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5000)

Use this numbered checklist to review the multi-tool orchestration, semantic extraction, and hybrid rule-engine patterns in the 🏖️ Capstone Travel Planner demo:

1. Combining Custom @tool Functions with Prebuilt strands_tools: A single Strands Agent can mix domain-specific custom tools (get_weather_forecast, suggest_packing_list, estimate_trip_cost) and off-the-shelf prebuilt tools (calculator) in the same tools=[...] list.
2. Declarative Workflow Steering via system_prompt: Instructing the agent in the system prompt ("check the weather first, then suggest what to pack based on that weather, then estimate the cost") enforces a logical tool order for full trip queries while keeping the agent adaptable to single-topic questions.
3. Cross-Tool Data Dependency (Weather → Packing): Because suggest_packing_list(high_c, low_c, days, conditions) requires temperature and condition values that are not in the user's prompt, the agent must execute get_weather_forecast first and feed its output into suggest_packing_list.
4. Pragmatic Natural-Language Inference ("with 2 friends" → travellers=3): Unlike regex or keyword parsers that would extract the literal number 2, the LLM understands that the speaker plus two friends equals 3 travellers and passes travellers=3 to estimate_trip_cost().
5. Optional Parameters with Default Values: Defining def estimate_trip_cost(city: str, days: int, travellers: int = 1) marks travellers as optional in the generated JSON Schema so solo trip prompts do not fail if headcount is omitted.
6. Hybrid Rule Engine + LLM Synthesis: suggest_packing_list() uses deterministic Python if/elif thresholds (high_c >= 30, low_c <= 15, "rain" in conditions) so essential gear is never forgotten, while the LLM formats the resulting list into natural prose.
7. Graceful Dictionary Fallbacks: Both get_weather_forecast() and estimate_trip_cost() normalize city names with city.lower() and provide sensible defaults via .get(...) so queries for unlisted cities still succeed without raising a KeyError.
8. Deterministic Budget Verification: By computing itemized stay, food, and local travel totals in Python (total_inr = 23700) and comparing against the user's ₹20,000 budget (with calculator available for exact differences), the agent accurately reports that the trip exceeds the budget by ₹3,700.
9. End-to-End Traceability (workflow_steps): Capturing every toolUse call in chronological order inside workflow_steps lets developers and users verify the exact sequence of tools and arguments executed during the capstone planning loop.