⚡ Async Parallel Tools#

Async Tools (Parallel Warehouse Check) tackles the #1 UX bottleneck in production agent systems: sequential I/O latency. If an agent needs to check stock across three regional warehouses (east, west, and central) and each network call takes 2 seconds, running them one after another blocks the user for 3 × 2 s = 6 s on tool I/O alone.

By declaring the tool as a Python coroutine — @tool async def check_warehouse_inventory(product_id: str, warehouse: str) using non-blocking await asyncio.sleep(2) — and invoking the Strands agent with await agent.invoke_async(query), all three warehouse lookups execute concurrently on Python's asyncio event loop, completing the entire tool fan-out in ~2 seconds instead of ~6 seconds.

1. Single-Turn Multi-Tool Fan-Out

When asked "Can we ship 100 units of PROD-123? Check all warehouses: east, west, and central", the Bedrock model recognizes that the three lookups do not depend on one another. Instead of emitting one tool call and waiting, it emits all three toolUse blocks in a single assistant turn:

• check_warehouse_inventory(product_id="PROD-123", warehouse="east") → 45
• check_warehouse_inventory(product_id="PROD-123", warehouse="west") → 30
• check_warehouse_inventory(product_id="PROD-123", warehouse="central") → 60

2. Blocking time.sleep vs. Non-Blocking asyncio.sleep

A synchronous time.sleep(2) (or blocking requests.get) freezes the OS thread, forcing calls to run serially. Declaring async def and using await asyncio.sleep(2) (or an async HTTP/DB driver like httpx.AsyncClient / asyncpg) yields control back to the event loop while waiting for network I/O, so all three 2-second waits overlap in wall-clock time: Ttotal ≈ max(t1, t2, t3) = 2 s rather than t1 + t2 + t3 = 6 s.

3. Bridging Async Agents into Flask Routes

The Flask route handler in app.py bridges synchronous HTTP request handling with the asynchronous Strands runtime via asyncio.run(run_async_tools_agent(query)), running the agent's invoke_async coroutine on an event loop and returning parallel_duration_seconds so you can verify the concurrent speedup empirically.

1. Core AI & Async Parallel Tool Components in 11_async_tools.py

• Pydantic Contracts (AsyncToolsRequest & AsyncToolsResponse): Define the default fulfillment query ("Can we ship 100 units of PROD-123? Check all warehouses: east, west, and central.") and the response schema including parallel_duration_seconds: float and tool_calls: list[str].
• Step ② — Model Selection (model_id): Resolves the target Bedrock model ID, defaulting to "amazon.nova-lite-v1:0".
• Async Tool — @tool async def check_warehouse_inventory(product_id: str, warehouse: str) -> dict:
  • Non-Blocking Delay (await asyncio.sleep(2)): Simulates a 2-second remote warehouse API or database query while yielding control back to Python's asyncio event loop so other warehouse lookups can execute concurrently.
  • Warehouse Catalog Lookup: Looks up warehouse ("east": 45 units of PROD-123, "west": 30 units, "central": 60 units) and returns {"warehouse": warehouse, "product_id": product_id, "quantity": quantity}.
• async def run_async_tools_agent(query: str) -> dict:
  • Step ③: Initializes Agent(model=BedrockModel(model_id=model_id), tools=[check_warehouse_inventory]).
  • Step ④: Records start = time.time(), awaits response = await agent.invoke_async(query) so all three warehouse tool calls emitted by Bedrock execute in parallel on the event loop, and computes elapsed = time.time() - start.
  • Step ⑤: Collects all "toolUse" blocks from agent.messages and returns query, response, parallel_duration_seconds: round(elapsed, 1), and tool_calls.

2. End-to-End AI Execution Walkthrough

1. run_async_tools_agent(query) creates a Strands Agent with tools=[check_warehouse_inventory], records the start timestamp, and awaits agent.invoke_async(query).
2. On Converse Turn 1, Bedrock sees that the east, west, and central lookups are independent and emits three toolUse blocks in a single assistant message.
3. Strands schedules all three check_warehouse_inventory coroutines concurrently on the event loop; all three enter await asyncio.sleep(2) simultaneously and finish together after ~2 seconds (instead of 3 × 2 s = 6 s sequentially), returning 45, 30, and 60.
4. On Converse Turn 2, Strands sends all three toolResult blocks back to Bedrock in one message; Bedrock sums 45 + 30 + 60 = 135 units and confirms that 100 units of PROD-123 can be shipped.
5. Step ⑤ measures the total wall-clock duration (parallel_duration_seconds), extracts the three tool_calls, and returns the dictionary.

