The Rise of AI Agents on AWS: From Chatbots to Autonomous Systems

# lambda_tool_handler.py — The tool executor

import json
import boto3
import requests
from datetime import datetime

def lambda_handler(event, context):
"""
Handles all tool calls from the Bedrock Agent.
The agent sends an event with actionGroup, function, and parameters.
"""
action_group = event.get('actionGroup')
function = event.get('function')
parameters = event.get('parameters', [])

# Convert parameters list to dict
params = {p['name']: p['value'] for p in parameters}

print(f"Agent calling: {action_group}/{function} with {params}")

if function == 'get_weather':
result = get_weather(params.get('city'))
elif function == 'search_database':
result = search_database(params.get('query'), params.get('limit', 10))
elif function == 'create_ticket':
result = create_support_ticket(params)
elif function == 'get_account_info':
result = get_account_info(params.get('account_id'))
else:
result = {"error": f"Unknown function: {function}"}

# CRITICAL: Response must follow this exact format for Bedrock Agents
return {
"messageVersion": "1.0",
"response": {
"actionGroup": action_group,
"function": function,
"functionResponse": {
"responseBody": {
"TEXT": {
"body": json.dumps(result)
}
}
}
}
}

def get_weather(city: str) -> dict:
"""Fetch weather for a city using a public API."""
# In production, use your weather API key
return {
"city": city,
"temperature": "22°C",
"condition": "Partly Cloudy",
"humidity": "65%",
"timestamp": datetime.utcnow().isoformat()
}

def search_database(query: str, limit: int) -> dict:
"""Search DynamoDB for relevant records."""
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('CustomerRecords')

# Simple scan with filter (use proper indexing in production)
response = table.scan(
FilterExpression='contains(#data, :query)',
ExpressionAttributeNames={'#data': 'searchable_text'},
ExpressionAttributeValues={':query': query},
Limit=limit
)

return {
"results": response.get('Items', []),
"count": len(response.get('Items', []))
}

def create_support_ticket(params: dict) -> dict:
"""Create a support ticket in the system."""
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('SupportTickets')

ticket_id = f"TKT-{datetime.utcnow().strftime('%Y%m%d%H%M%S')}"

table.put_item(Item={
'ticket_id': ticket_id,
'title': params.get('title'),
'description': params.get('description'),
'priority': params.get('priority', 'MEDIUM'),
'status': 'OPEN',
'created_at': datetime.utcnow().isoformat()
})

return {"ticket_id": ticket_id, "status": "OPEN", "message": "Ticket created successfully"}

def get_account_info(account_id: str) -> dict:
"""Retrieve account information."""
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('Accounts')

response = table.get_item(Key={'account_id': account_id})
item = response.get('Item')

if not item:
return {"error": f"Account {account_id} not found"}

# Remove sensitive fields
item.pop('payment_info', None)
item.pop('ssn', None)

return item

# deploy_agent.py — Create and configure the Bedrock Agent

import boto3
import json
import time

bedrock_agent = boto3.client('bedrock-agent', region_name='us-east-1')

AGENT_ROLE_ARN = "arn:aws:iam::123456789012:role/BedrockAgentExecutionRole"
LAMBDA_ARN = "arn:aws:lambda:us-east-1:123456789012:function:AgentToolHandler"

def create_agent():
"""Create the Bedrock Agent."""
response = bedrock_agent.create_agent(
agentName='CustomerSupportAgent',
agentResourceRoleArn=AGENT_ROLE_ARN,
description='Autonomous customer support agent with tool use',

# Use Claude 3 Sonnet for the best balance of speed and intelligence
foundationModel='anthropic.claude-3-5-sonnet-20241022-v2:0',

instruction="""You are an expert customer support agent for Acme Corp.

Your capabilities:
- Look up customer account information
- Check weather for shipping estimates
- Search the knowledge base for solutions
- Create support tickets when issues can't be resolved immediately

Guidelines:
- Always verify the customer's account before providing sensitive information
- Be empathetic and professional
- Escalate to a human if the issue is complex or the customer is distressed
- Always provide a ticket number when creating tickets
- If you don't know something, say so — never guess

You have access to tools. Use them proactively to help customers efficiently.""",

idleSessionTTLInSeconds=1800, # 30-minute session timeout
)

agent_id = response['agent']['agentId']
print(f"Created Agent ID: {agent_id}")
return agent_id

def create_action_group(agent_id: str):
"""Define the tools the agent can use."""

