Build an AI Agent with Tool Use
Build an autonomous AI agent on Amazon Bedrock with tool use and S3 audit logging.
Introduction
30 Second Summary
When you contact a company's support team, the best agents don't just read from a script. They check your order status, look up inventory, and escalate issues on the spot. Now imagine teaching an AI to do that same decision-making autonomously.
In this project, you will build an AI agent on Amazon Bedrock that receives a customer inquiry, decides which tools to call, executes them, and loops until it has a complete answer.
What You'll Build
An enterprise support AI agent powered by the Bedrock Converse API that autonomously selects and executes tools to handle business inquiries about orders, inventory, and issue escalation.
By the end of this project, you'll have:
- A Python script implementing the full agent loop pattern (observe-think-act) that calls Amazon Bedrock and loops until the task is complete.
- Three simulated enterprise tools (order lookup, inventory check, issue escalation) defined with JSON Schema and wired into the Converse API.
- An S3-based audit trail that logs every agent decision and tool call for enterprise compliance.
- Secret Mission: Add a cost-tracking tool that monitors token usage and alerts when spending exceeds a threshold.
Are there any prerequisites?
You need an AWS account (paid plan), Python 3.14.2 installed on your machine, and a code editor (such as VS Code). Step 1 walks you through installing the AWS CLI and enabling Bedrock model access.
Set Up AWS CLI, boto3, and Bedrock Access
To build an AI agent that talks to Amazon Bedrock, your local machine needs a secure connection to AWS. That connection requires two tools: the AWS CLI (which handles authentication) and boto3 (the Python library that sends API requests).
In this step, you will install both tools and enable access to the AI model your agent will use. Once everything is wired up, your Python script will be able to call Bedrock directly from your machine.
In this step, get ready to:
- Install and configure the AWS CLI with your credentials.
- Install boto3 so Python can talk to AWS services.
- Enable Amazon Nova Micro model access in the Bedrock console.
Install and configure the AWS CLI
The AWS CLI is a command-line tool that lets you interact with AWS services from your terminal. Your agent script uses it behind the scenes to authenticate every API request to Bedrock.
- Open a command prompt by pressing the Windows key, typing cmd, and pressing Enter.
- Check whether the AWS CLI is already installed by running this command:
aws --version✔️ I see aws-cli/2.x.x
Your output starts with aws-cli/2. followed by a version number. The AWS CLI v2 is already installed and you can skip ahead to the configuration step below.
ⓧ I see aws-cli/1.x.x
Your output starts with aws-cli/1. which means you have the older version installed. You need to remove it first and then install version 2.
- Open Programs and Features by pressing the Windows key and typing appwiz.cpl, then pressing Enter.
- Select AWS Command Line Interface and choose Uninstall.
- Confirm the uninstall and close the window.
- Close your command prompt and reopen it.
- Download and run the AWS CLI v2 MSI installer from this link:
https://awscli.amazonaws.com/AWSCLIV2.msi
- Follow the installer prompts, accepting the defaults.
- Close and reopen your command prompt after installation finishes.
- Verify the install by running:
aws --versionYou should see output starting with aws-cli/2. followed by a version number.
ⓧ Command not found
The AWS CLI is not installed on your machine yet. Let's install it now.
- Download the AWS CLI v2 MSI installer from this link:
https://awscli.amazonaws.com/AWSCLIV2.msi
- Run the downloaded AWSCLIV2.msi file and follow the installer prompts, accepting the defaults.
- Close and reopen your command prompt after installation finishes.
- Verify the install by running:
aws --versionYou should see output starting with aws-cli/2. followed by a version number.
On macOS or Linux?
This project shows Windows commands. If you are on macOS or Linux, follow the official install guide at docs.aws.amazon.com/cli/latest/userguide/getting-started-install.html to install the CLI for your platform.
Where do I get my Access Key ID and Secret Access Key?
You create access keys in the IAM console. Sign in to the AWS Management Console, navigate to IAM, choose Users, select your user, then open the Security credentials tab.
In the Access keys section, choose Create access key. Select Other as the use case, then Next. Click Create access key and download the .csv file immediately. This is the only time you can view the secret key.
- Configure the AWS CLI with your credentials by running:
aws configureWhat will aws configure ask me?
The command prompts you for four values, one at a time. Enter the following:
- AWS Access Key ID: paste the Access Key ID from your downloaded .csv file.
- AWS Secret Access Key: paste the Secret Access Key from the same file.
- Default region name: type us-east-1.
- Default output format: type json.
- Confirm your configuration worked by running:
aws sts get-caller-identityYou should see a JSON response containing your account ID, user ARN, and user ID. This confirms your credentials are valid.
Seeing an error after aws configure?
- Double-check that you pasted the Access Key ID and Secret Access Key exactly as shown in your downloaded file. Extra spaces or missing characters will cause authentication failures.
- Make sure you entered us-east-1 as the region (not a different region name).
- If aws sts get-caller-identity returns an "InvalidClientTokenId" error, your access key may be inactive or deleted. Return to the IAM console and create a new key.
Still stuck? Help me troubleshoot my AWS CLI configuration
Install boto3
boto3 is the official AWS SDK for Python. Your agent script imports it to make Converse API calls to Bedrock.
- Install boto3 by running this command:
pip install boto3- Verify the installation by running:
python -c "import boto3; print(boto3.__version__)"You should see a version number printed (e.g. 1.43.27 or higher). This confirms boto3 is installed and importable.
pip not found or permission error?
- If you see a "pip is not recognized" error, try pip3 install boto3 or python -m pip install boto3 instead.
- If you see a permissions error, try running your command prompt as Administrator.
Still stuck? Help me install boto3 on my system
Enable Amazon Nova Micro in the Bedrock console
Before your agent can call a model, you need to enable access to it in the Bedrock console. You will use Amazon Nova Micro, a fast and cost-effective text model from Amazon.
- Sign in to the AWS Management Console at console.aws.amazon.com.
- In the top-right region selector, make sure US East (N. Virginia) (us-east-1) is selected.
- Use the search bar at the top to search for Bedrock and select Amazon Bedrock.
- In the left sidebar, expand Bedrock configurations and click Model access.
- Click Modify model access.
- Find and select the checkbox next to Amazon Nova Micro.
- Click Next at the bottom of the page.
- Review your selection and click Submit.
Access is typically granted within a few seconds. You should see the status change to Access granted next to Amazon Nova Micro.
Why Amazon Nova Micro?
Amazon Nova Micro is Amazon's most cost-effective text model at $0.000035 per 1,000 input tokens. It supports the Converse API with tool use, which is exactly what your agent needs. For a learning project, it keeps costs to fractions of a cent.
Model access still pending?
- Refresh the page after 30 seconds. Amazon-owned models like Nova Micro are usually approved instantly.
- Make sure you are in the us-east-1 region. Model availability differs by region.
- If you see a permissions error, your IAM user may not have the AmazonBedrockFullAccess policy attached. Ask your account administrator to attach it.
Still stuck? Help me enable model access in Amazon Bedrock
Your local environment is fully connected to AWS. Next up, you will write your first API call to Bedrock and confirm the model responds.
Make Your First Bedrock Converse API Call
Your environment is set up with AWS credentials and Bedrock model access. Now it is time to talk to the AI model directly from your own Python code.
In this step, you will write a Python script that sends a message to Amazon Nova Micro using the Converse API. This is the foundation that every AI agent is built on.
In this step, get ready to:
- Create your project folder and write your first Bedrock API call.
- Add a system prompt that defines your agent's enterprise support role.
- Track token usage to monitor how much each request costs.
Create your project folder and make a basic API call
Before writing any agent logic, you need a project folder and a single Python file. You will send a simple greeting to Amazon Nova Micro to confirm your credentials and Bedrock access work end-to-end.
- Open your terminal. Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar, then type Terminal (macOS) or Command Prompt (Windows) and press Enter.
- Move to your Desktop by running this command:
cd ~/Desktop- Create your project folder and move into it by running these commands:
mkdir bedrock-agent
cd bedrock-agent- Confirm the folder exists by running:
lsYou should see the bedrock-agent folder listed on your Desktop.
- Create the agent.py file by running:
touch agent.pyNow open this file in your code editor. The next chunk of code is the first thing you will paste into it.
- Copy the following code into agent.py:
import boto3
import json
from datetime import datetime, timezone
from botocore.exceptions import ClientError
# --- Configuration ---
REGION = "us-east-1"
MODEL_ID = "amazon.nova-micro-v1:0"
# --- AWS Clients ---
bedrock_client = boto3.client("bedrock-runtime", region_name=REGION)
if __name__ == "__main__":
# Send a simple test message to confirm credentials work
response = bedrock_client.converse(
modelId=MODEL_ID,
messages=[{"role": "user", "content": [{"text": "Hello, what can you help me with?"}]}],
inferenceConfig={"maxTokens": 1024, "temperature": 0.1},
)
print(response["output"]["message"]["content"][0]["text"])What does this code do?
- The imports bring in boto3 (the AWS SDK for Python) and utility modules you will need later.
- REGION and MODEL_ID store your configuration. amazon.nova-micro-v1:0 is Amazon's cheapest Bedrock model, ideal for learning.
- bedrock_client creates a connection to the Bedrock Runtime service in us-east-1.
- The converse() call sends a single user message and prints the model's reply.
- Save agent.py.
- Run the script by executing:
python agent.pyYou should see a text response from Amazon Nova Micro describing what it can help with.
Not getting a response?
- If you see an AccessDeniedException, double-check that you enabled Amazon Nova Micro model access in the Bedrock console (Step 1).
- If you see a credentials error, confirm aws configure completed successfully and your Access Key ID is correct.
- Make sure you are running python agent.py from inside the bedrock-agent folder.
Help me debug my Bedrock Converse API call
Add a system prompt for the enterprise support role
A system prompt tells the model who it is and how to behave. Without one, the model gives generic answers. With a system prompt, it acts as a focused enterprise support assistant that uses tools instead of guessing.
- In agent.py, find the line that says if __name__ == "__main__":.
- Add the following system prompt definition directly above that line:
# --- System Prompt ---
system_prompt = [
{
"text": (
"You are an enterprise support AI agent for a technology company. "
"You help customers with order inquiries, inventory checks, and issue escalation. "
"Always use the available tools to look up real data before answering. "
"Never guess or make up order statuses or inventory levels. "
"If you cannot resolve an issue, escalate it with appropriate priority."
)
}
]What is a system prompt?
The system parameter in the Converse API sets the model's behavior before any user message arrives. It defines the agent's personality, boundaries, and instructions.
This system prompt instructs the model to use tools (which you will define in the next step) and to never fabricate data. This is critical in enterprise systems where incorrect information can cause real damage.
Now update the converse() call to include the system prompt.
- In the if __name__ == "__main__": block, find this line:
response = bedrock_client.converse(
modelId=MODEL_ID,
messages=[{"role": "user", "content": [{"text": "Hello, what can you help me with?"}]}],
inferenceConfig={"maxTokens": 1024, "temperature": 0.1},
)- Replace it with this version that includes system=system_prompt:
response = bedrock_client.converse(
modelId=MODEL_ID,
messages=[{"role": "user", "content": [{"text": "Hello, what can you help me with?"}]}],
system=system_prompt,
inferenceConfig={"maxTokens": 1024, "temperature": 0.1},
)- Save agent.py and run it again:
python agent.pyThis time the response should mention order inquiries, inventory checks, and issue escalation. The model is now acting in its assigned enterprise support role.
Response still looks generic?
- Make sure system=system_prompt is inside the converse() call, not outside it.
- Confirm the system_prompt variable is defined above the if __name__ block, not inside it.
Why is my system prompt not changing the model's response?
Add token usage tracking
Every API call to Bedrock costs tokens. Tracking token usage lets you monitor costs and understand how much context the model consumes per request.
- In agent.py, find the # --- AWS Clients --- section.
- Add the following token tracking dictionary between the client creation and the system prompt:
# --- Token Tracking ---
session_tokens = {"input": 0, "output": 0}Why track tokens across a session?
Amazon Nova Micro charges $0.000035 per 1,000 input tokens and $0.00014 per 1,000 output tokens. These numbers are tiny per request, but an agent that loops multiple times can accumulate significant usage.
By storing cumulative totals in session_tokens, you will be able to calculate your total cost at the end of a session.
Now update the main block to extract token counts from the response and print them.
- In the if __name__ == "__main__": block, find the print(response["output"]... line.
- Replace the entire main block content (everything inside if __name__ == "__main__":) with this updated version:
if __name__ == "__main__":
response = bedrock_client.converse(
modelId=MODEL_ID,
messages=[{"role": "user", "content": [{"text": "Hello, what can you help me with?"}]}],
system=system_prompt,
inferenceConfig={"maxTokens": 1024, "temperature": 0.1},
)
usage = response.get("usage", {})
session_tokens["input"] += usage.get("inputTokens", 0)
session_tokens["output"] += usage.get("outputTokens", 0)
print(response["output"]["message"]["content"][0]["text"])
print(f"Tokens - Input: {session_tokens['input']}, Output: {session_tokens['output']}, Total: {session_tokens['input'] + session_tokens['output']}")What does this code do?
- response.get("usage", {}) safely extracts the token usage metadata from the API response.
- inputTokens counts the tokens in your message and system prompt. outputTokens counts tokens the model generated.
- The totals are accumulated into session_tokens so you can track costs across multiple calls later.
- Save agent.py and run it:
python agent.pyYou should see the model's response followed by a line showing input tokens, output tokens, and the total.
Seeing 0 for all token counts?
- Double-check you are reading from response.get("usage", {}) and not response["output"]. Token metadata lives at the top level of the response, not inside the output.
- Ensure the variable name is session_tokens (not session_token without the s).
Help me debug my token usage tracking
✔️ Awesome, I've got everything!
Great. Double check you have saved agent.py before moving on.
ⓧ I'd like to double check the full code
Here is the complete agent.py at the end of this step:
import boto3
import json
from datetime import datetime, timezone
from botocore.exceptions import ClientError
# --- Configuration ---
REGION = "us-east-1"
MODEL_ID = "amazon.nova-micro-v1:0"
# --- AWS Clients ---
bedrock_client = boto3.client("bedrock-runtime", region_name=REGION)
# --- Token Tracking ---
session_tokens = {"input": 0, "output": 0}
# --- System Prompt ---
system_prompt = [
{
"text": (
"You are an enterprise support AI agent for a technology company. "
"You help customers with order inquiries, inventory checks, and issue escalation. "
"Always use the available tools to look up real data before answering. "
"Never guess or make up order statuses or inventory levels. "
"If you cannot resolve an issue, escalate it with appropriate priority."
)
}
]
if __name__ == "__main__":
response = bedrock_client.converse(
modelId=MODEL_ID,
messages=[{"role": "user", "content": [{"text": "Hello, what can you help me with?"}]}],
system=system_prompt,
inferenceConfig={"maxTokens": 1024, "temperature": 0.1},
)
usage = response.get("usage", {})
session_tokens["input"] += usage.get("inputTokens", 0)
session_tokens["output"] += usage.get("outputTokens", 0)
print(response["output"]["message"]["content"][0]["text"])
print(f"Tokens - Input: {session_tokens['input']}, Output: {session_tokens['output']}, Total: {session_tokens['input'] + session_tokens['output']}")Your script now talks directly to Amazon Nova Micro, behaves as an enterprise support agent, and reports its token costs. Next up, you will give this agent real tools to call.
Define Enterprise Tools with JSON Schema
Your agent can talk, but it cannot take action yet. Right now, if a customer asks about an order, the model can only guess.
To make it useful, you need to give the model access to tools. In this step, you will define three enterprise tools using JSON Schema, create Python functions that simulate backend systems, and wire everything into the Converse API so the model knows what capabilities it has.
In this step, get ready to:
- Define JSON Schema tool specifications that tell Bedrock what your tools accept.
- Create Python functions that simulate enterprise backend systems.
- Pass the tool definitions to the Converse API so the model can use them.
Create tool definitions with JSON Schema
The Bedrock Converse API uses a toolConfig parameter to tell the model what tools are available. Each tool needs a toolSpec containing a name, a description, and an inputSchema that describes the expected inputs using JSON Schema.
- In agent.py, find the # --- System Prompt --- section.
- Add the following tool_config dictionary directly above it:
# --- Tool Definitions ---
tool_config = {
"tools": [
{
"toolSpec": {
"name": "get_order_status",
"description": "Retrieves the current status of a customer order including shipping ETA and associated product ID. Use this when a customer asks about their order.",
"inputSchema": {
"json": {
"type": "object",
"properties": {
"order_id": {
"type": "string",
"description": "The order ID, e.g. ORD-1234",
}
},
"required": ["order_id"],
}
},
}
},
{
"toolSpec": {
"name": "check_inventory",
"description": "Checks current inventory levels for a product including stock count and warehouse location. Use this when checking product availability.",
"inputSchema": {
"json": {
"type": "object",
"properties": {
"product_id": {
"type": "string",
"description": "The product ID, e.g. PROD-567",
}
},
"required": ["product_id"],
}
},
}
},
{
"toolSpec": {
"name": "escalate_issue",
"description": "Escalates an unresolved issue to the support team with a priority level. Use this when the customer's problem cannot be resolved with available tools.",
"inputSchema": {
"json": {
"type": "object",
"properties": {
"issue_description": {
"type": "string",
"description": "A brief description of the issue",
},
"priority": {
"type": "string",
"description": "Priority level: low, medium, high, or critical",
},
},
"required": ["issue_description", "priority"],
}
},
}
},
]
}What does this code do?
Each toolSpec entry describes one tool the model can request. The name identifies the tool, the description tells the model when to use it, and the inputSchema defines what parameters the tool accepts.
The model reads these definitions to decide which tool to call and what arguments to pass. Better descriptions lead to smarter tool selection.
Why JSON Schema?
Amazon Nova models require the top-level inputSchema to be of type object with properties and required fields. This is the same JSON Schema standard used across the industry for API contracts and validation.
You will not see a change yet. The next substep adds the Python functions that these definitions point to.
Create simulated enterprise tool functions
Each tool definition needs a matching Python function that runs when the agent calls it. In a production system, these would connect to real databases or APIs like SAP. For this project, you will use simulated data so you can focus on the agent pattern.
- In agent.py, find the # --- Token Tracking --- section and the session_tokens line below it.
- Add the following get_order_status function below session_tokens, leaving one blank line after it:
# --- Simulated Enterprise Tools ---
def get_order_status(order_id):
orders = {
"ORD-1234": {"status": "shipped", "product_id": "PROD-567", "eta": "2026-06-13"},
"ORD-5678": {"status": "processing", "product_id": "PROD-890", "eta": "2026-06-15"},
"ORD-9999": {"status": "cancelled", "product_id": "PROD-123", "eta": None},
}
if order_id in orders:
return {"success": True, "order": orders[order_id]}
return {"success": False, "error": f"Order {order_id} not found"}What does this function do?
This function simulates looking up an order in your company's order management system. It stores three sample orders in a dictionary and returns the matching record, or an error if the order is not found.
The product_id field in each order links it to a product. This lets the agent chain tool calls (look up order, then check inventory for that product).
- Add the check_inventory function directly below get_order_status, leaving one blank line between them:
def check_inventory(product_id):
inventory = {
"PROD-567": {"name": "Enterprise Router X1", "in_stock": 42, "warehouse": "US-East"},
"PROD-890": {"name": "Cloud Gateway Pro", "in_stock": 0, "warehouse": "US-West"},
"PROD-123": {"name": "Network Switch S3", "in_stock": 15, "warehouse": "EU-Central"},
}
if product_id in inventory:
return {"success": True, "inventory": inventory[product_id]}
return {"success": False, "error": f"Product {product_id} not found"}What does this function do?
This simulates a warehouse management system. Notice that PROD-890 has 0 in stock. This lets you test how the agent handles out-of-stock scenarios and decides to escalate.
- Add the escalate_issue function directly below check_inventory, leaving one blank line between them:
def escalate_issue(issue_description, priority):
ticket_id = f"TKT-{datetime.now(timezone.utc).strftime('%Y%m%d%H%M%S')}"
return {
"success": True,
"ticket_id": ticket_id,
"priority": priority,
"message": f"Issue escalated: {issue_description}",
}What does this function do?
This simulates creating a support ticket. It generates a unique ticket ID using the current UTC timestamp and returns a confirmation with the priority level. In production, this would create a ticket in a system like Jira or ServiceNow.
You will not see a change yet. The next substep connects these functions to the Converse API call.
Wire the tool config into the Converse API call
With the tool definitions and functions ready, you need to pass tool_config to the Converse API so the model knows these tools exist. You will also add configuration constants for the next steps.
- In agent.py, find the # --- Configuration --- section at the top of the file.
- Find this line:
MODEL_ID = "amazon.nova-micro-v1:0"- Add these two lines directly below it:
BUCKET_NAME = "enterprise-agent-audit" # Append your initials and date when running
MAX_RECURSIONS = 5What are these constants for?
BUCKET_NAME is the S3 bucket you will create in a later step for audit logging. MAX_RECURSIONS limits how many times the agent can loop through tool calls, preventing infinite loops.
- Next, find the # --- AWS Clients --- section.
- Add an S3 client below the existing bedrock_client line:
s3_client = boto3.client("s3", region_name=REGION)- Now scroll down to the if __name__ == "__main__": block at the bottom.
- Replace the entire if __name__ block with the following code that passes toolConfig to the Converse API:
if __name__ == "__main__":
response = bedrock_client.converse(
modelId=MODEL_ID,
messages=[{"role": "user", "content": [{"text": "What can you help me with?"}]}],
system=system_prompt,
toolConfig=tool_config,
inferenceConfig={"maxTokens": 1024, "temperature": 0.1},
)
usage = response.get("usage", {})
session_tokens["input"] += usage.get("inputTokens", 0)
session_tokens["output"] += usage.get("outputTokens", 0)
print(response["output"]["message"]["content"][0]["text"])What changed here?
The key addition is toolConfig=tool_config. This tells the model about all three available tools. The model can now decide whether to respond with text or request a tool call based on the user's message.
- Save agent.py.
- Run the script by entering this command in your terminal:
python agent.pyYou should see the model describe its capabilities, mentioning that it can help with order status, inventory checks, and issue escalation.
Seeing an error about toolConfig?
- Make sure tool_config is defined above the if __name__ block. Python reads top to bottom, so the variable must exist before it is used.
- Check that the tool_config dictionary has no missing commas between the toolSpec entries. Each entry in the tools list needs a trailing comma.
- Verify that toolConfig=tool_config uses the exact variable name tool_config (with an underscore).
Still stuck? Help me debug a toolConfig error in my Bedrock Converse API call.
✔️ Awesome, I've got everything!
Great. Double check you have saved agent.py and your script ran without errors.
ⓧ I'd like to double check the full code
Here is the complete agent.py file at this point:
import boto3
import json
from datetime import datetime, timezone
from botocore.exceptions import ClientError
# --- Configuration ---
REGION = "us-east-1"
MODEL_ID = "amazon.nova-micro-v1:0"
BUCKET_NAME = "enterprise-agent-audit" # Append your initials and date when running
MAX_RECURSIONS = 5
# --- AWS Clients ---
bedrock_client = boto3.client("bedrock-runtime", region_name=REGION)
s3_client = boto3.client("s3", region_name=REGION)
# --- Token Tracking ---
session_tokens = {"input": 0, "output": 0}
# --- Simulated Enterprise Tools ---
def get_order_status(order_id):
orders = {
"ORD-1234": {"status": "shipped", "product_id": "PROD-567", "eta": "2026-06-13"},
"ORD-5678": {"status": "processing", "product_id": "PROD-890", "eta": "2026-06-15"},
"ORD-9999": {"status": "cancelled", "product_id": "PROD-123", "eta": None},
}
if order_id in orders:
return {"success": True, "order": orders[order_id]}
return {"success": False, "error": f"Order {order_id} not found"}
def check_inventory(product_id):
inventory = {
"PROD-567": {"name": "Enterprise Router X1", "in_stock": 42, "warehouse": "US-East"},
"PROD-890": {"name": "Cloud Gateway Pro", "in_stock": 0, "warehouse": "US-West"},
"PROD-123": {"name": "Network Switch S3", "in_stock": 15, "warehouse": "EU-Central"},
}
if product_id in inventory:
return {"success": True, "inventory": inventory[product_id]}
return {"success": False, "error": f"Product {product_id} not found"}
def escalate_issue(issue_description, priority):
ticket_id = f"TKT-{datetime.now(timezone.utc).strftime('%Y%m%d%H%M%S')}"
return {
"success": True,
"ticket_id": ticket_id,
"priority": priority,
"message": f"Issue escalated: {issue_description}",
}
# --- Tool Definitions ---
tool_config = {
"tools": [
{
"toolSpec": {
"name": "get_order_status",
"description": "Retrieves the current status of a customer order including shipping ETA and associated product ID. Use this when a customer asks about their order.",
"inputSchema": {
"json": {
"type": "object",
"properties": {
"order_id": {
"type": "string",
"description": "The order ID, e.g. ORD-1234",
}
},
"required": ["order_id"],
}
},
}
},
{
"toolSpec": {
"name": "check_inventory",
"description": "Checks current inventory levels for a product including stock count and warehouse location. Use this when checking product availability.",
"inputSchema": {
"json": {
"type": "object",
"properties": {
"product_id": {
"type": "string",
"description": "The product ID, e.g. PROD-567",
}
},
"required": ["product_id"],
}
},
}
},
{
"toolSpec": {
"name": "escalate_issue",
"description": "Escalates an unresolved issue to the support team with a priority level. Use this when the customer's problem cannot be resolved with available tools.",
"inputSchema": {
"json": {
"type": "object",
"properties": {
"issue_description": {
"type": "string",
"description": "A brief description of the issue",
},
"priority": {
"type": "string",
"description": "Priority level: low, medium, high, or critical",
},
},
"required": ["issue_description", "priority"],
}
},
}
},
]
}
# --- System Prompt ---
system_prompt = [
{
"text": (
"You are an enterprise support AI agent for a technology company. "
"You help customers with order inquiries, inventory checks, and issue escalation. "
"Always use the available tools to look up real data before answering. "
"Never guess or make up order statuses or inventory levels. "
"If you cannot resolve an issue, escalate it with appropriate priority."
)
}
]
if __name__ == "__main__":
response = bedrock_client.converse(
modelId=MODEL_ID,
messages=[{"role": "user", "content": [{"text": "What can you help me with?"}]}],
system=system_prompt,
toolConfig=tool_config,
inferenceConfig={"maxTokens": 1024, "temperature": 0.1},
)
usage = response.get("usage", {})
session_tokens["input"] += usage.get("inputTokens", 0)
session_tokens["output"] += usage.get("outputTokens", 0)
print(response["output"]["message"]["content"][0]["text"])Your tools are defined and the model knows about them. Next up, you will build the agent loop that actually executes these tools when the model requests them.
Build the Agent Loop
Your agent can see its tools, but it has no way to actually use them. Right now, if the model decides it needs to call get_order_status, nothing happens.
In this step, you will build the agent loop. This is the observe-think-act cycle that lets the agent call tools, read their results, and keep going until the task is complete.
In this step, get ready to:
- Create a tool dispatcher that routes the model's tool requests to the correct Python function.
- Implement the recursive agent loop with a max recursion limit to prevent infinite loops.
- Test the full loop with a multi-tool query that requires two tool calls to answer.
Create the tool invocation dispatcher
When Bedrock decides to use a tool, it returns the tool name and input as data in the response. Your code needs a function that takes that tool name and routes it to the correct Python function.
- In agent.py, find the closing bracket of the system_prompt list (just before the if __name__ == "__main__": block).
- Add the following tool dispatcher function below it:
# --- Tool Invocation ---
def invoke_tool(tool_name, tool_input):
if tool_name == "get_order_status":
return get_order_status(tool_input["order_id"])
elif tool_name == "check_inventory":
return check_inventory(tool_input["product_id"])
elif tool_name == "escalate_issue":
return escalate_issue(tool_input["issue_description"], tool_input["priority"])
else:
return {"error": f"Unknown tool: {tool_name}"}What does this code do?
This function acts as a router. When the model asks to call a tool by name, this function checks which name was requested and calls the matching Python function with the correct arguments.
The else branch handles the edge case where the model hallucinates a tool name that does not exist.
- Save agent.py.
Implement the recursive agent loop
A single tool call is often not enough. The agent might need to check an order first, then check the inventory for the product in that order. This requires a loop that keeps calling the model until it has all the information it needs to give a final answer.
- In agent.py, add the following two functions directly below your invoke_tool function:
# --- Agent Loop ---
def run_agent(user_query, conversation=None):
if conversation is None:
conversation = []
conversation.append({"role": "user", "content": [{"text": user_query}]})
audit_record = {"query": user_query, "tool_calls": [], "timestamp": datetime.now(timezone.utc).isoformat()}
response = process_conversation(conversation, audit_record, MAX_RECURSIONS)
return response, audit_record
def process_conversation(conversation, audit_record, recursions_left):
if recursions_left <= 0:
print("WARNING: Maximum recursions reached.")
return "I was unable to complete your request. Please try again."
response = bedrock_client.converse(
modelId=MODEL_ID,
messages=conversation,
system=system_prompt,
toolConfig=tool_config,
inferenceConfig={"maxTokens": 1024, "temperature": 0.1},
)
# Track tokens
usage = response.get("usage", {})
session_tokens["input"] += usage.get("inputTokens", 0)
session_tokens["output"] += usage.get("outputTokens", 0)
stop_reason = response["stopReason"]
message = response["output"]["message"]
conversation.append(message)
if stop_reason == "end_turn":
final_text = ""
for block in message["content"]:
if "text" in block:
final_text += block["text"]
audit_record["response"] = final_text
audit_record["tokens"] = dict(session_tokens)
return final_text
elif stop_reason == "tool_use":
tool_results = []
for block in message["content"]:
if "toolUse" in block:
tool_use = block["toolUse"]
tool_name = tool_use["name"]
tool_input = tool_use["input"]
tool_use_id = tool_use["toolUseId"]
print(f" [Agent] Calling tool: {tool_name} with input: {json.dumps(tool_input)}")
result = invoke_tool(tool_name, tool_input)
print(f" [Agent] Tool result: {json.dumps(result)}")
audit_record["tool_calls"].append(
{"tool": tool_name, "input": tool_input, "result": result}
)
tool_results.append(
{"toolResult": {"toolUseId": tool_use_id, "content": [{"json": result}]}}
)
conversation.append({"role": "user", "content": tool_results})
return process_conversation(conversation, audit_record, recursions_left - 1)
else:
return f"Unexpected stop reason: {stop_reason}"What does this code do?
These two functions work together to form the agent loop.
- run_agent is the entry point. It builds the initial conversation message and creates an audit_record dictionary to track everything the agent does.
- process_conversation calls the Bedrock Converse API, then checks stopReason. If it is "end_turn", the model is done and returns its final text. If it is "tool_use", the model wants to call a tool.
- When a tool is requested, the function extracts the tool name and input, calls invoke_tool, then sends the result back as a toolResult message and calls itself again.
- The recursions_left parameter decreases by 1 on every recursive call. If it hits 0, the loop stops. This prevents infinite loops if the model keeps requesting tools without finishing.
Why recursion instead of a while loop?
Each recursive call passes the updated conversation and audit record forward naturally. The recursion depth also acts as a built-in safety limit. When you see this pattern in production systems, it is often called the observe-think-act loop.
- Save agent.py.
Test the full agent loop with a multi-tool query
Time to see the agent loop in action. You will replace the existing test code with a query that forces the agent to call multiple tools before it can answer.
- In agent.py, find the existing if __name__ == "__main__": block at the bottom of your file.
- Delete the entire block (everything from that line to the end of the file).
- Replace it with the following code:
if __name__ == "__main__":
print("=" * 60)
print("Enterprise AI Agent - Powered by Amazon Bedrock")
print("=" * 60)
query = "What's the status of order ORD-1234 and is that product still in stock?"
print(f"User: {query}")
response, audit = run_agent(query)
print(f"\nAgent: {response}")What does this test do?
This query asks about an order AND its product's stock level. The agent should first call get_order_status to find the product ID, then call check_inventory with that product ID. Two tool calls, one final answer.
- Save agent.py.
- Run the script by executing this command in your terminal:
python agent.py- You should see [Agent] Calling tool: lines printed for each tool the agent calls, followed by a final synthesized answer that mentions the order status AND the inventory level.
Not seeing any tool calls?
- Make sure you deleted the old if __name__ == "__main__": block completely. Having two of them will cause only the first one to run.
- Check that your invoke_tool and run_agent functions are defined ABOVE the if __name__ block, not inside it.
- If you see a NameError, double-check that the function names match exactly between invoke_tool and the three tool functions defined earlier.
Help me debug my agent loop
✔️ Awesome, I've got everything!
Great. Double check you have saved your file before moving on.
ⓧ I'd like to double check the full code
Here is the complete agent.py file as it should look at the end of this step:
import boto3
import json
from datetime import datetime, timezone
from botocore.exceptions import ClientError
# --- Configuration ---
REGION = "us-east-1"
MODEL_ID = "amazon.nova-micro-v1:0"
BUCKET_NAME = "enterprise-agent-audit" # Append your initials and date when running
MAX_RECURSIONS = 5
# --- AWS Clients ---
bedrock_client = boto3.client("bedrock-runtime", region_name=REGION)
s3_client = boto3.client("s3", region_name=REGION)
# --- Token Tracking ---
session_tokens = {"input": 0, "output": 0}
# --- Simulated Enterprise Tools ---
def get_order_status(order_id):
orders = {
"ORD-1234": {"status": "shipped", "product_id": "PROD-567", "eta": "2026-06-13"},
"ORD-5678": {"status": "processing", "product_id": "PROD-890", "eta": "2026-06-15"},
"ORD-9999": {"status": "cancelled", "product_id": "PROD-123", "eta": None},
}
if order_id in orders:
return {"success": True, "order": orders[order_id]}
return {"success": False, "error": f"Order {order_id} not found"}
def check_inventory(product_id):
inventory = {
"PROD-567": {"name": "Enterprise Router X1", "in_stock": 42, "warehouse": "US-East"},
"PROD-890": {"name": "Cloud Gateway Pro", "in_stock": 0, "warehouse": "US-West"},
"PROD-123": {"name": "Network Switch S3", "in_stock": 15, "warehouse": "EU-Central"},
}
if product_id in inventory:
return {"success": True, "inventory": inventory[product_id]}
return {"success": False, "error": f"Product {product_id} not found"}
def escalate_issue(issue_description, priority):
ticket_id = f"TKT-{datetime.now(timezone.utc).strftime('%Y%m%d%H%M%S')}"
return {
"success": True,
"ticket_id": ticket_id,
"priority": priority,
"message": f"Issue escalated: {issue_description}",
}
# --- Tool Definitions ---
tool_config = {
"tools": [
{
"toolSpec": {
"name": "get_order_status",
"description": "Retrieves the current status of a customer order including shipping ETA and associated product ID. Use this when a customer asks about their order.",
"inputSchema": {
"json": {
"type": "object",
"properties": {
"order_id": {
"type": "string",
"description": "The order ID, e.g. ORD-1234",
}
},
"required": ["order_id"],
}
},
}
},
{
"toolSpec": {
"name": "check_inventory",
"description": "Checks current inventory levels for a product including stock count and warehouse location. Use this when checking product availability.",
"inputSchema": {
"json": {
"type": "object",
"properties": {
"product_id": {
"type": "string",
"description": "The product ID, e.g. PROD-567",
}
},
"required": ["product_id"],
}
},
}
},
{
"toolSpec": {
"name": "escalate_issue",
"description": "Escalates an unresolved issue to the support team with a priority level. Use this when the customer's problem cannot be resolved with available tools.",
"inputSchema": {
"json": {
"type": "object",
"properties": {
"issue_description": {
"type": "string",
"description": "A brief description of the issue",
},
"priority": {
"type": "string",
"description": "Priority level: low, medium, high, or critical",
},
},
"required": ["issue_description", "priority"],
}
},
}
},
]
}
# --- System Prompt ---
system_prompt = [
{
"text": (
"You are an enterprise support AI agent for a technology company. "
"You help customers with order inquiries, inventory checks, and issue escalation. "
"Always use the available tools to look up real data before answering. "
"Never guess or make up order statuses or inventory levels. "
"If you cannot resolve an issue, escalate it with appropriate priority."
)
}
]
# --- Tool Invocation ---
def invoke_tool(tool_name, tool_input):
if tool_name == "get_order_status":
return get_order_status(tool_input["order_id"])
elif tool_name == "check_inventory":
return check_inventory(tool_input["product_id"])
elif tool_name == "escalate_issue":
return escalate_issue(tool_input["issue_description"], tool_input["priority"])
else:
return {"error": f"Unknown tool: {tool_name}"}
# --- Agent Loop ---
def run_agent(user_query, conversation=None):
if conversation is None:
conversation = []
conversation.append({"role": "user", "content": [{"text": user_query}]})
audit_record = {"query": user_query, "tool_calls": [], "timestamp": datetime.now(timezone.utc).isoformat()}
response = process_conversation(conversation, audit_record, MAX_RECURSIONS)
return response, audit_record
def process_conversation(conversation, audit_record, recursions_left):
if recursions_left <= 0:
print("WARNING: Maximum recursions reached.")
return "I was unable to complete your request. Please try again."
response = bedrock_client.converse(
modelId=MODEL_ID,
messages=conversation,
system=system_prompt,
toolConfig=tool_config,
inferenceConfig={"maxTokens": 1024, "temperature": 0.1},
)
# Track tokens
usage = response.get("usage", {})
session_tokens["input"] += usage.get("inputTokens", 0)
session_tokens["output"] += usage.get("outputTokens", 0)
stop_reason = response["stopReason"]
message = response["output"]["message"]
conversation.append(message)
if stop_reason == "end_turn":
final_text = ""
for block in message["content"]:
if "text" in block:
final_text += block["text"]
audit_record["response"] = final_text
audit_record["tokens"] = dict(session_tokens)
return final_text
elif stop_reason == "tool_use":
tool_results = []
for block in message["content"]:
if "toolUse" in block:
tool_use = block["toolUse"]
tool_name = tool_use["name"]
tool_input = tool_use["input"]
tool_use_id = tool_use["toolUseId"]
print(f" [Agent] Calling tool: {tool_name} with input: {json.dumps(tool_input)}")
result = invoke_tool(tool_name, tool_input)
print(f" [Agent] Tool result: {json.dumps(result)}")
audit_record["tool_calls"].append(
{"tool": tool_name, "input": tool_input, "result": result}
)
tool_results.append(
{"toolResult": {"toolUseId": tool_use_id, "content": [{"json": result}]}}
)
conversation.append({"role": "user", "content": tool_results})
return process_conversation(conversation, audit_record, recursions_left - 1)
else:
return f"Unexpected stop reason: {stop_reason}"
if __name__ == "__main__":
print("=" * 60)
print("Enterprise AI Agent - Powered by Amazon Bedrock")
print("=" * 60)
query = "What's the status of order ORD-1234 and is that product still in stock?"
print(f"User: {query}")
response, audit = run_agent(query)
print(f"\nAgent: {response}")Your agent can now autonomously decide which tools to call, execute them, and keep looping until it has a complete answer. Next up, you will add S3 audit logging so every decision the agent makes is permanently recorded for compliance.
Add S3 Audit Logging
Your agent can now autonomously select tools and loop until it answers a question. In production, that's only half the story.
Enterprise AI systems require a complete audit trail of every decision the agent makes. Regulators, security teams, and compliance officers need to trace exactly which tools were called, what data was accessed, and what the agent said. In this step, you'll add S3-based audit logging so every agent interaction is permanently recorded.
In this step, get ready to:
- Create an S3 bucket to store the audit trail.
- Write functions that serialize and upload each agent interaction to S3.
- Run multiple test queries and verify the audit logs accumulate in your bucket.
Create the S3 audit bucket function
Before your agent can log anything, it needs a place to store the data. You'll write a function that creates an S3 bucket programmatically using boto3.
Why a unique bucket name?
S3 bucket names are globally unique across all AWS accounts. Two people cannot have a bucket with the same name. Before you run the script later, you'll append your initials and today's date to the BUCKET_NAME constant (e.g. enterprise-agent-audit-jm-20260611).
- In agent.py, scroll to the bottom of the file and find the line if __name__ == "__main__":.
- Add the following create_audit_bucket function ABOVE that line:
# --- S3 Audit Logging ---
def create_audit_bucket(bucket_name):
try:
s3_client.create_bucket(Bucket=bucket_name)
print(f"Created audit bucket: {bucket_name}")
except ClientError as e:
if e.response["Error"]["Code"] == "BucketAlreadyOwnedByYou":
print(f"Audit bucket already exists: {bucket_name}")
else:
raiseWhat does this code do?
- s3_client.create_bucket(Bucket=bucket_name) calls the S3 API to create a new bucket with the given name.
- The try/except block catches the case where the bucket already exists from a previous run. If you own it, the script continues normally. If there's a different error, it raises the exception so you can debug.
- Save agent.py.
You won't see a change yet. The next function will use this bucket to store audit logs.
Write the logging and cleanup functions
Now you need a function that takes the audit record your agent already builds (the query, tool calls, results, and final response) and uploads it to S3 as a JSON file with a timestamped path.
- In agent.py, add the following log_to_s3 function directly below create_audit_bucket:
def log_to_s3(bucket_name, audit_record):
timestamp = datetime.now(timezone.utc).strftime("%Y/%m/%d/%H%M%S")
key = f"audit-logs/{timestamp}-agent-interaction.json"
s3_client.put_object(
Bucket=bucket_name,
Key=key,
Body=json.dumps(audit_record, indent=2, default=str),
ContentType="application/json",
)
print(f" [Audit] Logged to s3://{bucket_name}/{key}")What does this code do?
- datetime.now(timezone.utc).strftime("%Y/%m/%d/%H%M%S") generates a timestamp that becomes the folder path in S3 (e.g. audit-logs/2026/06/11/143022-agent-interaction.json). This organizes logs by date automatically.
- s3_client.put_object uploads the serialized JSON to S3. The default=str argument handles any non-serializable objects (like datetime values) by converting them to strings.
- Save agent.py.
You won't see a change yet. The next chunk adds a cleanup utility, then we'll wire everything together.
- Add the following cleanup_bucket function below log_to_s3:
# --- Cleanup ---
def cleanup_bucket(bucket_name):
try:
objects = s3_client.list_objects_v2(Bucket=bucket_name)
if "Contents" in objects:
for obj in objects["Contents"]:
s3_client.delete_object(Bucket=bucket_name, Key=obj["Key"])
s3_client.delete_bucket(Bucket=bucket_name)
print(f"Deleted bucket: {bucket_name}")
except ClientError as e:
print(f"Cleanup error: {e}")What does this code do?
S3 won't let you delete a bucket that still contains objects. This function first lists everything in the bucket, deletes each object one by one, then deletes the empty bucket itself.
You'll use this in the Before You Go section to clean up your resources. For now, it sits ready at the bottom of your file.
- Save agent.py.
You won't see a change yet. The next chunk replaces your main block to run everything together.
Seeing a syntax error?
- Make sure create_audit_bucket, log_to_s3, and cleanup_bucket are all at the top level (no indentation) and placed above the if __name__ == "__main__": block.
- Check that each function definition starts with def and has a colon at the end of the line.
Help me fix a syntax error in my S3 logging functions in agent.py.
Run multiple test queries and verify the audit trail
Time to wire everything together. You'll replace the existing main block with one that creates the bucket, runs three different business queries, logs each interaction to S3, and prints a cost summary.
- First, update the BUCKET_NAME constant at the top of agent.py to include your initials and today's date. For example, change "enterprise-agent-audit" to something like "enterprise-agent-audit-jm-20260611".
- Find the existing if __name__ == "__main__": block at the bottom of the file.
- Replace the entire block (from if __name__ to the end of the file) with the following code:
# --- Main ---
if __name__ == "__main__":
print("=" * 60)
print("Enterprise AI Agent - Powered by Amazon Bedrock")
print("=" * 60)
create_audit_bucket(BUCKET_NAME)
test_queries = [
"What's the status of order ORD-1234 and is that product still in stock?",
"Check if product PROD-890 is available. If not, escalate as high priority.",
"I need the status of order ORD-9999.",
]
for query in test_queries:
print(f"\n{'─' * 60}")
print(f"User: {query}")
print(f"{'─' * 60}")
response, audit = run_agent(query)
print(f"\nAgent: {response}")
log_to_s3(BUCKET_NAME, audit)
# Print cost summary
input_cost = (session_tokens["input"] / 1000) * 0.000035
output_cost = (session_tokens["output"] / 1000) * 0.00014
total_cost = input_cost + output_cost
print(f"\n{'=' * 60}")
print(f"Session Summary")
print(f" Input tokens: {session_tokens['input']}")
print(f" Output tokens: {session_tokens['output']}")
print(f" Estimated cost: ${total_cost:.6f}")
print(f"{'=' * 60}")
# Uncomment to clean up:
# cleanup_bucket(BUCKET_NAME)What does this code do?
- create_audit_bucket(BUCKET_NAME) creates your S3 bucket (or confirms it exists) before running any queries.
- The for loop sends three different business queries to the agent. Each one exercises different tools (order lookup, inventory check, escalation).
- log_to_s3(BUCKET_NAME, audit) uploads the complete audit record after each interaction. Every tool call, result, and final response is preserved.
- The cost summary calculates your total spend using Amazon Nova Micro pricing ($0.000035 per 1K input tokens, $0.00014 per 1K output tokens).
- Save agent.py.
- Run the script by executing this command in your terminal:
python agent.pyYou should see the agent process three queries, calling tools like get_order_status and check_inventory for each one. After each query, an [Audit] line prints the S3 path where the log was saved.
At the end, you'll see a Session Summary with your total token usage and estimated cost (likely under $0.001).
What is the enterprise compliance pattern?
Every AI decision your agent makes is now permanently recorded in S3 with a timestamped path. Compliance teams can query these logs to answer questions like "What did the AI tell customer X?" or "Which tools did the agent use on this date?"
This is the same pattern used in regulated industries (finance, healthcare, government) where AI decisions must be auditable.
Seeing an AccessDenied error?
- Make sure your AWS credentials have S3 permissions. The IAM user or role you configured with aws configure needs s3:CreateBucket, s3:PutObject, and s3:ListBucket permissions.
- If you see a bucket name error, make sure you personalized BUCKET_NAME with your initials and date. Someone else may already own the generic name.
- If you see a region error, confirm your REGION constant matches the region you configured with aws configure.
Help me fix an AccessDenied or bucket error when running my Bedrock agent with S3 logging.
✔️ Awesome, I've got everything!
Great. Double check you've saved agent.py and that your terminal shows three successful queries with [Audit] log confirmations.
ⓧ I'd like to double check the full code
Here's the complete agent.py at this point:
import boto3
import json
from datetime import datetime, timezone
from botocore.exceptions import ClientError
# --- Configuration ---
REGION = "us-east-1"
MODEL_ID = "amazon.nova-micro-v1:0"
BUCKET_NAME = "enterprise-agent-audit" # Append your initials and date when running
MAX_RECURSIONS = 5
# --- AWS Clients ---
bedrock_client = boto3.client("bedrock-runtime", region_name=REGION)
s3_client = boto3.client("s3", region_name=REGION)
# --- Token Tracking ---
session_tokens = {"input": 0, "output": 0}
# --- Simulated Enterprise Tools ---
def get_order_status(order_id):
orders = {
"ORD-1234": {"status": "shipped", "product_id": "PROD-567", "eta": "2026-06-13"},
"ORD-5678": {"status": "processing", "product_id": "PROD-890", "eta": "2026-06-15"},
"ORD-9999": {"status": "cancelled", "product_id": "PROD-123", "eta": None},
}
if order_id in orders:
return {"success": True, "order": orders[order_id]}
return {"success": False, "error": f"Order {order_id} not found"}
def check_inventory(product_id):
inventory = {
"PROD-567": {"name": "Enterprise Router X1", "in_stock": 42, "warehouse": "US-East"},
"PROD-890": {"name": "Cloud Gateway Pro", "in_stock": 0, "warehouse": "US-West"},
"PROD-123": {"name": "Network Switch S3", "in_stock": 15, "warehouse": "EU-Central"},
}
if product_id in inventory:
return {"success": True, "inventory": inventory[product_id]}
return {"success": False, "error": f"Product {product_id} not found"}
def escalate_issue(issue_description, priority):
ticket_id = f"TKT-{datetime.now(timezone.utc).strftime('%Y%m%d%H%M%S')}"
return {
"success": True,
"ticket_id": ticket_id,
"priority": priority,
"message": f"Issue escalated: {issue_description}",
}
# --- Tool Definitions ---
tool_config = {
"tools": [
{
"toolSpec": {
"name": "get_order_status",
"description": "Retrieves the current status of a customer order including shipping ETA and associated product ID. Use this when a customer asks about their order.",
"inputSchema": {
"json": {
"type": "object",
"properties": {
"order_id": {
"type": "string",
"description": "The order ID, e.g. ORD-1234",
}
},
"required": ["order_id"],
}
},
}
},
{
"toolSpec": {
"name": "check_inventory",
"description": "Checks current inventory levels for a product including stock count and warehouse location. Use this when checking product availability.",
"inputSchema": {
"json": {
"type": "object",
"properties": {
"product_id": {
"type": "string",
"description": "The product ID, e.g. PROD-567",
}
},
"required": ["product_id"],
}
},
}
},
{
"toolSpec": {
"name": "escalate_issue",
"description": "Escalates an unresolved issue to the support team with a priority level. Use this when the customer's problem cannot be resolved with available tools.",
"inputSchema": {
"json": {
"type": "object",
"properties": {
"issue_description": {
"type": "string",
"description": "A brief description of the issue",
},
"priority": {
"type": "string",
"description": "Priority level: low, medium, high, or critical",
},
},
"required": ["issue_description", "priority"],
}
},
}
},
]
}
# --- System Prompt ---
system_prompt = [
{
"text": (
"You are an enterprise support AI agent for a technology company. "
"You help customers with order inquiries, inventory checks, and issue escalation. "
"Always use the available tools to look up real data before answering. "
"Never guess or make up order statuses or inventory levels. "
"If you cannot resolve an issue, escalate it with appropriate priority."
)
}
]
# --- Tool Invocation ---
def invoke_tool(tool_name, tool_input):
if tool_name == "get_order_status":
return get_order_status(tool_input["order_id"])
elif tool_name == "check_inventory":
return check_inventory(tool_input["product_id"])
elif tool_name == "escalate_issue":
return escalate_issue(tool_input["issue_description"], tool_input["priority"])
else:
return {"error": f"Unknown tool: {tool_name}"}
# --- Agent Loop ---
def run_agent(user_query, conversation=None):
if conversation is None:
conversation = []
conversation.append({"role": "user", "content": [{"text": user_query}]})
audit_record = {"query": user_query, "tool_calls": [], "timestamp": datetime.now(timezone.utc).isoformat()}
response = process_conversation(conversation, audit_record, MAX_RECURSIONS)
return response, audit_record
def process_conversation(conversation, audit_record, recursions_left):
if recursions_left <= 0:
print("WARNING: Maximum recursions reached.")
return "I was unable to complete your request. Please try again."
response = bedrock_client.converse(
modelId=MODEL_ID,
messages=conversation,
system=system_prompt,
toolConfig=tool_config,
inferenceConfig={"maxTokens": 1024, "temperature": 0.1},
)
# Track tokens
usage = response.get("usage", {})
session_tokens["input"] += usage.get("inputTokens", 0)
session_tokens["output"] += usage.get("outputTokens", 0)
stop_reason = response["stopReason"]
message = response["output"]["message"]
conversation.append(message)
if stop_reason == "end_turn":
final_text = ""
for block in message["content"]:
if "text" in block:
final_text += block["text"]
audit_record["response"] = final_text
audit_record["tokens"] = dict(session_tokens)
return final_text
elif stop_reason == "tool_use":
tool_results = []
for block in message["content"]:
if "toolUse" in block:
tool_use = block["toolUse"]
tool_name = tool_use["name"]
tool_input = tool_use["input"]
tool_use_id = tool_use["toolUseId"]
print(f" [Agent] Calling tool: {tool_name} with input: {json.dumps(tool_input)}")
result = invoke_tool(tool_name, tool_input)
print(f" [Agent] Tool result: {json.dumps(result)}")
audit_record["tool_calls"].append(
{"tool": tool_name, "input": tool_input, "result": result}
)
tool_results.append(
{"toolResult": {"toolUseId": tool_use_id, "content": [{"json": result}]}}
)
conversation.append({"role": "user", "content": tool_results})
return process_conversation(conversation, audit_record, recursions_left - 1)
else:
return f"Unexpected stop reason: {stop_reason}"
# --- S3 Audit Logging ---
def create_audit_bucket(bucket_name):
try:
s3_client.create_bucket(Bucket=bucket_name)
print(f"Created audit bucket: {bucket_name}")
except ClientError as e:
if e.response["Error"]["Code"] == "BucketAlreadyOwnedByYou":
print(f"Audit bucket already exists: {bucket_name}")
else:
raise
def log_to_s3(bucket_name, audit_record):
timestamp = datetime.now(timezone.utc).strftime("%Y/%m/%d/%H%M%S")
key = f"audit-logs/{timestamp}-agent-interaction.json"
s3_client.put_object(
Bucket=bucket_name,
Key=key,
Body=json.dumps(audit_record, indent=2, default=str),
ContentType="application/json",
)
print(f" [Audit] Logged to s3://{bucket_name}/{key}")
# --- Cleanup ---
def cleanup_bucket(bucket_name):
try:
objects = s3_client.list_objects_v2(Bucket=bucket_name)
if "Contents" in objects:
for obj in objects["Contents"]:
s3_client.delete_object(Bucket=bucket_name, Key=obj["Key"])
s3_client.delete_bucket(Bucket=bucket_name)
print(f"Deleted bucket: {bucket_name}")
except ClientError as e:
print(f"Cleanup error: {e}")
# --- Main ---
if __name__ == "__main__":
print("=" * 60)
print("Enterprise AI Agent - Powered by Amazon Bedrock")
print("=" * 60)
create_audit_bucket(BUCKET_NAME)
test_queries = [
"What's the status of order ORD-1234 and is that product still in stock?",
"Check if product PROD-890 is available. If not, escalate as high priority.",
"I need the status of order ORD-9999.",
]
for query in test_queries:
print(f"\n{'─' * 60}")
print(f"User: {query}")
print(f"{'─' * 60}")
response, audit = run_agent(query)
print(f"\nAgent: {response}")
log_to_s3(BUCKET_NAME, audit)
# Print cost summary
input_cost = (session_tokens["input"] / 1000) * 0.000035
output_cost = (session_tokens["output"] / 1000) * 0.00014
total_cost = input_cost + output_cost
print(f"\n{'=' * 60}")
print(f"Session Summary")
print(f" Input tokens: {session_tokens['input']}")
print(f" Output tokens: {session_tokens['output']}")
print(f" Estimated cost: ${total_cost:.6f}")
print(f"{'=' * 60}")
# Uncomment to clean up:
# cleanup_bucket(BUCKET_NAME)Your agent now produces a permanent, searchable audit trail of every decision it makes. This is the foundation of trustworthy AI in enterprise environments. Next up, you'll take on the Secret Mission and add real-time cost monitoring with budget alerts.
Secret mission
Add Token Cost Tracking and Budget Alerts
Your agent handles enterprise tasks, but what about the cost of running it? In production, unmonitored AI spending can spiral fast. Add a budget-tracking tool that the agent can call on itself to monitor token usage and trigger alerts when spending crosses a threshold.
Clean Up Your Resources
Clean Up Your Resources
Decide whether to keep your resources running or delete them entirely. This project used pay-per-token Bedrock calls (no persistent resources) and one S3 bucket that sits within the always-free tier (5GB storage, 20K GET, 2K PUT requests per month).
Resources you used:
- 1 S3 bucket (enterprise-agent-audit with your initials and date) storing audit log JSON files.
- Local project folder on your Desktop containing agent.py.
Keep everything running
No action needed. Choose this if you plan to keep experimenting with your agent.
- Your S3 bucket costs nothing within the always-free tier limits (5GB storage, 20K GET, 2K PUT requests per month).
- Bedrock charges are pay-per-token with no idle cost. You only pay when you run the script.
- AWS CLI and boto3 remain installed on your machine for future projects.
Pause - I'll come back to this later
There are no running services to pause in this project. Your S3 bucket and local files sit idle at zero cost.
- The S3 bucket holds static JSON files. There is no running process or server to stop.
- Bedrock has no persistent resources. You only incur charges when you actively call the API.
- When you are ready to return, open your project folder and run python agent.py again.
Delete - I don't want to use this again
Remove all project resources and start fresh. Follow both steps below to fully clean up.
Delete the S3 bucket
The cleanup_bucket function in your script empties and deletes the bucket. Run this Python command from inside your project folder:
python -c "import agent; agent.cleanup_bucket(agent.BUCKET_NAME)"You should see a message confirming the bucket was deleted.
Delete the local project folder
- Navigate to your Desktop and delete the project folder that contains agent.py by running:
macOS
cd ~/Desktop
rm -rf bedrock-agentWindows
cd ~\Desktop
rmdir /s /q bedrock-agentConfirm the folder is gone by checking your Desktop.
Nice Work!
Nice Work!
Nice work! You just built a fully functional AI agent that autonomously selects tools, executes them, and loops until it completes a task.
You've learned how to:
- Build an AI agent using the Bedrock Converse API with tool use (function calling) to autonomously handle enterprise support requests.
- Implement the observe-think-act agent loop, the core pattern behind every agentic AI system, with recursion handling and tool definitions using JSON Schema.
- Create enterprise-grade S3 audit logging that records every AI decision and tool call for compliance traceability.
- Secret Mission: Added real-time cost monitoring with budget alerts that track token usage across conversations.
Ready to quiz yourself?