☁️ AWS Cloud Assistant#

Natural Language AWS Agent (use_aws) equips a Strands agent with the prebuilt use_aws tool from strands_tools, which acts as a universal bridge between natural-language requests and the AWS SDK for Python (boto3).

When you ask "List all S3 buckets in my account" or "Describe my EC2 instances in us-east-1", the agent translates your plain-English question into a structured use_aws invocation specifying service_name (e.g., "s3"), operation_name (e.g., "list_buckets"), parameters, region, and a human-readable label. The tool executes the real boto3 API call against the configured AWS account and returns the response for the agent to summarize.

1. One Meta-Tool vs. Thousands of Individual API Tools

AWS has over 200 services and tens of thousands of API actions. Registering a separate @tool function for every AWS operation would blow past any LLM's context window. Instead, use_aws is a single parameterized meta-tool: the model relies on its pretraining knowledge of boto3 client names and snake_case operation names to fill in service_name, operation_name, and parameters dynamically.

2. IAM Least Privilege Is the True Security Boundary

Prompt instructions like "Please do not delete anything" are soft guardrails that can be bypassed by prompt injection or model mistakes. When giving an agent cloud access, the hard security boundary is the AWS IAM Policy attached to the credentials or instance role — always scope agent credentials to read-only permissions (ReadOnlyAccess or specific List* / Describe* actions).

3. Built-In Consent Refusal for Mutating & Sensitive Calls

As a second layer of defense, use_aws inspects the operation name before calling boto3. Mutating operations (such as create_*, delete_*, terminate_*, put_*) and sensitive secret reads (such as secretsmanager.get_secret_value) trigger an interactive confirmation prompt, which automatically fails closed when running inside a non-interactive web server process.

1. Core AI & Prebuilt use_aws Components in 09_use_aws.py

• Pydantic Contracts (UseAwsRequest & UseAwsResponse): Specify the input payload (prompt: str defaulting to "List all S3 buckets in my account") and output payload (prompt: str, response: str, and operations_executed: list[str]).
• Step ② — Model Selection (model_id): Resolves the target Bedrock model ID, defaulting to "amazon.nova-lite-v1:0".
• Prebuilt Meta-Tool (from strands_tools import use_aws): Exposes the AWS SDK for Python (boto3) through a single structured tool schema accepting service_name (e.g., "s3", "sts", "dynamodb"), operation_name (e.g., "list_buckets", "get_caller_identity"), parameters (dict), region (str), and label (str).
• run_use_aws_agent(prompt: str) -> dict:
  • Step ③: Instantiates Agent(model=BedrockModel(model_id=model_id), tools=[use_aws], system_prompt="You are an AWS assistant that helps manage cloud resources.").
  • Step ④: Calls response = agent(prompt), where Bedrock maps the user's English request to the exact boto3 service and snake_case operation name, invokes use_aws, and summarizes the returned AWS API response.
  • Step ⑤: Extracts all "toolUse" blocks from agent.messages into operations_executed and returns {"prompt": prompt, "response": str(response), "operations_executed": tool_calls}.

2. End-to-End AI Execution Walkthrough

1. run_use_aws_agent(prompt) initializes a Strands Agent with BedrockModel(model_id=model_id), tools=[use_aws], and system_prompt="You are an AWS assistant that helps manage cloud resources.".
2. Calling agent(prompt) (e.g., for "List all S3 buckets in my account") sends the prompt and the use_aws tool specification to Amazon Bedrock via the Converse API.
3. Bedrock translates the natural-language request into a toolUse block calling use_aws(service_name="s3", operation_name="list_buckets", parameters={}, region="us-east-1", label="List all S3 buckets").
4. Inside use_aws, the tool verifies that list_buckets is a non-mutating read operation, creates a boto3.client("s3"), executes client.list_buckets() against AWS, and returns the bucket list in a toolResult block.
5. Bedrock formats the bucket names and creation dates into a clean answer, and Step ⑤ collects operations_executed from agent.messages and returns {"prompt": prompt, "response": str(response), "operations_executed": tool_calls}.

