In a previous series of articles I have described the design and implementation of a sample ECommerce serverless full-stack solution on AWS at: https://dev.to/brycepc/building-the-better-store-an-agile-cloud-native-ecommerce-system-on-aws-part-1-introduction-to-27ii. The solution's source code is also available at: https://github.com/TheBetterStore.
This is the first of what is expected to be a series of new articles aimed at describing the continuous evolution of a new GenAI-based ChatBot in the BetterStore web UI to support product inventory and product review queries from users; first to illustrate basic agentic implementation concepts, followed by increasing its capabilities with addition of latest patterns, frameworks (e.g. Strands Agents SDK) and AWS tooling; such as from the Amazon Bedrock AgentCore suite of services. Note in contrast to previous articles which have described NodeJS/Typescript-based backend solutions, we'll be focusing on Python 3.12-based backend solutions here, due to it traditionally receiving greater support for GenAI-based libraries.
To help set the context, the chat facility that will be made available to end users is illustrated below, with a query example:
Figure 1: Illustrating a GenAI chat window added to the BetterStore web UI. In this example a user has submitted 2 queries; the first asking for a recommendation for a Windows PC in the store's inventory that is suitable for a high-school student, and a second asking for reasoning against its recommendation response.
Each submitted user query in the chat prompt is passed to a newly-implemented RESTful API endpoint ('/query'), which uses a Large Language Model (LLM) inside an 'Agentic Loop', to interpret user requests, invoke 'tools' (supporting API's) to obtain supporting information, and generate a useful response.
Figure 2: Illustrating AWS services used for the first implementation; specifically the Web UI submits queries to the backend API which is hosted with a lambda-backed API Gateway. The lambda implements logic within an Agentic Loop, calling an LLM (Anthropic Claude Sonnet 4.5) to interpret the query, and invoking a set of tools (RESTful API's) to obtain supporting information.
This article will focus on describing a basic implementation of the 'Agentic Loop' pattern, whereby an LLM (Anthropic Claude Sonnet 4.5 in this case) is used to interpret user queries, and then in a repetitive loop, query API's or functions ('tools') that are registered with it to obtain needed information, until the LLM has determined that it has sufficient information to answer the query.
The second part of this article then introduces a simplified approach to implementing the Agentic Loop using Strands Agents SDK, alongside additional features that will be discussed in further detail in subsequent articles.
Part 1: Agentic Loop Implementation
An "agentic loop" may be described as a cycle where a GenAI LLM autonomously repeats the following steps to obtain a result to a user query:
- Gather context: the LLM reads the current state; e.g. a user request.
- Take action: it decides on a next step and calls a "tool" it knows about which it deems may help to provide an answer (e.g. calling an API, invoking a command or searching the web).
- Verify the result: It observes the response of the tool's action (e.g. what did an API or web query return?)
- Repeat or stop: Based on the result, it either loops back with an updated context to perform another action, or it judges that it has sufficient information to answer the initial query, and returns this as a final response.
The following diagram describes the envisaged Agentic Loop pattern and components for our initial solution:
Figure 3: Agentic Loop pattern and components for the initial solution.
Components used within the Agentic Loop include:
A. A Large Language Model (LLM) - Anthropic Claude Sonnet 4.5. This is invoked by Amazon Bedrock's Converse API, which provides an agnostic interface to a range of LLMs. This is initially instantiated with a System Prompt to set a context for the LLM, instructing how to respond to user queries.
e.g.
'You are a helpful shopping assistant for The Better Store, an electronics and books retailer. Keep responses brief. Use the lookup_inventory tool to check product availability when asked about products.'
B. A set of "tools" for the LLM to use to obtain information for resolving user queries; specifically these are:
i. Products API (The BetterStore); which defines items in The Better Store's inventory, which can be filtered by product category (BOOKS or COMPUTERS).
ii. OpenLibrary API (Public); provides free authoritative structured metadata about books; e.g. author, ISBN, publication year, subject tags, page count, and community ratings, to provide a factual overview to an AI before using GenAI to add opinions and sentiment.
iii. Brave Search API (Public); retrieves live web content including user and expert reviews, benchmarks, forum discussions, and user sentiment, to provide current opinion about a product that is unavailable within static databases.
Each of these tools is described within a JSON schema and configured against the Converse API, to tell the LLM what information it can use to answer a user query:
TOOL_TIMEOUT_S = 5
TOOLS = [
{
"toolSpec": {
"name": "lookup_inventory",
"description": (
"Search The Better Store product catalog. Returns products with "
"name, category, price, description, and details."
),
"inputSchema": {
"json": {
"type": "object",
"properties": {
"productName": {
"type": "string",
"description": "Optional product name to filter by",
},
"brandId": {
"type": "string",
"description": (
"Optional brand name to filter by (e.g. ASUS, Dell, Apple, "
"Lenovo, HP, Samsung, Microsoft, Acer, Razer, MSI, "
"Penguin Books, HarperCollins, Bloomsbury)"
),
},
"category": {
"type": "string",
"description": "Optional product category to filter by",
"enum": ["BOOKS", "COMPUTERS", "MOBILE"],
},
},
}
},
}
},
{
"toolSpec": {
"name": "get_computer_reviews",
"description": (
"Search for reviews and opinions about a specific computer or laptop. "
"Use when the user asks about reviews for a computer product."
),
"inputSchema": {
"json": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "Computer or laptop name/model to search reviews for",
}
},
"required": ["query"],
}
},
}
},
{
"toolSpec": {
"name": "get_book_info",
"description": (
"Search for book information including ratings and reviews from Open Library. "
"Use when the user asks about a book."
),
"inputSchema": {
"json": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "Book title or author to search for",
}
},
"required": ["query"],
}
},
}
},
]
An example flow of events for a query is as follows:
- A user submits their query in The Better Store's chat window; for example: "What Windows PC do you have which is best for a high school student?"; which posts this to the lambda-backed Chat API endpoint.
-
The invoked lambda initialises the "toolSpec" JSON definition and system prompt as above, then invokes Amazon Bedrock's Converse API, passing the user's prompt as messages, as per the following:
@property def _converse_params(self) -> dict: return { "modelId": self._bedrock_model, "system": [{"text": self._system_prompt}], "toolConfig": {"tools": TOOLS}, } def query(self, messages: list[dict]) -> dict: """Synchronous agentic loop - call Bedrock Converse, handle tool use.""" logger.info("querying against model: %s", self._bedrock_model) def converse(): return self._client.converse( **self._converse_params, messages=messages, ) response = converse() iterations = 0 while response.get("stopReason") == "tool_use" and iterations < 5: iterations += 1 assistant_msg = response["output"]["message"] messages.append(assistant_msg) logger.debug("assistant msg: %s", json.dumps(assistant_msg)) tool_results = self._execute_tool_calls(assistant_msg.get("content", [])) messages.append({"role": "user", "content": tool_results}) response = converse() logger.info("response: %s", json.dumps(response, default=str)) return responseA key processing function is execute_tool_calls, which is defined as:
def _execute_tool_calls(self, content_blocks: list[dict]) -> list[dict]: """Execute tool calls in parallel and return tool results.""" tool_blocks = [b for b in content_blocks if "toolUse" in b] def execute_single(block: dict) -> dict: tool_use = block["toolUse"] tool_input = tool_use.get("input", {}) name = tool_use["name"] if name == "get_book_info": result = self._get_book_info(tool_input["query"]) elif name == "get_computer_reviews": result = self._get_computer_reviews(tool_input["query"]) else: result = { "products": lookup_inventory( product_name=tool_input.get("productName"), brand_id=tool_input.get("brandId"), category=tool_input.get("category"), ) } return { "toolResult": { "toolUseId": tool_use["toolUseId"], "content": [{"json": result}], } } # Execute in parallel using threads results = [] with ThreadPoolExecutor(max_workers=len(tool_blocks)) as executor: futures = {executor.submit(execute_single, b): i for i, b in enumerate(tool_blocks)} # Collect in original order ordered = [None] * len(tool_blocks) for future in as_completed(futures): idx = futures[future] ordered[idx] = future.result() results = ordered return resultsFunctions for calling 2 other tools are defined as below:
def _get_computer_reviews(self, query: str) -> dict: """Search for computer reviews via Brave Search API.""" try: search_query = quote_plus(query + " review") url = f"https://api.search.brave.com/res/v1/web/search?q={search_query}&count=5" req = Request(url, headers={ "Accept": "application/json", "Accept-Encoding": "gzip", "X-Subscription-Token": self._brave_api_key, }) with urlopen(req, timeout=TOOL_TIMEOUT_S) as resp: if resp.status != 200: return { "error": f"Review service unavailable (HTTP {resp.status}). Unable to fetch reviews at this time."} data = json.loads(resp.read().decode()) return { "results": [ { "title": r.get("title"), "url": r.get("url"), "description": r.get("description"), } for r in data.get("web", {}).get("results", []) ] } except Exception as err: logger.error("getComputerReviews failed: %s", err) return {"error": "Review search timed out or failed. Please try again."} def _get_book_info(self, query: str) -> dict: """Search for book info via Open Library API.""" try: search_query = quote_plus(query) url = ( f"https://openlibrary.org/search.json?q={search_query}" "&limit=3&fields=title,author_name,first_publish_year,ratings_average,ratings_count,subject" ) req = Request(url) with urlopen(req, timeout=TOOL_TIMEOUT_S) as resp: data = json.loads(resp.read().decode()) return { "results": [ { "title": doc.get("title"), "authors": doc.get("author_name"), "year": doc.get("first_publish_year"), "avgRating": doc.get("ratings_average"), "ratingsCount": doc.get("ratings_count"), "subjects": (doc.get("subject") or [])[:5], } for doc in data.get("docs", []) ] } except Exception as err: logger.error("getBookInfo failed: %s", err) return {"error": "Book search timed out or failed. Please try again."} -
The LLM interprets the user's query and maps it to the category enum (e.g. COMPUTERS in this case). It responds with stopReason: tool_use and a content block like:
{ "toolUse": { "name": "lookup_inventory", "toolUseId": "abc123", "input": { "category": "COMPUTERS" } } }No productName or brandId is included as the user did not ask for a specific title or publisher.
The products endpoint is invoked with the category as a query parameter:
i. Products are retrieved using GET /products?category=COMPUTERS
ii. Filtering is performed server-side; the inventory service passes through the parameters provided by the
LLM (productName, brandId, category). Results are cached locally for 30 seconds to avoid repeated calls.-
The filtered results are returned to Bedrock as a tool result which is appended to the conversation:
{ "toolResult": { "toolUseId": "abc123", "content": [{ "json": { "products": [ /* all computers */ ] } }] } }
The while response.get("stopReason") == 'tool_use' loop calls converse() again with the updated messages.
- Bedrock formulates the final answer when it has the full list of computers. It inspects the product data (descriptions, specifications, price — whatever fields exist in the product objects) to determine which computer best fits the user's requirements, and returns a natural language answer.
Key considerations
- The model's ability to make recommendations depends entirely on what fields the products API returns. If the product objects include detailed specifications, descriptions, or ratings, the model can reason over them effectively. If they don't, the model may either admit it lacks sufficient data, or rely on its general knowledge.
- Each tool call is constrained by a 5-second timeout (
TOOL_TIMEOUT_S = 5) to prevent the loop from hanging on unresponsive external services. - Tool functions return structured error objects (e.g.
{ error: 'Review search timed out or failed.' }) rather than throwing exceptions, allowing the LLM to gracefully inform the user when a service is unavailable. - The full conversation history (messages array) is passed with each
converse()call, growing with each iteration. This is what enables the LLM to maintain context across tool calls and reason over accumulated information. - If the 5-iteration cap (
iterations < 5) is reached whilestopReasonis stilltool_use, the loop exits with a response whose content is still a tool-use request rather than a final natural-language answer. This basic implementation doesn't handle that case explicitly — production code should detect it and return a fallback message (or raise an error) rather than passing an incomplete response back to the caller.
The following sequence diagram summarizes this end-to-end flow:
Figure 4: Sequence diagram illustrating events in an agentic loop to resolve a user's product query in The Better Store.
Part 2: Agentic Loop Simplification - Introduction to Strands SDK
In 2025, AWS created an open-source framework called Strands Agents SDK, after it emerged that internal Amazon teams were having to write the same boilerplate code for their Bedrock-based agentic applications to accommodate the agentic loop, tool dispatch, message management and error handling. The Strands SDK was designed to standardise and implement the common patterns and best practices for simpler implementation of agentic applications, by working at the following 3 layers:
- Upon configuration of a system prompt, tools, and LLM model for a new agent, the agent loop is implicitly-provided rather than being coded. The SDK internally runs the converse→check stopReason→execute tools→append results→reconverse cycle until the model signals it's done. You never write while or check stopReason.
e.g.
```python
class ChatService:
"""Uses Strands Agents SDK to manage the agentic loop."""
# Agent attributes are initialised e.g. from properties passed by Lambda variables
def __init__(
self,
bedrock_model: str,
max_tokens: str,
system_prompt: str,
brave_api_key: str,
):
self._bedrock_model_id = bedrock_model
self._max_tokens = int(max_tokens) if max_tokens else 1024
self._system_prompt = system_prompt
# Create the Brave reviews tool with the API key bound
get_computer_reviews = _create_get_computer_reviews(brave_api_key)
# Configure the Bedrock model provider
model = BedrockModel(
model_id=bedrock_model,
region_name="ap-southeast-2",
max_tokens=self._max_tokens,
)
# Create the Strands agent — handles the agentic loop internally
self._agent = Agent(
model=model,
system_prompt=system_prompt,
tools=[lookup_inventory_tool, get_computer_reviews, get_book_info],
)
def query(self, messages: list[dict]) -> dict:
"""Invoke the Strands agent with conversation messages."""
logger.info("querying against model: %s", self._bedrock_model_id)
# Extract the current user turn, then seed the agent with everything
# BEFORE it. Strands appends whatever text is passed into __call__
# as the new user message, so leaving that same turn in the history
# we assign to self._agent.messages would duplicate it.
last_user_text = self._extract_last_user_text(messages)
self._agent.messages = messages[:-1]
result = self._agent(last_user_text)
logger.info("response stop_reason: %s", result.stop_reason)
# Return the response in a format compatible with the existing API contract
return {
"output": {
"message": {
"role": "assistant",
"content": [{"text": result.message["content"][-1]["text"] if result.message.get("content") else str(result)}],
}
},
"stopReason": result.stop_reason,
"metrics": {},
}
@staticmethod
def _extract_last_user_text(messages: list[dict]) -> str:
"""Extract the text from the last user message."""
for msg in reversed(messages):
if msg.get("role") == "user":
content = msg.get("content", [])
if isinstance(content, str):
return content
if isinstance(content, list):
for block in content:
if isinstance(block, dict) and "text" in block:
return block["text"]
if isinstance(block, str):
return block
return ""
```
-
Tools are handled as functions; rather than defining JSON schemas for each, a function is simply annotated with @tool and associated attributes such as request and response definitions, plus purpose.
def _create_get_computer_reviews(brave_api_key: str): """Factory to create the get_computer_reviews tool with the API key bound.""" @tool def get_computer_reviews(query: str) -> dict: """Search for reviews and opinions about a specific computer or laptop. Use when the user asks about reviews for a computer product. Args: query: Computer or laptop name/model to search reviews for """ try: search_query = quote_plus(query + " review") url = f"https://api.search.brave.com/res/v1/web/search?q={search_query}&count=5" req = Request(url, headers={ "Accept": "application/json", "Accept-Encoding": "gzip", "X-Subscription-Token": brave_api_key, }) with urlopen(req, timeout=TOOL_TIMEOUT_S) as resp: if resp.status != 200: return {"error": f"Review service unavailable (HTTP {resp.status}). Unable to fetch reviews at this time."} data = json.loads(resp.read().decode()) return { "results": [ { "title": r.get("title"), "url": r.get("url"), "description": r.get("description"), } for r in data.get("web", {}).get("results", []) ] } except Exception as err: logger.error("get_computer_reviews failed: %s", err) return {"error": "Review search timed out or failed. Please try again."} return get_computer_reviews @tool def get_book_info(query: str) -> dict: """Search for book information including ratings and reviews from Open Library. Use when the user asks about a book. Args: query: Book title or author to search for """ try: search_query = quote_plus(query) url = ( f"https://openlibrary.org/search.json?q={search_query}" "&limit=3&fields=title,author_name,first_publish_year,ratings_average,ratings_count,subject" ) req = Request(url) with urlopen(req, timeout=TOOL_TIMEOUT_S) as resp: data = json.loads(resp.read().decode()) return { "results": [ { "title": doc.get("title"), "authors": doc.get("author_name"), "year": doc.get("first_publish_year"), "avgRating": doc.get("ratings_average"), "ratingsCount": doc.get("ratings_count"), "subjects": (doc.get("subject") or [])[:5], } for doc in data.get("docs", []) ] } except Exception as err: logger.error("get_book_info failed: %s", err) return {"error": "Book search timed out or failed. Please try again."} -
Conversation state including the messages array is automatically managed. In the manual implementation, the developer must explicitly append each assistant response and tool result to maintain conversation context:
# BEFORE: Manual message management (Part 1) while response.get("stopReason") == "tool_use" and iterations < 5: iterations += 1 assistant_msg = response["output"]["message"] messages.append(assistant_msg) # developer appends assistant msg tool_results = self._execute_tool_calls(assistant_msg.get("content", [])) messages.append({"role": "user", "content": tool_results}) # developer appends tool results response = converse()
With Strands, this is entirely eliminated. The SDK manages the messages array internally — appending assistant responses, tool invocations, and tool results automatically across each loop iteration:
```python
# AFTER: Strands manages conversation state
def query(self, messages: list[dict]) -> dict:
last_user_text = self._extract_last_user_text(messages)
self._agent.messages = messages[:-1] # hand prior history to the agent
result = self._agent(last_user_text) # SDK appends this turn & manages the rest
return result
```
The developer never constructs toolResult blocks, correlates toolUseId values, or decides where to insert messages in the array. The SDK owns the full conversation lifecycle — the one thing the caller still needs to get right is not leaving the current user turn in both places, since the SDK appends it exactly once when the agent is invoked.
To summarise, the following table shows the key differences between the manual agentic loop implementation, and that provided by Strands SDK:
Figure 5: Differences between the manual agentic loop implementation and that provided by Strands SDK.
Next Steps
In the next article I will describe rehosting this from API Gateway/Lambda to the newer Amazon Bedrock AgentCore service, and explore the benefits that it and associated services have to offer.
References:
- What Is an Agentic Loop? The Core Pattern Behind Autonomous AI Agents
- Strands Agents SDK
- The Better Store Chat API - AgenticLoop
- The Better Store Chat API - StrandsSDK
Disclaimer: The views and opinions expressed in this article are those of the author only.
Top comments (0)