🔌 Function to Tool#

Function to Tool (Website Health Checker) demonstrates how a standard Python function that performs real network I/O — checking whether a live website is reachable via requests.get(url, timeout=5) — becomes an autonomous agent capability with a single @tool decorator in AWS Strands.

When you ask "Is https://www.google.com reachable right now?", the Bedrock model reads the schema Strands generated from check_website(url: str) -> str, extracts the URL, triggers the live HTTP request from the server, inspects the returned status string (e.g., "https://www.google.com is up (status 200)"), and reports the result.

1. The Three Pillars of a Strands @tool Function

When you decorate a function with @tool, Strands compiles a Bedrock Converse toolSpec from three parts of your Python definition:

1. Decorator (@tool): Wraps the function in a metadata descriptor that the Agent can register and invoke.
2. Type Hints (url: str → str): Defines the JSON Schema input types so the model knows to pass a string URL.
3. Docstring: Becomes the semantic prompt that tells the model when to call the tool and what the url parameter represents.

2. Docstrings Are Prompts (The "Does a Thing" Experiment)

The LLM never sees the body of check_website() (it cannot see that your code calls requests.get). It only sees the function name, parameter types, and docstring. If you replace the docstring with """Does a thing.""", the model loses confidence in what the tool does and frequently refuses to call it or passes malformed arguments.

3. Defensive Error Handling Inside Tools

Real-world network calls fail (DNS errors, connection timeouts, SSL errors). Wrapping requests.get(url, timeout=5) in try/except requests.RequestException as e: and returning a readable string (f"{url} is down ({e})") allows the agent to report the outage gracefully instead of crashing the Flask worker with an unhandled exception.

4. Auditing Tool Invocations via agent.messages

After agent(prompt) completes, iterating over agent.messages and extracting every content block containing a "toolUse" key lets the backend surface the exact function call and arguments (check_website({'url': ...})) in the UI.

1. Core AI & Tool Components in 03_function_to_tool.py

• Pydantic Contracts (FunctionToToolRequest & FunctionToToolResponse): Define the input question: str (defaulting to "Is the staging server running? Check https://httpbin.org/get") 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_server_status(server_url: str) -> str: Demonstrates the three building blocks that Strands inspects to generate a Bedrock Converse toolSpec:
  1. Decorator (@tool): Attaches tool specification metadata to the Python function so Strands can register and invoke it.
  2. Type Hints (server_url: str → str): Tells Strands that server_url is a required string property in the generated JSON Schema.
  3. Docstring (Args: & Returns:): Acts as the instruction manual the LLM reads to decide when to call the tool and what argument to pass. Inside the function body, a try/except requests.exceptions.RequestException block performs requests.get(server_url, timeout=5) and returns either f"Server is up. Status code: {response.status_code}" or "Server is down or unreachable" so network errors are returned cleanly as tool observations rather than crashing the agent loop.
• run_function_to_tool_agent(question: str) -> dict:
  • Step ③: Creates a fresh Agent(model=BedrockModel(model_id=model_id), tools=[check_server_status]).
  • Step ④: Calls response = agent(question), executing the Converse tool-use cycle.
  • Step ⑤: Iterates over every message in agent.messages and every content block in message["content"], formatting any "toolUse" block as f"{block['toolUse']['name']}({block['toolUse']['input']})" and returning {"question": question, "response": str(response), "tool_calls": tool_calls}.

2. End-to-End AI Execution Walkthrough

1. run_function_to_tool_agent(question) instantiates a Strands Agent with tools=[check_server_status]. Strands introspects the function's signature and docstring to build the Bedrock toolSpec.
2. On Turn 1, Bedrock Nova Lite reads the user question (e.g., "Is the staging server running? Check https://httpbin.org/get") and the tool spec, extracts "https://httpbin.org/get", and returns an assistant message containing a toolUse block for check_server_status.
3. Strands intercepts the toolUse block, calls check_server_status(server_url="https://httpbin.org/get") in Python, receives "Server is up. Status code: 200" from requests.get(), and sends a toolResult block back to Bedrock.
4. On Turn 2, Bedrock reads the status code result and generates a natural-language confirmation.
5. Step ⑤ extracts the toolUse record from agent.messages and returns {"question": question, "response": str(response), "tool_calls": tool_calls}.