Natural Language Cloud / AWS Agent Service.

"""Natural Language Cloud / AWS Agent Service.

Demonstrates `use_aws` natural language cloud operations:
1. The prebuilt `use_aws` tool from `strands_tools` exposes the whole AWS API (via boto3) to the agent.
2. The agent translates a plain-English request into service_name + operation_name + parameters.
3. Calls run against the real AWS account of the configured credentials, so give them a read-only IAM policy.
   Mutating calls and secret reads ask for confirmation, which is refused in the server (no terminal).
"""

import os

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

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


class UseAwsRequest(BaseModel):
    """Input payload for Use AWS Agent."""

    prompt: str = "List all S3 buckets in my account"


class UseAwsResponse(BaseModel):
    """Output payload from Use AWS Agent."""

    prompt: str
    response: str
    operations_executed: list[str]


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


def run_use_aws_agent(prompt: str = "List all S3 buckets in my account") -> dict:
    """Run the Use AWS Agent."""
    # ③ Initialize the Strands Agent with the prebuilt use_aws tool and cloud assistant system prompt
    agent = Agent(
        model=BedrockModel(model_id=model_id),
        tools=[use_aws],
        system_prompt="You are an AWS assistant that helps manage cloud resources.",
    )

    # ④ Execute the natural language AWS request
    response = agent(prompt)

    # ⑤ List the AWS API operations the agent chose (service, operation, parameters)
    tool_calls = [
        f"{block['toolUse']['name']}({block['toolUse']['input']})"
        for message in agent.messages
        for block in message["content"]
        if "toolUse" in block
    ]

    return {"prompt": prompt, "response": str(response), "operations_executed": tool_calls}

Flask Blueprint server and route handler (use_aws_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 core concepts, boto3 meta-tool architecture, and cloud security guardrails in the ☁️ AWS Cloud Assistant demo:

1. The Meta-Tool Pattern (use_aws): Instead of registering thousands of separate tools for 200+ AWS services (which would overflow the LLM context window), use_aws exposes a single parameterized schema (service_name, operation_name, parameters, region, label) backed by boto3.
2. Leveraging Pretrained boto3 Knowledge: Frontier and Nova models already know boto3 service names (like "s3", "ec2", "sts") and snake_case operation names (like "list_buckets", "describe_instances") from pretraining, allowing zero-shot translation from English to AWS API calls.
3. Hard Security Boundary via IAM Least Privilege: Because use_aws executes real API calls using the host environment's AWS credentials, the true security boundary is the IAM Policy attached to those credentials — always restrict agent roles to read-only permissions.
4. Built-In Consent Guardrail for Mutating Calls: use_aws inspects operation_name before invoking boto3; mutating verbs (create_*, delete_*, terminate_*, put_*) and secret reads (get_secret_value) require interactive terminal confirmation, which automatically fails closed in a headless web server.
5. Self-Documenting Calls via label: The use_aws schema requires a human-readable label argument on every call, making audit logs and UI tool traces immediately understandable alongside the raw boto3 parameters.
6. Multi-Call Discovery & Error Recovery: If an operation requires a resource identifier (such as a bucket name or table name) or if the model supplies an invalid parameter on its first attempt, the Strands Converse loop lets the agent list resources first or self-correct based on the returned boto3 exception.
7. Auditing Cloud Actions (operations_executed): Extracting every toolUse block from agent.messages into operations_executed gives operators full visibility into which AWS services, regions, and parameters were queried.
8. Shared Credentials Between Bedrock and use_aws: Both BedrockModel (which calls bedrock-runtime:Converse) and use_aws (which calls target services like S3 or STS) authenticate through the standard boto3 credential chain loaded via load_dotenv() or IAM roles.