# OpenAPI schema defining available functions
api_schema = {
"openapi": "3.0.0",
"info": {"title": "Customer Support Tools", "version": "1.0.0"},
"paths": {
"/get_account_info": {
"get": {
"operationId": "get_account_info",
"description": "Retrieve customer account details by account ID",
"parameters": [
{
"name": "account_id",
"in": "query",
"required": True,
"schema": {"type": "string"},
"description": "The unique customer account identifier"
}
],
"responses": {"200": {"description": "Account information"}}
}
},
"/search_database": {
"get": {
"operationId": "search_database",
"description": "Search the customer database for relevant records",
"parameters": [
{
"name": "query",
"in": "query",
"required": True,
"schema": {"type": "string"}
},
{
"name": "limit",
"in": "query",
"required": False,
"schema": {"type": "integer", "default": 10}
}
],
"responses": {"200": {"description": "Search results"}}
}
},
"/create_ticket": {
"post": {
"operationId": "create_ticket",
"description": "Create a new support ticket for tracking an issue",
"requestBody": {
"content": {
"application/json": {
"schema": {
"type": "object",
"properties": {
"title": {"type": "string"},
"description": {"type": "string"},
"priority": {
"type": "string",
"enum": ["LOW", "MEDIUM", "HIGH", "CRITICAL"]
}
},
"required": ["title", "description"]
}
}
}
},
"responses": {"200": {"description": "Created ticket"}}
}
}
}
}

response = bedrock_agent.create_agent_action_group(
agentId=agent_id,
agentVersion='DRAFT',
actionGroupName='CustomerSupportActions',
description='Tools for customer support operations',
actionGroupExecutor={
'lambda': LAMBDA_ARN
},
apiSchema={
'payload': json.dumps(api_schema)
}
)

print(f"Created Action Group: {response['agentActionGroup']['actionGroupId']}")

def prepare_and_deploy(agent_id: str):
"""Prepare agent for testing and create an alias."""

# Prepare the agent (validate and compile)
bedrock_agent.prepare_agent(agentId=agent_id)
time.sleep(15) # Wait for preparation

# Create an alias for the agent version
alias = bedrock_agent.create_agent_alias(
agentId=agent_id,
agentAliasName='production',
description='Production deployment'
)

alias_id = alias['agentAlias']['agentAliasId']
print(f"Agent Alias ID: {alias_id}")
print(f"\nDeploy complete!")
print(f"Agent ID: {agent_id}")
print(f"Alias ID: {alias_id}")
return alias_id

# Run deployment
agent_id = create_agent()
create_action_group(agent_id)
alias_id = prepare_and_deploy(agent_id)

# knowledge_base_setup.py

import boto3
import json

bedrock_agent = boto3.client('bedrock-agent', region_name='us-east-1')

def create_knowledge_base():
"""
Create a vector knowledge base backed by Amazon OpenSearch Serverless.
Documents in S3 are automatically chunked, embedded, and indexed.
"""

response = bedrock_agent.create_knowledge_base(
name='ProductKnowledgeBase',
description='Product documentation and support articles',
roleArn='arn:aws:iam::123456789012:role/BedrockKBRole',

knowledgeBaseConfiguration={
'type': 'VECTOR',
'vectorKnowledgeBaseConfiguration': {
# Use Amazon's Titan embedding model
'embeddingModelArn': 'arn:aws:bedrock:us-east-1::foundation-model/amazon.titan-embed-text-v2:0'
}
},

storageConfiguration={
'type': 'OPENSEARCH_SERVERLESS',
'opensearchServerlessConfiguration': {
'collectionArn': 'arn:aws:aoss:us-east-1:123456789012:collection/my-kb-collection',
'vectorIndexName': 'bedrock-knowledge-base-default-index',
'fieldMapping': {
'vectorField': 'embedding',
'textField': 'AMAZON_BEDROCK_TEXT_CHUNK',
'metadataField': 'AMAZON_BEDROCK_METADATA'
}
}
}
)

kb_id = response['knowledgeBase']['knowledgeBaseId']
print(f"Knowledge Base ID: {kb_id}")
return kb_id

def create_data_source(kb_id: str, bucket_name: str):
"""Connect an S3 bucket as the document source."""

response = bedrock_agent.create_data_source(
knowledgeBaseId=kb_id,
name='ProductDocs-S3',
description='Product documentation from S3',

dataSourceConfiguration={
'type': 'S3',
's3Configuration': {
'bucketArn': f'arn:aws:s3:::{bucket_name}',
'inclusionPrefixes': ['docs/', 'articles/', 'manuals/']
}
},

vectorIngestionConfiguration={
'chunkingConfiguration': {
'chunkingStrategy': 'HIERARCHICAL',
'hierarchicalChunkingConfiguration': {
'levelConfigurations': [
{'maxTokens': 1500}, # Parent chunks
{'maxTokens': 300} # Child chunks (retrieved)
],
'overlapTokens': 60
}
}
}
)

ds_id = response['dataSource']['dataSourceId']
print(f"Data Source ID: {ds_id}")
return ds_id

def sync_knowledge_base(kb_id: str, ds_id: str):
"""Trigger ingestion — parse, chunk, embed, and index all documents."""