Async Parallel Tools Agent Service.

"""Async Parallel Tools Agent Service.

Demonstrates parallel asynchronous tool execution:
1. Problem: Sequential tool invocations cause linear latency accumulation (3 x 2s = 6s).
2. Solution: An `async` tool lets the agent run all lookups concurrently (~2s total).
3. Tool: `check_warehouse_inventory(product_id: str, warehouse: str)` across `east`, `west`, `central`.
4. Measures and returns the total duration (sequential would be ~6s).
"""

import asyncio
import os
import time

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 AsyncToolsRequest(BaseModel):
    """Input payload for Async Tools Agent."""

    query: str = "Can we ship 100 units of PROD-123? Check all warehouses: east, west, and central."


class AsyncToolsResponse(BaseModel):
    """Output payload from Async Tools Agent."""

    query: str
    response: str
    parallel_duration_seconds: float
    tool_calls: list[str]


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


@tool
async def check_warehouse_inventory(product_id: str, warehouse: str) -> dict:
    """Check inventory at a specific warehouse.

    Args:
        product_id: Product ID to check
        warehouse: Warehouse identifier (e.g., "east", "west", "central")
    """
    # Simulate a 2-second network I/O delay without blocking the asyncio event loop
    await asyncio.sleep(2)

    # Look up stock count for the requested warehouse and product SKU
    data = {
        "east":    {"PROD-123": 45, "PROD-456": 12},
        "west":    {"PROD-123": 30, "PROD-456": 0},
        "central": {"PROD-123": 60, "PROD-456": 25},
    }

    quantity = data.get(warehouse, {}).get(product_id, 0)
    return {"warehouse": warehouse, "product_id": product_id, "quantity": quantity}


async def run_async_tools_agent(
    query: str = "Can we ship 100 units of PROD-123? Check all warehouses: east, west, and central.",
) -> dict:
    """Run the Async Tools Agent."""
    # ③ Initialize the Strands Agent with the async warehouse inventory tool
    agent = Agent(
        model=BedrockModel(model_id=model_id),
        tools=[check_warehouse_inventory],
    )

    # ④ Record start time and invoke the agent asynchronously so tool calls run in parallel
    start = time.time()
    response = await agent.invoke_async(query)
    elapsed = time.time() - start

    # ⑤ Collect the tool calls
    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),
        "parallel_duration_seconds": round(elapsed, 1),
        "tool_calls": tool_calls,
    }

Flask Blueprint server and route handler (async_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 concurrency concepts, async Strands primitives, and latency optimizations in the ⚡ Async Parallel Tools demo:

1. The Sequential I/O Latency Bottleneck: When an agent calls three independent network-bound tools that each take 2 seconds, running them synchronously one after another blocks the user for 2 + 2 + 2 = 6 seconds of pure I/O wait time.
2. Single-Turn Multi-Tool Fan-Out: Modern tool-calling LLMs on Amazon Bedrock can emit multiple independent toolUse blocks (for east, west, and central) inside a single assistant turn rather than taking three separate Converse round-trips.
3. Declaring Async Tools (@tool async def): Decorating a Python coroutine function with @tool tells Strands that the tool can be awaited concurrently on an asyncio event loop.
4. Non-Blocking I/O (await asyncio.sleep vs. time.sleep): Using await asyncio.sleep(2) (or async drivers like httpx.AsyncClient and aioboto3) yields control to the event loop during network waits, whereas blocking calls like time.sleep(2) or requests.get() freeze the entire thread and destroy parallelism.
5. Invoking the Agent Asynchronously (await agent.invoke_async(query)): Calling agent.invoke_async() runs the Strands agent loop as an async coroutine, allowing concurrent tool coroutines in the same turn to overlap in wall-clock time (Ttotal ≈ max(t1, t2, t3) ≈ 2 s).
6. Bridging Sync WSGI to Async Coroutines (asyncio.run): Because standard Flask routes are synchronous functions, async_tools_route() in app.py uses asyncio.run(s_async_tool.run_async_tools_agent(query)) to execute the async agent workflow cleanly.
7. Batching toolResult Blocks: Once all three concurrent warehouse coroutines resolve, Strands packages all three results into a single user message containing three toolResult blocks (matched by toolUseId) and sends them to Bedrock in one Converse call.
8. Empirical Latency Verification (parallel_duration_seconds): Measuring elapsed = time.time() - start around await agent.invoke_async(query) and returning parallel_duration_seconds proves that the three 2-second lookups plus two Bedrock LLM turns finish in ~3.5–4.5 seconds total rather than >7.5 seconds.