📈 Sales Pipeline Agent#

Multi-Tool Sales Pipeline Agent demonstrates autonomous multi-step planning and data dependency chaining across three specialized tools: get_sales_data(quarter), analyze_sales(revenue, deals), and send_email(to, subject, body).

When given a high-level goal like "Get Q1 sales data, analyze it, and email the summary to boss@company.com", no Python code specifies the execution order. Instead, the agent inspects the tool signatures, deduces that analyze_sales requires revenue and deals numbers that only get_sales_data can provide, and chains all three tools across sequential turns until the email is dispatched.

1. Implicit Dependency Resolution via Tool Signatures

How does the agent know which tool to call first? Look at the input parameters:

1. get_sales_data(quarter: str) only needs "Q1", which is already present in the user's prompt — so the agent calls it on Turn 1.
2. analyze_sales(revenue: float, deals: int) needs numerical metrics that were unknown until get_sales_data returned {"revenue": 125000, "deals": 42} — so the agent calls it on Turn 2 using Turn 1's output.
3. send_email(to: str, subject: str, body: str) needs the analysis summary text from Turn 2 — so the agent calls it on Turn 3.

2. Single-Responsibility Principle for Agent Tools

A common anti-pattern is building a monolithic tool like fetch_analyze_and_email_sales(quarter, recipient). While that works for one rigid workflow, it cannot answer "Just compare Q1 and Q2 sales" without sending an unwanted email. Keeping get_sales_data, analyze_sales, and send_email atomic lets the agent recombine them for any ad-hoc query.

3. Mocking External Side Effects During Development

By returning deterministic mock data in get_sales_data and simulating SMTP delivery in send_email, you can test and verify the agent's multi-step planning sequence (via tool_sequence in the response) safely before wiring real CRM or email APIs.

1. Core AI & Multi-Tool Pipeline Components in 05_multi_tool_sales.py

• Pydantic Contracts (MultiToolSalesRequest & MultiToolSalesResponse): Document the input task: str and output dictionary fields (task: str, response: str, tool_calls: list[str]).
• Step ② Model Identifier (model_id): Resolves the target Bedrock model ID, defaulting to "amazon.nova-lite-v1:0".
• @tool def get_sales_data(quarter: str) -> dict: Step 1 of the implicit data dependency chain. Accepts a quarter identifier (e.g., "Q1" or "4") and returns mock CRM metrics: {"revenue": 1250000, "deals": 47, "quarter": quarter}.
• @tool def analyze_sales(revenue: int, deals: int, quarter: str) -> str: Step 2 of the chain. Accepts revenue: int, deals: int, and quarter: str (values discovered from get_sales_data), computes avg_deal = revenue / deals, and returns the formatted executive summary string f"Q{quarter}: ${revenue:,} revenue, {deals} deals, ${avg_deal:,.0f} avg deal size".
• @tool def send_email(to: str, subject: str, body: str) -> str: Step 3 of the chain. Accepts the recipient address, subject line, and the summary body produced by analyze_sales, and returns f"Email sent to {to}".
• run_multi_tool_sales_agent(task: str) -> dict:
  • Step ③: Instantiates Agent(model=BedrockModel(model_id=model_id), tools=[get_sales_data, analyze_sales, send_email]) without hardcoding an execution order.
  • Step ④: Calls response = agent(task), allowing the agentic loop to plan and execute the three-step workflow across turns.
  • Step ⑤: Extracts every "toolUse" block from agent.messages in chronological order into tool_calls and returns {"task": task, "response": str(response), "tool_calls": tool_calls}.

2. End-to-End AI Execution Walkthrough

1. run_multi_tool_sales_agent(task) initializes the Strands Agent with [get_sales_data, analyze_sales, send_email] and passes the user task (e.g., "Pull last quarter's sales data and email a summary to the team").
2. On Turn 1, Bedrock inspects all three tool schemas, recognizes that analyze_sales requires revenue and deals that are not yet known, and invokes get_sales_data(quarter=...) first.
3. Strands executes get_sales_data() and returns {"revenue": 1250000, "deals": 47, ...} to Bedrock in a toolResult block.
4. On Turn 2, Bedrock feeds those exact integers into analyze_sales(revenue=1250000, deals=47, quarter=...), receiving the formatted summary string with the $26,596 average deal size.
5. On Turn 3, Bedrock passes that summary string into send_email(to=..., subject=..., body=...) and receives "Email sent to ...".
6. On Turn 4, Bedrock produces the final confirmation message, and Step ⑤ collects all three tool calls in chronological order into tool_calls.

