89 lines
2.8 KiB
Python
89 lines
2.8 KiB
Python
import logging
|
|
import json
|
|
from typing import Dict, Any
|
|
from http import HTTPStatus
|
|
import urllib.request
|
|
|
|
logger = logging.getLogger()
|
|
logger.setLevel(logging.INFO)
|
|
|
|
def lambda_handler(event: Dict[str, Any], context: Any) -> Dict[str, Any]:
|
|
"""
|
|
AWS Lambda handler for processing Bedrock agent requests and geocoding location names using Nominatim.
|
|
|
|
Args:
|
|
event (Dict[str, Any]): The Lambda event containing action details
|
|
context (Any): The Lambda context object
|
|
|
|
Returns:
|
|
Dict[str, Any]: Response formatted for Bedrock Agents
|
|
"""
|
|
try:
|
|
action_group = event['actionGroup']
|
|
function = event['function']
|
|
message_version = event.get('messageVersion', 1)
|
|
parameters = event.get('parameters', [])
|
|
print(parameters)
|
|
parameters_dict ={parameter["name"]:parameter["value"] for parameter in parameters}
|
|
location_name = parameters_dict.get('location_name')
|
|
if not location_name:
|
|
raise KeyError("Missing required parameter: 'location_name'")
|
|
|
|
# Call Nominatim API to geocode the location
|
|
query = urllib.parse.urlencode({
|
|
"q": location_name,
|
|
"format": "json",
|
|
"limit": 1
|
|
})
|
|
headers = {
|
|
"User-Agent": "aws-lambda-geocoder/1.0"
|
|
}
|
|
url = f"https://nominatim.openstreetmap.org/search?{query}"
|
|
|
|
req = urllib.request.Request(url, headers=headers)
|
|
with urllib.request.urlopen(req) as response:
|
|
response_data = response.read()
|
|
print(response_data)
|
|
results = json.loads(response_data)
|
|
if not results:
|
|
raise ValueError(f"No coordinates found for '{location_name}'")
|
|
|
|
lat = float(results[0]["lat"])
|
|
lon = float(results[0]["lon"])
|
|
|
|
# Prepare Bedrock-compatible response
|
|
response_body = {
|
|
'TEXT': {
|
|
'body': f"Coordinates for '{location_name}': Latitude {lat}, Longitude {lon}"
|
|
}
|
|
}
|
|
|
|
action_response = {
|
|
'actionGroup': action_group,
|
|
'function': function,
|
|
'functionResponse': {
|
|
'responseBody': response_body
|
|
}
|
|
}
|
|
|
|
final_response = {
|
|
'response': action_response,
|
|
'messageVersion': message_version
|
|
}
|
|
|
|
logger.info('Success: %s', final_response)
|
|
return final_response
|
|
|
|
except KeyError as e:
|
|
logger.error('Missing required field: %s', str(e))
|
|
return {
|
|
'statusCode': HTTPStatus.BAD_REQUEST,
|
|
'body': f'Error: {str(e)}'
|
|
}
|
|
except Exception as e:
|
|
logger.error('Unexpected error: %s', str(e))
|
|
return {
|
|
'statusCode': HTTPStatus.INTERNAL_SERVER_ERROR,
|
|
'body': f'Internal server error: {str(e)}'
|
|
}
|