⚡ Warehouse Check

async parallel tools via Strands + Bedrock

Concept

Three warehouse lookups at ~2 seconds each would take ~6 seconds sequentially. By making the tool async and using agent.invoke_async(), the calls run in parallel, finishing in ~2 seconds instead of ~6. This is the async tools pattern from Module 2.

Theory & Concepts

The Warehouse Check demo illustrates how to drastically reduce latency when an agent needs to perform multiple tasks at once. In this demo, the agent needs to check inventory across "east", "west", and "central" warehouses. Because we added an artificial 2-second delay to the check_warehouse_inventory tool, querying them one by one would take about 6 seconds.

To solve this, we define the tool as an async function and use agent.invoke_async() in warehouse_agent.py. When Amazon Nova Pro recognizes it needs data from all three locations to answer the user's browser query, it emits three tool calls simultaneously.

The Strands framework executes all three asynchronous lookups in parallel. As a result, the entire operation takes roughly 2 seconds (the time of the longest single lookup) instead of 6. The results are merged, and the final response is rapidly returned to the user, highlighting how async parallel execution is vital for responsive agent applications.

Request flow

Browser question POST /warehouse
↓
warehouse_agent.py lookup() asyncio.run
↓
Agent.invoke_async Bedrock
↓
check_warehouse_inventory called for east, west, central in parallel (all ~2s) answers merged
↓
Browser JSON result

Code flow

flowchart TD A["Browser<br/>warehouse question"] -->|"POST /warehouse"| B["app.py<br/>warehouse_route"] B -->|"question"| C["warehouse_agent.py<br/>lookup"] C -->|"asyncio.run"| D["Agent<br/>invoke_async"] D -->|"question + tool"| E["Bedrock<br/>Nova Pro"] E -->|"parallel tool calls"| F["check_warehouse<br/>east"] E -->|"parallel tool calls"| G["check_warehouse<br/>west"] E -->|"parallel tool calls"| H["check_warehouse<br/>central"] F -->|"qty"| E G -->|"qty"| E H -->|"qty"| E E -->|"merged answer"| D D -->|"answer text"| C C -->|"answer text"| B B -->|"JSON result"| A

Backend

warehouse_agent.py 
"""Module 2 demo — async tools (parallel execution).

Three warehouse lookups at ~2 seconds each would take ~6 seconds one by one.
By making the tool ``async`` and using ``agent.invoke_async()``, the calls run
in PARALLEL, so the whole thing takes about 2 seconds instead.
"""

import asyncio

from strands import Agent, tool
from strands.models.bedrock import BedrockModel

from config import MODEL_ID, agent_text


@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 an API call delay so parallelism is observable.
    await asyncio.sleep(2)
    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}


def lookup(question: str) -> str:
    """Answer a multi-warehouse question, running the lookups in parallel."""

    async def run():
        agent = Agent(
            model=BedrockModel(model_id=MODEL_ID),
            tools=[check_warehouse_inventory],
            callback_handler=None,
        )
        return await agent.invoke_async(question)

    return agent_text(asyncio.run(run()))