🏛️ Class-Based Tools#

Class-Based Tools (Shared-Resource Pattern) solves a critical production engineering problem: how do multiple agent tools share state, database connection pools, or authenticated API clients without opening a brand-new connection on every single tool call or littering the codebase with global variables?

This demo encapsulates an inventory store (self.products) and two decorated methods — @tool def check_stock(self, product_id: str) and @tool def update_stock(self, product_id: str, quantity: int) — inside an InventoryTools class. Passing the bound methods tools=[inventory.check_stock, inventory.update_stock] to the Strands Agent lets a single request like "Update PROD-456 stock to 25 units, then confirm the new level" mutate state in one tool call and immediately verify the updated value in the next.

1. Why Standalone Tool Functions Break at Scale

If check_stock() and update_stock() are written as standalone functions that each call psycopg2.connect(...) inside their body, a 5-step agent turn opens and tears down 5 separate database connections — exhausting connection limits under concurrent traffic. Grouping related tools inside a class lets __init__() initialize a single connection pool, HTTP session, or state dictionary shared by every method.

2. Bound Methods and Automatic self Hiding

When you pass a bound method like inventory.check_stock to Agent(tools=[...]), Python binds the self parameter to the inventory instance, and Strands strips self from the generated JSON Schema. The LLM only sees product_id: str (and quantity: int for update_stock) — it never knows or cares that the tool is a method on a class.

3. Read-After-Write Verification in One Turn

Because both tools reference the exact same self.products dictionary, when the agent executes update_stock(product_id="PROD-456", quantity=25) on step 1 and check_stock(product_id="PROD-456") on step 2, the second call immediately sees 25 units. The response also returns a snapshot of current_inventory so the UI displays the live state after every run.

1. Core AI & Stateful Class-Based Tool Components in 10_class_based_tools.py

• Pydantic Contracts (ClassBasedToolsRequest & ClassBasedToolsResponse): Define the default multi-step prompt ("Update PROD-456 stock to 25 units, then confirm the new level") and the output schema (query: str, response: str, tool_calls: list[str], and current_inventory: dict[str, int]).
• Step ② — Model Selection (model_id): Resolves the target Bedrock model ID, defaulting to "amazon.nova-lite-v1:0".
• class InventoryTools (Shared-State Tool Container):
  • __init__(self): Initializes self.products with three SKUs (PROD-123 Wireless Mouse with 15 units at $29.99, PROD-456 USB-C Cable with 0 units at $12.99, and PROD-789 Laptop Stand with 8 units at $49.99), modeling a shared database connection pool or state store.
  • @tool def check_stock(self, product_id: str) -> str: Looks up product_id in self.products and returns either f"{product['name']}: {product['quantity']} units at ${product['price']}" or a not-found message.
  • @tool def update_stock(self, product_id: str, quantity: int) -> str: Mutates self.products[product_id]["quantity"] = quantity in place and returns f"Updated {product_id} to {quantity} units".
• Step ③ — Module-Level Singleton (inventory = InventoryTools()): Instantiates InventoryTools once so state mutations persist across tool invocations.
• run_class_based_tools_agent(query: str) -> dict:
  • Step ④: Creates Agent(model=BedrockModel(model_id=model_id), tools=[inventory.check_stock, inventory.update_stock]), passing bound instance methods so both tools operate on the same inventory object.
  • Step ⑤: Executes response = agent(query), allowing the agent to call update_stock and then check_stock to verify its own write.
  • Step ⑥: Extracts tool_calls from agent.messages and builds current_inventory = {pid: p["quantity"] for pid, p in inventory.products.items()}.

2. End-to-End AI Execution Walkthrough

1. run_class_based_tools_agent(query) constructs a Strands Agent with the bound methods [inventory.check_stock, inventory.update_stock]; Strands automatically excludes the bound self parameter from the Bedrock tool schemas.
2. On Converse Turn 1, for "Update PROD-456 stock to 25 units, then confirm the new level", Bedrock emits a toolUse block calling update_stock(product_id="PROD-456", quantity=25).
3. Strands invokes inventory.update_stock("PROD-456", 25), which mutates inventory.products["PROD-456"]["quantity"] from 0 to 25 and returns "Updated PROD-456 to 25 units" in a toolResult block.
4. On Converse Turn 2, to fulfill "then confirm the new level", Bedrock emits a second toolUse block calling check_stock(product_id="PROD-456"); because check_stock reads the exact same inventory.products dict, it returns "USB-C Cable: 25 units at $12.99".
5. On Converse Turn 3, Bedrock writes the final confirmation message, and Step ⑥ returns {"query": query, "response": str(response), "tool_calls": tool_calls, "current_inventory": ...}.