response = bedrock_agent.start_ingestion_job(
knowledgeBaseId=kb_id,
dataSourceId=ds_id
)

job_id = response['ingestionJob']['ingestionJobId']
print(f"Ingestion started. Job ID: {job_id}")
print("Monitor in the AWS Console or poll the ingestion job status.")
return job_id

# Direct KB query (without agent)
def query_knowledge_base(kb_id: str, question: str):
"""Query the KB directly for testing."""

bedrock_rt = boto3.client('bedrock-agent-runtime', region_name='us-east-1')

response = bedrock_rt.retrieve_and_generate(
input={'text': question},
retrieveAndGenerateConfiguration={
'type': 'KNOWLEDGE_BASE',
'knowledgeBaseConfiguration': {
'knowledgeBaseId': kb_id,
'modelArn': 'arn:aws:bedrock:us-east-1::foundation-model/anthropic.claude-3-5-sonnet-20241022-v2:0',
'retrievalConfiguration': {
'vectorSearchConfiguration': {
'numberOfResults': 5,
'overrideSearchType': 'HYBRID' # Combines semantic + keyword search
}
}
}
}
)

return {
'answer': response['output']['text'],
'citations': response.get('citations', [])
}

# multi_agent_system.py
# Amazon Bedrock now supports native multi-agent collaboration

import boto3
import json
import uuid

bedrock_agent = boto3.client('bedrock-agent', region_name='us-east-1')
bedrock_rt = boto3.client('bedrock-agent-runtime', region_name='us-east-1')

# --- Step 1: Create Sub-Agents ---

def create_data_analyst_agent():
"""Specialist in querying and analyzing data."""
response = bedrock_agent.create_agent(
agentName='DataAnalystAgent',
agentResourceRoleArn='arn:aws:iam::123456789012:role/BedrockAgentRole',
foundationModel='anthropic.claude-3-5-sonnet-20241022-v2:0',
instruction="""You are a data analyst specialist.
Your job is to query databases, run SQL, analyze datasets, and return
structured results. Always return data in JSON format.
Be precise and focus only on data analysis tasks."""
)
return response['agent']['agentId']

def create_code_writer_agent():
"""Specialist in writing and executing code."""
response = bedrock_agent.create_agent(
agentName='CodeWriterAgent',
agentResourceRoleArn='arn:aws:iam::123456789012:role/BedrockAgentRole',
foundationModel='anthropic.claude-3-5-sonnet-20241022-v2:0',
instruction="""You are a Python expert and software engineer.
Write clean, well-documented code. Always include error handling.
When asked to analyze data, write efficient Pandas/NumPy code.
Return code results in structured JSON."""
)
return response['agent']['agentId']

# --- Step 2: Create Supervisor Agent with Sub-Agent Collaboration ---

def create_supervisor_agent(analyst_agent_alias_arn: str, code_agent_alias_arn: str):
"""Orchestrator that delegates to specialist sub-agents."""

response = bedrock_agent.create_agent(
agentName='SupervisorAgent',
agentResourceRoleArn='arn:aws:iam::123456789012:role/BedrockAgentRole',
foundationModel='anthropic.claude-3-5-sonnet-20241022-v2:0',
instruction="""You are a business intelligence supervisor.

When given a complex task, break it down and delegate to specialists:
- For data queries and analysis → use the DataAnalystAgent
- For code writing and execution → use the CodeWriterAgent

Synthesize results from sub-agents into a clear, actionable response.
Always cite which sub-agent produced which insights.""",

# Enable multi-agent collaboration
agentCollaboration='SUPERVISOR'
)

supervisor_id = response['agent']['agentId']

# Associate sub-agents with the supervisor
bedrock_agent.associate_agent_collaborator(
agentId=supervisor_id,
agentVersion='DRAFT',
agentDescriptor={'aliasArn': analyst_agent_alias_arn},
collaboratorName='DataAnalystAgent',
collaborationInstruction='Use for data queries, SQL, and statistical analysis tasks.'
)

bedrock_agent.associate_agent_collaborator(
agentId=supervisor_id,
agentVersion='DRAFT',
agentDescriptor={'aliasArn': code_agent_alias_arn},
collaboratorName='CodeWriterAgent',
collaborationInstruction='Use for writing Python code, data processing scripts, and automation.'
)

return supervisor_id

# --- Step 3: Invoke the Multi-Agent System ---

def run_multi_agent_task(task: str, supervisor_id: str, alias_id: str):
"""Send a complex task to the supervisor agent system."""

session_id = str(uuid.uuid4())

response = bedrock_rt.invoke_agent(
agentId=supervisor_id,
agentAliasId=alias_id,
sessionId=session_id,
inputText=task,
enableTrace=True
)

full_response = ""
agent_calls = []

