📦 Custom Inventory Tool#

Custom Inventory Tool Agent shows how to bridge an LLM's general language understanding with proprietary enterprise data (such as SKU catalogs, stock levels, and warehouse locations) that no public model could ever know from pretraining.

The service defines @tool def check_inventory(product_id: str) -> str backed by an internal SKU dictionary (PROD-123 Wireless Mouse, PROD-456 USB-C Cable, PROD-789 Laptop Stand). When you ask about one or more product IDs, the Bedrock agent extracts each SKU, invokes check_inventory(), and turns the returned stock status into a clear customer-ready answer.

1. Bridging the Knowledge Cutoff & Private Data Gap

Foundation models are trained on public internet text up to a fixed cutoff date; they have zero visibility into your company's internal SKUs (PROD-123) or live stock counts. Exposing a read-only lookup function via @tool gives the model on-demand access to authoritative internal state without fine-tuning or stuffing the entire catalog into the prompt.

2. Google-Style Args: Docstrings for Parameter Guidance

Strands parses the Args: block inside the function's docstring: product_id: The product ID to check (e.g., "PROD-123"). Including a concrete format example ("PROD-123") inside the parameter description teaches the model to normalize user inputs (such as "product 123" or "PROD-123") into the exact string pattern the backend expects.

3. Graceful Handling of Unknown Identifiers

When a user asks about a non-existent SKU like PROD-999, the tool does not raise a Python KeyError. Instead, it returns a clear domain message: "Product PROD-999 not found in inventory". Returning structured, human-readable not-found messages lets the LLM inform the user politely rather than crashing or hallucinating.

4. Drop-In Backend Replacement

Because the agent only interacts with the function signature and return string of check_inventory(product_id), swapping the in-memory dictionary for a PostgreSQL SELECT query or a Shopify REST API call requires zero changes to the agent setup.

1. Core AI & Domain Tool Components in 06_custom_tool_inventory.py

• Pydantic Contracts (CustomInventoryRequest & CustomInventoryResponse): Document the input question: str = "Is PROD-123 in stock?" and output dictionary fields (question: 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 check_inventory(product_id: str) -> str:
  • Docstring & Example Format: The Args: block specifies product_id: The product ID to check (e.g., "PROD-123"), giving the LLM a concrete pattern so it extracts SKU identifiers accurately.
  • Mock Inventory Catalog: Defines inventory = {"PROD-123": 15, "PROD-456": 0, "PROD-789": 8} representing an internal domain data store.
  • Safe Lookup & Status Formatting: Uses quantity = inventory.get(product_id, 0) so unknown SKUs safely default to 0 without raising a KeyError. Returns f"Product {product_id} is in stock. We have {quantity} units available." when quantity > 0, or f"Product {product_id} is currently out of stock." otherwise.
• run_custom_inventory_agent(question: str) -> dict:
  • Step ③: Instantiates Agent(model=BedrockModel(model_id=model_id), tools=[check_inventory]).
  • Step ④: Executes response = agent(question) through the Strands agent loop.
  • Step ⑤: Extracts every "toolUse" block from agent.messages into tool_calls and returns {"question": question, "response": str(response), "tool_calls": tool_calls}.

2. End-to-End AI Execution Walkthrough

1. run_custom_inventory_agent(question) creates a fresh Strands Agent registered with check_inventory and calls agent(question) (e.g., for "Is PROD-123 in stock?").
2. On Turn 1, Bedrock Nova Lite inspects the tool schema, extracts product_id="PROD-123", and emits a toolUse block.
3. Strands invokes check_inventory("PROD-123") in Python, which looks up "PROD-123" in inventory, finds 15, and returns "Product PROD-123 is in stock. We have 15 units available." in a toolResult block.
4. On Turn 2, Bedrock reads the tool output and writes a natural-language reply confirming 15 units are available.
5. Step ⑤ records check_inventory({'product_id': 'PROD-123'}) in tool_calls and returns {"question": question, "response": str(response), "tool_calls": tool_calls}.

Custom Inventory Tool Agent Service.

"""Custom Inventory Tool Agent Service.

Demonstrates building domain-specific custom tools:
1. Simulates an internal proprietary system or inventory database.
2. Custom tool `check_inventory(product_id: str)` with strict typing & docstrings.
3. Natural language query resolution against mock product inventory.
"""

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 CustomInventoryRequest(BaseModel):
    """Input payload for Custom Inventory Agent."""

    question: str = "Is PROD-123 in stock?"


class CustomInventoryResponse(BaseModel):
    """Output payload from Custom Inventory Agent."""

    question: 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 check_inventory(product_id: str) -> str:
    """Check if a product is in stock.

    Args:
        product_id: The product ID to check (e.g., "PROD-123")
    """
    # Define the mock product inventory catalog (replace with a real DB query in production)
    inventory = {
        "PROD-123": 15,
        "PROD-456": 0,
        "PROD-789": 8,
    }

    # Look up the available unit count for the requested SKU
    quantity = inventory.get(product_id, 0)

    # Return a human-readable stock status message
    if quantity > 0:
        return f"Product {product_id} is in stock. We have {quantity} units available."
    else:
        return f"Product {product_id} is currently out of stock."


def run_custom_inventory_agent(question: str = "Is PROD-123 in stock?") -> dict:
    """Run the Custom Inventory Agent."""
    # ③ Initialize the Strands Agent with BedrockModel and the custom check_inventory tool
    agent = Agent(model=BedrockModel(model_id=model_id), tools=[check_inventory])

    # ④ Run the agent on the inventory question
    response = agent(question)

    # ⑤ Collect every tool call the agent made from its message history
    tool_calls = [
        f"{block['toolUse']['name']}({block['toolUse']['input']})"
        for message in agent.messages
        for block in message["content"]
        if "toolUse" in block
    ]

    return {"question": question, "response": str(response), "tool_calls": tool_calls}

Flask Blueprint server and route handler (inventory_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 these takeaways to review how custom Strands tools connect foundation models to proprietary enterprise data:

1. Overcoming Pretraining Knowledge Boundaries: Public LLMs have zero visibility into private SKU catalogs or real-time warehouse counts; custom @tool functions give the model on-demand, read-only access to authoritative internal records.
2. Format Examples in Parameter Docstrings: Including (e.g., "PROD-123") inside the Args: docstring for product_id guides the LLM to normalize user queries into the exact identifier format expected by the database.
3. Defensive Lookups with dict.get(): Using inventory.get(product_id, 0) prevents unhandled KeyError exceptions when a user asks about an unknown or mistyped SKU.
4. Clear Domain Status Strings: Returning explicit natural-language status messages ("Product PROD-123 is in stock. We have 15 units available." vs. "...currently out of stock.") eliminates ambiguity during the LLM's observation step.
5. Clean Separation of Interface and Storage: Because the agent only sees the signature and docstring of check_inventory(product_id: str) -> str, replacing the mock dictionary with a live SQL or ERP API call requires zero changes to the agent configuration.
6. Multi-SKU Fan-Out Support: If a user asks about two products in one question (e.g., "Check PROD-123 and PROD-456"), the agent automatically invokes check_inventory twice and compares both results in its reply.
7. Complete Tool-Use Audit Trail: Extracting toolUse blocks from agent.messages lets the API caller verify which SKUs were queried during the turn.