Class-Based Tools (Shared-Resource Pattern) Agent Service.

"""Class-Based Tools (Shared-Resource Pattern) Agent Service.

Demonstrates class-based tools that share state and resources across calls:
1. Architectural Problem: Individual tool functions opening separate DB connections flood databases.
2. The Solution: Group related tools in a class instance sharing single state/connection pool.
3. Class `InventoryTools`:
   - `check_stock(product_id: str)`
   - `update_stock(product_id: str, quantity: int)`
4. Handles multi-step state mutations within a single conversation session.
"""

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 ClassBasedToolsRequest(BaseModel):
    """Input payload for Class-Based Tools Agent."""

    query: str = "Update PROD-456 stock to 25 units, then confirm the new level"


class ClassBasedToolsResponse(BaseModel):
    """Output payload from Class-Based Tools Agent."""

    query: str
    response: str
    tool_calls: list[str]
    current_inventory: dict[str, int]


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


class InventoryTools:
    """Grouped inventory tools sharing a single state."""

    def __init__(self):
        # Initialize the shared in-memory product store (represents a single DB connection pool)
        self.products = {
            "PROD-123": {"name": "Wireless Mouse", "quantity": 15, "price": 29.99},
            "PROD-456": {"name": "USB-C Cable",    "quantity": 0,  "price": 12.99},
            "PROD-789": {"name": "Laptop Stand",   "quantity": 8,  "price": 49.99},
        }

    @tool
    def check_stock(self, product_id: str) -> str:
        """Check the current stock level and price for a product.

        Args:
            product_id: The product ID to look up (e.g. 'PROD-123')
        """
        # Look up the product record in the shared self.products dictionary
        product = self.products.get(product_id)
        if not product:
            return f"Product {product_id} not found"
        return f"{product['name']}: {product['quantity']} units at ${product['price']}"

    @tool
    def update_stock(self, product_id: str, quantity: int) -> str:
        """Update product stock quantity.

        Args:
            product_id: The product ID to update
            quantity: New quantity to set
        """
        # Mutate the product quantity in the shared self.products store
        if product_id in self.products:
            self.products[product_id]["quantity"] = quantity
            return f"Updated {product_id} to {quantity} units"
        return f"Product {product_id} not found"


# ③ Instantiate InventoryTools once so stock updates persist across requests
inventory = InventoryTools()


def run_class_based_tools_agent(
    query: str = "Update PROD-456 stock to 25 units, then confirm the new level",
) -> dict:
    """Run the Class-Based Tools Agent."""
    # ④ Bind the shared instance's methods to the Strands Agent
    agent = Agent(
        model=BedrockModel(model_id=model_id),
        tools=[inventory.check_stock, inventory.update_stock],
    )

    # ⑤ Run the query against the shared InventoryTools instance
    response = agent(query)

    # ⑥ 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 {
        "query": query,
        "response": str(response),
        "tool_calls": tool_calls,
        "current_inventory": {pid: p["quantity"] for pid, p in inventory.products.items()},
    }

Flask Blueprint server and route handler (class_tools_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 object-oriented design patterns, Strands bound-method mechanics, and state management principles in the 🏛️ Class-Based Tools demo:

1. The Shared-Resource Architectural Pattern: Grouping related tools inside a class (like InventoryTools) allows all tool methods to share a single database connection pool, authenticated HTTP session, or state dictionary initialized once in __init__().
2. Avoiding Connection Exhaustion: Standalone tool functions that open a new database connection on every invocation quickly exhaust database connection limits during multi-turn agent loops; class-based tools reuse the shared resource across every call.
3. Decorating Class Methods with @tool: The Strands @tool decorator works seamlessly on class methods (def check_stock(self, product_id: str)), inspecting parameter type hints and docstrings just like standalone functions.
4. Passing Bound Methods to Agent(tools=[...]): Registering tools=[inventory.check_stock, inventory.update_stock] passes Python bound methods in which self is already fixed to the inventory instance.
5. Automatic self Parameter Stripping: When generating the Bedrock Converse JSON Schema for a bound method, Strands automatically omits self so the LLM only sees domain parameters (product_id, quantity).
6. Read-After-Write Consistency Across Tool Calls: Because update_stock and check_stock reference the same self.products dictionary, mutating PROD-456 to 25 on step 1 is immediately visible when check_stock("PROD-456") runs on step 2.
7. Module-Level Singleton Persistence: Instantiating inventory = InventoryTools() at module scope in 10_class_based_tools.py keeps stock updates alive across multiple HTTP requests for the lifetime of the server process.
8. Exposing State Snapshots (current_inventory): Returning current_inventory alongside the LLM's natural-language response lets the frontend UI render a deterministic live stock badge independent of how the LLM phrases its prose.