for event in response['completion']:
if 'chunk' in event:
full_response += event['chunk']['bytes'].decode('utf-8')

if 'trace' in event:
trace = event['trace']['trace']
# Track sub-agent invocations
if 'orchestrationTrace' in trace:
orch = trace['orchestrationTrace']
if 'invocationInput' in orch:
inv = orch['invocationInput']
if 'agentCollaboratorInvocationInput' in inv:
collab = inv['agentCollaboratorInvocationInput']
agent_calls.append(collab.get('agentCollaboratorName'))
print(f"Delegating to: {collab.get('agentCollaboratorName')}")

return full_response, agent_calls

# Example usage
task = """
Analyze our Q4 2024 sales data:
1. Find the top 5 products by revenue
2. Write a Python script to generate a trend chart
3. Identify any anomalies in the data
Provide a summary suitable for an executive presentation.
"""

result, agents_used = run_multi_agent_task(task, "supervisor-id", "alias-id")
print(f"\nResult: {result}")
print(f"Sub-agents used: {agents_used}")

# observability.py — Production monitoring for AI agents

import boto3
import json
from datetime import datetime, timedelta

cloudwatch = boto3.client('cloudwatch')

def publish_agent_metrics(agent_id: str, session_id: str, metrics: dict):
"""Publish custom agent performance metrics to CloudWatch."""

cloudwatch.put_metric_data(
Namespace='AIAgents/Bedrock',
MetricData=[
{
'MetricName': 'AgentLatencyMs',
'Dimensions': [
{'Name': 'AgentId', 'Value': agent_id},
{'Name': 'Environment', 'Value': 'production'}
],
'Value': metrics['latency_ms'],
'Unit': 'Milliseconds',
'Timestamp': datetime.utcnow()
},
{
'MetricName': 'ToolInvocations',
'Dimensions': [
{'Name': 'AgentId', 'Value': agent_id},
{'Name': 'ToolName', 'Value': metrics.get('tool_name', 'unknown')}
],
'Value': metrics.get('tool_count', 1),
'Unit': 'Count',
'Timestamp': datetime.utcnow()
},
{
'MetricName': 'InputTokens',
'Dimensions': [{'Name': 'AgentId', 'Value': agent_id}],
'Value': metrics.get('input_tokens', 0),
'Unit': 'Count',
'Timestamp': datetime.utcnow()
},
{
'MetricName': 'OutputTokens',
'Dimensions': [{'Name': 'AgentId', 'Value': agent_id}],
'Value': metrics.get('output_tokens', 0),
'Unit': 'Count',
'Timestamp': datetime.utcnow()
}
]
)

def create_agent_alarms(agent_id: str, sns_topic_arn: str):
"""Create CloudWatch alarms for agent health monitoring."""

alarms = [
{
'AlarmName': f'AgentHighLatency-{agent_id}',
'AlarmDescription': 'Agent P95 latency exceeds 30 seconds',
'MetricName': 'AgentLatencyMs',
'Statistic': 'p95',
'Threshold': 30000,
'ComparisonOperator': 'GreaterThanThreshold',
'EvaluationPeriods': 3,
'Period': 300, # 5 minutes
},
{
'AlarmName': f'AgentHighToolFailure-{agent_id}',
'AlarmDescription': 'Tool invocation failure rate > 5%',
'MetricName': 'ToolInvocationErrors',
'Statistic': 'Sum',
'Threshold': 10,
'ComparisonOperator': 'GreaterThanThreshold',
'EvaluationPeriods': 1,
'Period': 300,
}
]

for alarm in alarms:
cloudwatch.put_metric_alarm(
**alarm,
Namespace='AIAgents/Bedrock',
Dimensions=[{'Name': 'AgentId', 'Value': agent_id}],
AlarmActions=[sns_topic_arn],
OKActions=[sns_topic_arn],
TreatMissingData='notBreaching'
)
print(f"Created alarm: {alarm['AlarmName']}")

def get_agent_dashboard_data(agent_id: str, hours: int = 24):
"""Retrieve agent performance metrics for a dashboard."""

end_time = datetime.utcnow()
start_time = end_time - timedelta(hours=hours)

metrics_to_fetch = ['AgentLatencyMs', 'ToolInvocations', 'InputTokens', 'OutputTokens']
dashboard_data = {}

for metric_name in metrics_to_fetch:
response = cloudwatch.get_metric_statistics(
Namespace='AIAgents/Bedrock',
MetricName=metric_name,
Dimensions=[{'Name': 'AgentId', 'Value': agent_id}],
StartTime=start_time,
EndTime=end_time,
Period=3600, # 1-hour buckets
Statistics=['Average', 'Sum', 'Maximum']
)

dashboard_data[metric_name] = sorted(
response['Datapoints'],
key=lambda x: x['Timestamp']
)

return dashboard_data