AWS Bedrock Agent Example 1: Weather and Temperature Conversion
This guide demonstrates how to build an AI agent using AWS Bedrock that can fetch weather data and convert temperatures, based on the weather example from the LLM guide.
Table of Contents
- Overview
- Architecture
- Prerequisites
- Implementation Options
- Complete Implementation
- Testing the Agent
- Cost Considerations
Overview
We'll implement an agent that can: 1. Get current weather for a city 2. Convert temperatures between Fahrenheit and Celsius 3. Provide formatted weather information
Task Example: "What's the weather in Paris and convert the temperature to Celsius?"
Architecture
User Request
↓
Agent Orchestrator (Lambda or local)
↓
┌─────────────────────────────────────┐
│ Amazon Bedrock (Claude/Titan) │
│ - Understands request │
│ - Plans actions │
│ - Decides what tools to use │
└─────────────────────────────────────┘
↓
Action Groups / Tools
├── get_weather (Lambda → Weather API)
└── convert_temperature (Lambda)
↓
Results fed back to Bedrock
↓
Final Response to User
Prerequisites
# AWS CLI configured
aws configure
# Python dependencies
pip install boto3 requests
# Required AWS services
# - Amazon Bedrock (Claude 3 or Titan)
# - AWS Lambda (for action groups)
# - IAM roles with appropriate permissions
# External API
# - Weather API key (OpenWeatherMap, WeatherAPI, etc.)
Implementation Options
Option 1: Bedrock Agents (Managed)
AWS Bedrock Agents provides a fully managed agent framework.
Step 1: Create Action Groups (Lambda Functions)
Lambda: get_weather
import json
import os
import requests
def lambda_handler(event, context):
"""Get current weather for a city"""
# Parse input from Bedrock Agent
parameters = event.get('parameters', [])
city = next((p['value'] for p in parameters if p['name'] == 'city'), None)
# Weather API configuration (using OpenWeatherMap as example)
api_key = os.environ.get('WEATHER_API_KEY')
base_url = "http://api.openweathermap.org/data/2.5/weather"
try:
# Call weather API
response = requests.get(
base_url,
params={
'q': city,
'appid': api_key,
'units': 'imperial' # Get temperature in Fahrenheit
},
timeout=10
)
response.raise_for_status()
weather_data = response.json()
# Extract relevant information
result = {
'success': True,
'city': weather_data['name'],
'temperature': weather_data['main']['temp'],
'temperature_unit': 'Fahrenheit',
'feels_like': weather_data['main']['feels_like'],
'humidity': weather_data['main']['humidity'],
'description': weather_data['weather'][0]['description'],
'wind_speed': weather_data['wind']['speed']
}
return {
'messageVersion': '1.0',
'response': {
'actionGroup': event['actionGroup'],
'apiPath': event['apiPath'],
'httpMethod': event['httpMethod'],
'httpStatusCode': 200,
'responseBody': {
'application/json': {
'body': json.dumps(result)
}
}
}
}
except requests.exceptions.RequestException as e:
return {
'messageVersion': '1.0',
'response': {
'actionGroup': event['actionGroup'],
'apiPath': event['apiPath'],
'httpMethod': event['httpMethod'],
'httpStatusCode': 500,
'responseBody': {
'application/json': {
'body': json.dumps({
'success': False,
'error': f'Weather API error: {str(e)}'
})
}
}
}
}
except Exception as e:
return {
'messageVersion': '1.0',
'response': {
'actionGroup': event['actionGroup'],
'apiPath': event['apiPath'],
'httpMethod': event['httpMethod'],
'httpStatusCode': 500,
'responseBody': {
'application/json': {
'body': json.dumps({
'success': False,
'error': str(e)
})
}
}
}
}
Lambda: convert_temperature
import json
def lambda_handler(event, context):
"""Convert temperature between Fahrenheit and Celsius"""
# Parse input from Bedrock Agent
parameters = event.get('parameters', [])
value = float(next((p['value'] for p in parameters if p['name'] == 'value'), 0))
from_unit = next((p['value'] for p in parameters if p['name'] == 'from_unit'), 'F')
to_unit = next((p['value'] for p in parameters if p['name'] == 'to_unit'), 'C')
try:
# Perform conversion
if from_unit.upper() == 'F' and to_unit.upper() == 'C':
# Fahrenheit to Celsius
converted = (value - 32) * 5/9
result = {
'success': True,
'original_value': value,
'original_unit': 'Fahrenheit',
'converted_value': round(converted, 1),
'converted_unit': 'Celsius',
'formula': '(°F - 32) × 5/9 = °C'
}
elif from_unit.upper() == 'C' and to_unit.upper() == 'F':
# Celsius to Fahrenheit
converted = (value * 9/5) + 32
result = {
'success': True,
'original_value': value,
'original_unit': 'Celsius',
'converted_value': round(converted, 1),
'converted_unit': 'Fahrenheit',
'formula': '(°C × 9/5) + 32 = °F'
}
else:
# Same unit or invalid
result = {
'success': True,
'original_value': value,
'original_unit': from_unit,
'converted_value': value,
'converted_unit': to_unit,
'note': 'No conversion needed'
}
return {
'messageVersion': '1.0',
'response': {
'actionGroup': event['actionGroup'],
'apiPath': event['apiPath'],
'httpMethod': event['httpMethod'],
'httpStatusCode': 200,
'responseBody': {
'application/json': {
'body': json.dumps(result)
}
}
}
}
except Exception as e:
return {
'messageVersion': '1.0',
'response': {
'actionGroup': event['actionGroup'],
'apiPath': event['apiPath'],
'httpMethod': event['httpMethod'],
'httpStatusCode': 500,
'responseBody': {
'application/json': {
'body': json.dumps({
'success': False,
'error': str(e)
})
}
}
}
}
Step 2: Create OpenAPI Schema for Action Groups
weather-action-group-schema.json
{
"openapi": "3.0.0",
"info": {
"title": "Weather Tools",
"version": "1.0.0",
"description": "Tools for getting weather information and converting temperatures"
},
"paths": {
"/get-weather": {
"post": {
"summary": "Get current weather for a city",
"description": "Retrieves current weather conditions including temperature, humidity, and description",
"operationId": "getWeather",
"requestBody": {
"required": true,
"content": {
"application/json": {
"schema": {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "City name (e.g., 'Paris', 'New York', 'Tokyo')"
}
},
"required": ["city"]
}
}
}
},
"responses": {
"200": {
"description": "Weather data retrieved successfully",
"content": {
"application/json": {
"schema": {
"type": "object",
"properties": {
"success": { "type": "boolean" },
"city": { "type": "string" },
"temperature": { "type": "number" },
"temperature_unit": { "type": "string" },
"feels_like": { "type": "number" },
"humidity": { "type": "number" },
"description": { "type": "string" },
"wind_speed": { "type": "number" }
}
}
}
}
}
}
}
},
"/convert-temperature": {
"post": {
"summary": "Convert temperature between Fahrenheit and Celsius",
"description": "Converts a temperature value from one unit to another",
"operationId": "convertTemperature",
"requestBody": {
"required": true,
"content": {
"application/json": {
"schema": {
"type": "object",
"properties": {
"value": {
"type": "number",
"description": "Temperature value to convert"
},
"from_unit": {
"type": "string",
"description": "Source unit: 'F' for Fahrenheit or 'C' for Celsius",
"enum": ["F", "C", "f", "c"]
},
"to_unit": {
"type": "string",
"description": "Target unit: 'F' for Fahrenheit or 'C' for Celsius",
"enum": ["F", "C", "f", "c"]
}
},
"required": ["value", "from_unit", "to_unit"]
}
}
}
},
"responses": {
"200": {
"description": "Temperature converted successfully",
"content": {
"application/json": {
"schema": {
"type": "object",
"properties": {
"success": { "type": "boolean" },
"original_value": { "type": "number" },
"original_unit": { "type": "string" },
"converted_value": { "type": "number" },
"converted_unit": { "type": "string" },
"formula": { "type": "string" }
}
}
}
}
}
}
}
}
}
}
Step 3: Create Bedrock Agent via AWS Console or CLI
Using AWS CLI:
# Create agent
aws bedrock-agent create-agent \
--agent-name weather-assistant \
--foundation-model anthropic.claude-3-sonnet-20240229-v1:0 \
--instruction "You are a helpful weather assistant. When asked about weather:
1. Use get_weather to fetch current weather for the requested city
2. If temperature conversion is needed, use convert_temperature
3. Provide clear, friendly responses with all relevant information
Always explain what you're doing at each step." \
--agent-resource-role-arn arn:aws:iam::ACCOUNT_ID:role/BedrockAgentRole
# Create action group
aws bedrock-agent create-agent-action-group \
--agent-id AGENT_ID \
--agent-version DRAFT \
--action-group-name weather-tools \
--action-group-executor lambda=arn:aws:lambda:REGION:ACCOUNT_ID:function:get_weather \
--api-schema file://weather-action-group-schema.json
# Prepare agent
aws bedrock-agent prepare-agent --agent-id AGENT_ID
# Create alias
aws bedrock-agent create-agent-alias \
--agent-id AGENT_ID \
--agent-alias-name production
Step 4: Invoke the Agent
import boto3
import json
bedrock_agent_runtime = boto3.client('bedrock-agent-runtime')
def invoke_weather_agent(agent_id, agent_alias_id, session_id, prompt):
"""Invoke Bedrock Weather Agent"""
response = bedrock_agent_runtime.invoke_agent(
agentId=agent_id,
agentAliasId=agent_alias_id,
sessionId=session_id,
inputText=prompt
)
# Stream the response
completion = ""
for event in response.get('completion', []):
chunk = event.get('chunk')
if chunk:
completion += chunk.get('bytes').decode()
return completion
# Example usage
result = invoke_weather_agent(
agent_id='YOUR_AGENT_ID',
agent_alias_id='YOUR_ALIAS_ID',
session_id='session-456',
prompt="What's the weather in Paris and convert the temperature to Celsius?"
)
print(result)
Option 2: Agent Core with Bedrock (Custom)
Build a custom agent orchestrator using Python and Bedrock's Converse API.
Complete Agent Implementation
weather_agent.py
import boto3
import json
import re
import requests
import os
from typing import List, Dict, Any
class WeatherAgent:
"""Custom weather agent using Amazon Bedrock"""
def __init__(self, model_id: str = "anthropic.claude-3-sonnet-20240229-v1:0"):
self.bedrock = boto3.client('bedrock-runtime')
self.model_id = model_id
self.conversation_history = []
self.max_iterations = 10
self.weather_api_key = os.environ.get('WEATHER_API_KEY', 'demo_key')
# Register available tools
self.tools = {
'get_weather': self.get_weather,
'convert_temperature': self.convert_temperature
}
def get_system_prompt(self) -> str:
"""System instructions for the agent"""
return """You are a helpful weather assistant with access to weather tools.
When given a task, follow this pattern:
1. Think about what you need to do
2. Use available tools to gather information
3. Analyze the results
4. Continue until you have all needed information
Available tools:
- get_weather(city): Get current weather for a city (returns temperature in Fahrenheit)
- convert_temperature(value, from_unit, to_unit): Convert temperature between F and C
Always explain your reasoning before using a tool.
To use a tool, format your response as:
THOUGHT: [your reasoning]
ACTION: tool_name(param1="value1", param2="value2")
When you have completed the task, respond with:
FINAL_ANSWER: [your response to the user]"""
def get_weather(self, city: str) -> Dict[str, Any]:
"""Tool: Get current weather for a city"""
try:
# Using OpenWeatherMap API as example
base_url = "http://api.openweathermap.org/data/2.5/weather"
response = requests.get(
base_url,
params={
'q': city,
'appid': self.weather_api_key,
'units': 'imperial' # Fahrenheit
},
timeout=10
)
response.raise_for_status()
data = response.json()
return {
'success': True,
'city': data['name'],
'temperature': data['main']['temp'],
'temperature_unit': 'Fahrenheit',
'feels_like': data['main']['feels_like'],
'humidity': data['main']['humidity'],
'description': data['weather'][0]['description'],
'wind_speed': data['wind']['speed']
}
except Exception as e:
return {
'success': False,
'error': str(e)
}
def convert_temperature(self, value: float, from_unit: str, to_unit: str) -> Dict[str, Any]:
"""Tool: Convert temperature between Fahrenheit and Celsius"""
try:
value = float(value)
if from_unit.upper() == 'F' and to_unit.upper() == 'C':
# Fahrenheit to Celsius
converted = (value - 32) * 5/9
return {
'success': True,
'original_value': value,
'original_unit': 'Fahrenheit',
'converted_value': round(converted, 1),
'converted_unit': 'Celsius',
'formula': '(°F - 32) × 5/9 = °C'
}
elif from_unit.upper() == 'C' and to_unit.upper() == 'F':
# Celsius to Fahrenheit
converted = (value * 9/5) + 32
return {
'success': True,
'original_value': value,
'original_unit': 'Celsius',
'converted_value': round(converted, 1),
'converted_unit': 'Fahrenheit',
'formula': '(°C × 9/5) + 32 = °F'
}
else:
return {
'success': True,
'original_value': value,
'original_unit': from_unit,
'converted_value': value,
'converted_unit': to_unit,
'note': 'No conversion needed'
}
except Exception as e:
return {
'success': False,
'error': str(e)
}
def parse_action(self, text: str) -> tuple:
"""Parse ACTION from LLM response"""
action_pattern = r'ACTION:\s*(\w+)\((.*?)\)'
match = re.search(action_pattern, text, re.DOTALL)
if not match:
return None, None
tool_name = match.group(1)
params_str = match.group(2)
# Parse parameters
params = {}
# Handle string parameters
param_pattern = r'(\w+)=["\'](.*?)["\']'
for param_match in re.finditer(param_pattern, params_str):
params[param_match.group(1)] = param_match.group(2)
# Handle numeric parameters
num_pattern = r'(\w+)=([\d.]+)'
for param_match in re.finditer(num_pattern, params_str):
if param_match.group(1) not in params:
params[param_match.group(1)] = param_match.group(2)
return tool_name, params
def call_bedrock(self, messages: List[Dict]) -> str:
"""Call Bedrock Converse API"""
response = self.bedrock.converse(
modelId=self.model_id,
messages=messages,
system=[{"text": self.get_system_prompt()}],
inferenceConfig={
"temperature": 0.7,
"maxTokens": 2048
}
)
return response['output']['message']['content'][0]['text']
def run(self, user_prompt: str) -> str:
"""Main agent loop"""
# Initialize conversation
self.conversation_history = [
{
"role": "user",
"content": [{"text": user_prompt}]
}
]
iteration = 0
while iteration < self.max_iterations:
iteration += 1
print(f"\n--- Iteration {iteration} ---")
# Get LLM response
llm_response = self.call_bedrock(self.conversation_history)
print(f"LLM Response:\n{llm_response}\n")
# Add to history
self.conversation_history.append({
"role": "assistant",
"content": [{"text": llm_response}]
})
# Check for final answer
if "FINAL_ANSWER:" in llm_response:
final_answer = llm_response.split("FINAL_ANSWER:")[1].strip()
return final_answer
# Parse and execute action
tool_name, params = self.parse_action(llm_response)
if tool_name and tool_name in self.tools:
print(f"Executing: {tool_name}({params})")
# Execute tool
tool_result = self.tools[tool_name](**params)
print(f"Tool Result: {tool_result}\n")
# Add result to conversation
observation = f"OBSERVATION: {json.dumps(tool_result, indent=2)}"
self.conversation_history.append({
"role": "user",
"content": [{"text": observation}]
})
else:
# No valid action found, ask LLM to continue
self.conversation_history.append({
"role": "user",
"content": [{"text": "Please continue with your next action or provide the final answer."}]
})
return "Max iterations reached. Task may be incomplete."
# Example usage
if __name__ == "__main__":
agent = WeatherAgent()
result = agent.run(
"What's the weather in Paris and convert the temperature to Celsius?"
)
print("\n=== FINAL RESULT ===")
print(result)
Example Agent Execution Flow
User: "What's the weather in Paris and convert the temperature to Celsius?"
--- Iteration 1 ---
LLM Response:
THOUGHT: I need to get the current weather in Paris first.
ACTION: get_weather(city="Paris")
Executing: get_weather({'city': 'Paris'})
Tool Result: {
"success": true,
"city": "Paris",
"temperature": 72,
"temperature_unit": "Fahrenheit",
"feels_like": 70,
"humidity": 65,
"description": "Sunny",
"wind_speed": 5.2
}
--- Iteration 2 ---
LLM Response:
THOUGHT: I have the temperature in Fahrenheit (72°F). Now I need to convert it to Celsius.
ACTION: convert_temperature(value=72, from_unit="F", to_unit="C")
Executing: convert_temperature({'value': '72', 'from_unit': 'F', 'to_unit': 'C'})
Tool Result: {
"success": true,
"original_value": 72.0,
"original_unit": "Fahrenheit",
"converted_value": 22.2,
"converted_unit": "Celsius",
"formula": "(°F - 32) × 5/9 = °C"
}
--- Iteration 3 ---
LLM Response:
FINAL_ANSWER: The weather in Paris is sunny with a temperature of 22.2°C (72°F).
It feels like 70°F, humidity is at 65%, and wind speed is 5.2 mph.
Testing the Agent
Test Script
from weather_agent import WeatherAgent
def test_weather_agent():
"""Test the weather agent"""
agent = WeatherAgent()
# Test cases
test_cases = [
"What's the weather in Paris?",
"Convert 72 Fahrenheit to Celsius",
"What's the weather in Paris and convert the temperature to Celsius?",
"Tell me the weather in Tokyo and New York, with temperatures in Celsius"
]
for i, test in enumerate(test_cases, 1):
print(f"\n{'='*60}")
print(f"Test Case {i}: {test}")
print('='*60)
result = agent.run(test)
print(f"\nResult: {result}")
if __name__ == "__main__":
test_weather_agent()
Mock Weather API for Testing
If you don't have a weather API key, you can mock the weather service:
def get_weather_mock(self, city: str) -> Dict[str, Any]:
"""Mock weather data for testing"""
mock_data = {
'Paris': {'temp': 72, 'desc': 'Sunny'},
'Tokyo': {'temp': 68, 'desc': 'Cloudy'},
'New York': {'temp': 55, 'desc': 'Rainy'},
'London': {'temp': 60, 'desc': 'Foggy'}
}
city_data = mock_data.get(city, {'temp': 70, 'desc': 'Clear'})
return {
'success': True,
'city': city,
'temperature': city_data['temp'],
'temperature_unit': 'Fahrenheit',
'feels_like': city_data['temp'] - 2,
'humidity': 65,
'description': city_data['desc'],
'wind_speed': 5.0
}
Cost Considerations
Bedrock Pricing (Example for Claude 3 Sonnet)
Input tokens: $0.003 per 1K tokens Output tokens: $0.015 per 1K tokens
Example agent execution:
Iteration 1: 300 input + 150 output tokens (get weather)
Iteration 2: 500 input + 200 output tokens (convert temp)
Iteration 3: 700 input + 100 output tokens (final answer)
Total: 1,500 input tokens + 450 output tokens
Cost: (1.5 × $0.003) + (0.45 × $0.015) = $0.0045 + $0.0068 = $0.0113
External API Costs
OpenWeatherMap: - Free tier: 1,000 calls/day - Paid: $0.0012 per call (beyond free tier)
Total cost per query: ~$0.01 - $0.02
Advanced Features
Multi-City Weather Comparison
def compare_weather(self, cities: List[str]) -> Dict[str, Any]:
"""Compare weather across multiple cities"""
results = {}
for city in cities:
weather = self.get_weather(city)
if weather['success']:
results[city] = {
'temp_f': weather['temperature'],
'temp_c': round((weather['temperature'] - 32) * 5/9, 1),
'description': weather['description']
}
return {'success': True, 'cities': results}
# Add to tools
self.tools['compare_weather'] = self.compare_weather
Weather Alerts Integration
def get_weather_alerts(self, city: str) -> Dict[str, Any]:
"""Get weather alerts for a city"""
# Integration with weather alert APIs
# Returns severe weather warnings, advisories, etc.
pass
Summary
This guide demonstrated how to build a weather agent with AWS Bedrock that:
- Fetches real-time weather data from external APIs
- Converts temperatures between units
- Follows the ReAct pattern (Reasoning + Acting)
- Provides natural, conversational responses
The agent handles multi-step tasks by: - Understanding the user's request - Breaking it into subtasks - Executing tools in sequence - Combining results into a coherent answer
This pattern can be extended to other domains like travel booking, restaurant recommendations, or any task requiring multiple API calls and data transformations.