Multi-Tool Sales Pipeline Agent Service.

"""Multi-Tool Sales Pipeline Agent Service.

Demonstrates autonomous multi-tool planning and chained orchestration:
1. Three tools provided without explicit execution order:
   - `get_sales_data(quarter: str)`
   - `analyze_sales(revenue: int, deals: int, quarter: str)`
   - `send_email(to: str, subject: str, body: str)`
2. The agent plans the workflow: Fetch Data -> Compute Analysis -> Dispatch Email.
3. Passes the output of each tool directly as inputs into the next step of the agentic loop.
"""

import os

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

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


class MultiToolSalesRequest(BaseModel):
    """Input payload for Multi-Tool Sales Agent."""

    task: str = "Pull last quarter's sales data and email a summary to the team"


class MultiToolSalesResponse(BaseModel):
    """Output payload from Multi-Tool Sales Agent."""

    task: str
    response: str
    tool_calls: list[str]


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


@tool
def get_sales_data(quarter: str) -> dict:
    """Retrieve sales data for a specific quarter."""
    # Return mock quarterly revenue and closed deal count
    return {"revenue": 1250000, "deals": 47, "quarter": quarter}


@tool
def analyze_sales(revenue: int, deals: int, quarter: str) -> str:
    """Calculate key metrics from sales data."""
    # Compute average deal size and format the executive summary string
    avg_deal = revenue / deals
    return f"Q{quarter}: ${revenue:,} revenue, {deals} deals, ${avg_deal:,.0f} avg deal size"


@tool
def send_email(to: str, subject: str, body: str) -> str:
    """Send an email message."""
    # Simulate dispatching the summary email to the target recipient
    return f"Email sent to {to}"


def run_multi_tool_sales_agent(
    task: str = "Pull last quarter's sales data and email a summary to the team",
) -> dict:
    """Run the Multi-Tool Sales Agent."""
    # ③ Initialize the Strands Agent with all three sales pipeline tools (no order given)
    agent = Agent(
        model=BedrockModel(model_id=model_id),
        tools=[get_sales_data, analyze_sales, send_email],
    )

    # ④ Execute the multi-step task (get_sales_data -> analyze_sales -> send_email)
    response = agent(task)

    # ⑤ 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 {"task": task, "response": str(response), "tool_calls": tool_calls}


# RULE OF THUMB: build focused, single-purpose tools.
# One mega-tool that does everything is harder to maintain, harder for the
# agent to reason about, and impossible to reuse elsewhere.

Flask Blueprint server and route handler (sales_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)

Review these key architectural lessons on multi-tool planning and sequential orchestration in AWS Strands:

1. Autonomous Multi-Step Planning: You pass tools=[get_sales_data, analyze_sales, send_email] in any order, and the LLM plans the execution sequence dynamically based on what data each tool requires and produces.
2. Implicit Data Dependency Resolution: Because analyze_sales(revenue: int, deals: int, quarter: str) requires revenue and deals that are not in the user's prompt, the agent deduces it must call get_sales_data(quarter) first.
3. Single-Responsibility Tool Design: Splitting data retrieval (get_sales_data), computation (analyze_sales), and side-effect delivery (send_email) into atomic tools makes each tool reusable across arbitrary combinations of user requests.
4. Why Mega-Tools Fail in Production: A single monolithic function that fetches, analyzes, and emails in one step cannot answer read-only questions (like "What was Q1 revenue?") without triggering unwanted side effects.
5. Multi-Turn State Accumulation: Across the 3 tool turns, agent.messages accumulates every assistant toolUse request and user-role toolResult observation so the model has full context when composing the email body.
6. Safe Side-Effect Mocking: Simulating SMTP delivery inside send_email() allows end-to-end verification of the agent's planning logic before connecting production email credentials.
7. Chronological Execution Trace (tool_calls): Iterating through agent.messages in order produces the exact step-by-step sequence (get_sales_data → analyze_sales → send_email) for observability and UI display.