Function-to-Tool Agent Service.

"""Function-to-Tool Agent Service.

Demonstrates converting a plain Python function into an LLM Agent Tool:
1. Decorator: `@tool` exposes the function to the agent loop.
2. Type hints: inform the agent of expected parameter and return types.
3. Docstring: guides the agent on WHEN and HOW to call the tool.
4. Tool: `check_server_status(server_url: str)` making live HTTP checks.
"""

import os

import requests
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 FunctionToToolRequest(BaseModel):
    """Input payload for Function-to-Tool Agent."""

    question: str = "Is the staging server running? Check https://httpbin.org/get"


class FunctionToToolResponse(BaseModel):
    """Output payload from Function-to-Tool 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                                              # 1. decorator: exposes function to the agent
def check_server_status(server_url: str) -> str:   # 2. type hints: defines parameter & return types
    """Check if a server is responding by making an HTTP request.

    Args:
        server_url: The URL of the server to check

    Returns:
        A message indicating whether the server is up or down
    """                                            # 3. docstring: guides the LLM on when to call it
    try:
        # Send an HTTP GET request with a 5-second timeout and report the status code
        response = requests.get(server_url, timeout=5)
        return f"Server is up. Status code: {response.status_code}"
    except requests.exceptions.RequestException:
        # Catch network or HTTP errors and return a readable diagnostic message
        return "Server is down or unreachable"


def run_function_to_tool_agent(question: str = "Is the staging server running? Check https://httpbin.org/get") -> dict:
    """Execute the Function-to-Tool agent."""
    # ③ Initialize the Strands Agent with BedrockModel and the check_server_status tool
    agent = Agent(model=BedrockModel(model_id=model_id), tools=[check_server_status])

    # ④ Run the agent on the user 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}


# TRY THIS EXPERIMENT:
# Change the docstring above to just  """Does a thing."""  and re-run.
# The agent will often STOP calling the tool.
# Why? The docstring is not a comment — it is the manual the MODEL reads
# to decide whether your tool is relevant. Vague docstring = unreliable agent.

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

Keep these essential principles in mind when converting Python functions into AWS Strands agent tools:

1. The Three Pillars of @tool: Strands builds a Bedrock Converse toolSpec from (1) the @tool decorator, (2) Python type annotations (server_url: str -> str), and (3) the function docstring.
2. Docstrings Are Model Prompts: The LLM never sees your function's Python implementation; it relies entirely on the docstring and parameter descriptions to decide when and how to invoke the tool. Vague docstrings (like "Does a thing.") cause tool-calling failures.
3. LLMs Do Not Execute Code Directly: When Bedrock decides to use check_server_status, it only emits a structured JSON toolUse request. The Strands Agent runtime executes the actual Python function locally and sends the output back in a toolResult block.
4. Always Set Timeouts on External I/O: Calling requests.get(server_url, timeout=5) prevents an unresponsive upstream server from hanging the agent loop and blocking the Flask worker indefinitely.
5. Return Diagnostic Strings Instead of Raising Exceptions: Catching requests.exceptions.RequestException and returning "Server is down or unreachable" gives the LLM a clear observation it can report to the user rather than crashing the request.
6. Automatic URL Extraction from Natural Language: Even when the URL is embedded inside a conversational question ("Is the staging server running? Check https://httpbin.org/get"), the model isolates the exact URL string to pass as server_url.
7. Auditing Tool Calls via agent.messages: Scanning content blocks in agent.messages for "toolUse" keys provides a complete, ordered audit trail of every tool name and input dictionary the agent invoked.
8. Per-Request Agent Isolation: Instantiating Agent(model=..., tools=[check_server_status]) inside run_function_to_tool_agent() guarantees that each API call has an isolated message history.