| name | bedrock-agents |
| description | Amazon Bedrock Agents for building autonomous AI agents with foundation model orchestration, action groups, knowledge bases, and session management. Use when creating AI agents, orchestrating multi-step workflows, integrating tools with LLMs, building conversational agents, implementing RAG patterns, managing agent sessions, deploying production agents, or connecting knowledge bases to agents. |
| allowed-tools | ["Bash","Read","Write","Edit","Glob","Grep"] |
Amazon Bedrock Agents
Complete guide to building and managing Amazon Bedrock Agents - autonomous AI agents that orchestrate foundation models with action groups, knowledge bases, and multi-turn conversations.
Overview
Amazon Bedrock Agents enables you to create autonomous AI agents that can:
- Orchestrate foundation models to execute multi-step tasks
- Integrate with APIs and Lambda functions via action groups
- Access enterprise knowledge through knowledge bases (RAG)
- Maintain conversation context across sessions
- Reason about complex problems and break them into steps
- Invoke tools and APIs based on natural language requests
Control Plane: bedrock-agent client for agent configuration
Runtime Plane: bedrock-agent-runtime client for agent invocation
Core Concepts
Agents
The foundation model orchestrator that processes requests, plans actions, and coordinates responses.
Key Properties:
- Agent name and description
- Foundation model (Claude, Titan, etc.)
- Instructions (system prompt defining behavior)
- Action groups (tools the agent can use)
- Knowledge bases (data sources for RAG)
- Aliases (deployment versions)
Action Groups
Define tools and APIs the agent can invoke to accomplish tasks.
Types:
- Lambda functions (direct AWS Lambda invocation)
- OpenAPI schemas (REST API integration)
- Code interpreter (Python code execution)
- Return of control (return to application)
Knowledge Bases
Vector databases with RAG capabilities that agents can query.
Integration:
- Associate knowledge bases with agents
- Agent automatically retrieves relevant documents
- Foundation model synthesizes answers from retrieved content
Aliases
Versioned endpoints for agent deployment.
Benefits:
- Stable endpoint for applications
- Safe deployment with version control
- Routing between agent versions
- Production/staging environments
Sessions
Maintain conversation context across multiple invocations.
Session Management:
- Session IDs track conversations
- Memory stores chat history
- Context persists between turns
- Delete memory to reset conversations
Architecture
┌─────────────────────────────────────────────────────────────┐
│ Application Layer │
│ (Your code using bedrock-agent-runtime.invoke_agent) │
└───────────────────────────┬─────────────────────────────────┘
│
┌───────────────────────────▼─────────────────────────────────┐
│ Bedrock Agent Runtime │
│ - Session management │
│ - Streaming responses │
│ - Trace/reasoning visibility │
└───────┬──────────────────┬──────────────────┬───────────────┘
│ │ │
┌───────▼────────┐ ┌─────▼──────┐ ┌────────▼──────────┐
│ Foundation │ │ Action │ │ Knowledge │
│ Model │ │ Groups │ │ Bases │
│ (Claude, etc.) │ │ (Lambda, │ │ (Vector DB + │
│ │ │ OpenAPI) │ │ Documents) │
└────────────────┘ └────────────┘ └───────────────────┘
Operations
1. Create Agent
Create a new Bedrock Agent with foundation model and instructions.
boto3 Example:
import boto3
import json
bedrock_agent = boto3.client('bedrock-agent', region_name='us-east-1')
def create_agent(
agent_name: str,
foundation_model: str = "anthropic.claude-3-5-sonnet-20241022-v2:0",
instructions: str = None,
description: str = None,
idle_session_ttl: int = 600
) -> dict:
"""
Create a Bedrock Agent.
Args:
agent_name: Name of the agent
foundation_model: Model ID (Claude 3.5 Sonnet, Haiku, etc.)
instructions: System prompt defining agent behavior
description: Human-readable description
idle_session_ttl: Session timeout in seconds (default 600)
Returns:
Agent details including agentId, agentArn, agentStatus
"""
if instructions is None:
instructions = """You are a helpful AI assistant.
Follow user requests carefully and provide accurate, helpful responses.
When you need to use tools or access knowledge, do so to provide better answers."""
try:
response = bedrock_agent.create_agent(
agentName=agent_name,
foundationModel=foundation_model,
instruction=instructions,
description=description or f"Bedrock Agent: {agent_name}",
idleSessionTTLInSeconds=idle_session_ttl,
tags={
'Environment': 'production',
'ManagedBy': 'bedrock-agents-skill'
}
)
agent = response['agent']
print()
()
()
()
agent
Exception e:
()
agent = create_agent(
agent_name=,
foundation_model=,
instructions=,
description=,
idle_session_ttl=
)
agent_id = agent[]
Foundation Models:
anthropic.claude-3-5-sonnet-20241022-v2:0 - Best reasoninganthropic.claude-3-5-haiku-20241022-v1:0 - Fast responsesanthropic.claude-3-opus-20240229-v1:0 - Maximum capabilityamazon.titan-text-premier-v1:0 - AWS native model
2. Prepare Agent
Prepare the agent for use (required after creation or updates).
def prepare_agent(agent_id: str) -> dict:
"""
Prepare agent for use. Required after creation or configuration changes.
Args:
agent_id: ID of the agent to prepare
Returns:
Prepared agent details
"""
try:
response = bedrock_agent.prepare_agent(agentId=agent_id)
print(f"✓ Agent preparation started")
print(f" Agent ID: {agent_id}")
print(f" Status: {response['agentStatus']}")
print(f" Prepared At: {response['preparedAt']}")
return response
except Exception as e:
print(f"✗ Failed to prepare agent: {str(e)}")
raise
prepare_agent(agent_id)
import time
def wait_for_agent_ready(agent_id: str, max_wait: int = 120):
"""Wait for agent to be in PREPARED or VERSIONED status."""
start_time = time.time()
while time.time() - start_time < max_wait:
response = bedrock_agent.get_agent(agentId=agent_id)
status = response['agent']['agentStatus']
if status in ['PREPARED', ]:
()
status == :
()
()
time.sleep()
()
wait_for_agent_ready(agent_id)
3. Create Action Group
Add Lambda or OpenAPI action groups to enable tool use.
Lambda Action Group:
def create_lambda_action_group(
agent_id: str,
action_group_name: str,
lambda_arn: str,
description: str,
api_schema: dict
) -> dict:
"""
Create an action group that invokes Lambda functions.
Args:
agent_id: ID of the agent
action_group_name: Name of the action group
lambda_arn: ARN of the Lambda function to invoke
description: Description of what this action group does
api_schema: OpenAPI schema defining available operations
Returns:
Action group details
"""
try:
response = bedrock_agent.create_agent_action_group(
agentId=agent_id,
agentVersion='DRAFT',
actionGroupName=action_group_name,
description=description,
actionGroupExecutor={
'lambda': lambda_arn
},
apiSchema={
'payload': json.dumps(api_schema)
},
actionGroupState='ENABLED'
)
action_group = response['agentActionGroup']
print(f"✓ Created action group: {action_group['actionGroupName']}")
print(f" Action Group ID: {action_group['actionGroupId']}")
print(f" Lambda: {lambda_arn}")
return action_group
except Exception as e:
print(f"✗ Failed to create action group: {str(e)}")
raise
order_lookup_schema = {
"openapi": ,
: {
: ,
: ,
:
},
: {
: {
: {
: ,
: ,
: ,
: [
{
: ,
: ,
: ,
: ,
: {
:
}
}
],
: {
: {
: ,
: {
: {
: {
: ,
: {
: {: },
: {: },
: {: },
: {: }
}
}
}
}
}
}
}
},
: {
: {
: ,
: ,
: ,
: {
: ,
: {
: {
: {
: ,
: {
: {: },
: {: , : },
: {: , : }
}
}
}
}
},
: {
: {
:
}
}
}
}
}
}
action_group = create_lambda_action_group(
agent_id=agent_id,
action_group_name=,
lambda_arn=,
description=,
api_schema=order_lookup_schema
)
OpenAPI Action Group (External REST API):
def create_openapi_action_group(
agent_id: str,
action_group_name: str,
api_schema: dict,
description: str
) -> dict:
"""
Create an action group for external REST APIs.
Agent will make HTTP requests to the API defined in the schema.
"""
try:
response = bedrock_agent.create_agent_action_group(
agentId=agent_id,
agentVersion='DRAFT',
actionGroupName=action_group_name,
description=description,
actionGroupExecutor={
'customControl': 'RETURN_CONTROL'
},
apiSchema={
'payload': json.dumps(api_schema)
},
actionGroupState='ENABLED'
)
return response['agentActionGroup']
except Exception as e:
print(f"✗ Failed to create OpenAPI action group: {str(e)}")
raise
4. Associate Knowledge Base
Connect a knowledge base to enable RAG capabilities.
def associate_knowledge_base(
agent_id: str,
knowledge_base_id: str,
description: str = "Agent knowledge base"
) -> dict:
"""
Associate a knowledge base with an agent for RAG.
Args:
agent_id: ID of the agent
knowledge_base_id: ID of the knowledge base to associate
description: Description of the knowledge base purpose
Returns:
Association details
"""
try:
response = bedrock_agent.associate_agent_knowledge_base(
agentId=agent_id,
agentVersion='DRAFT',
knowledgeBaseId=knowledge_base_id,
description=description,
knowledgeBaseState='ENABLED'
)
kb = response['agentKnowledgeBase']
print(f"✓ Associated knowledge base")
print(f" Knowledge Base ID: {kb['knowledgeBaseId']}")
print(f" Description: {kb['description']}")
return kb
except Exception as e:
print(f"✗ Failed to associate knowledge base: {str(e)}")
raise
kb_association = associate_knowledge_base(
agent_id=agent_id,
knowledge_base_id="KB123456789",
description="Product catalog and FAQ knowledge base"
)
prepare_agent(agent_id)
wait_for_agent_ready(agent_id)
5. Create Agent Alias
Create a stable endpoint for agent invocation.
def create_agent_alias(
agent_id: str,
alias_name: str,
description: str = None
) -> dict:
"""
Create an agent alias for deployment.
Args:
agent_id: ID of the agent
alias_name: Name for the alias (e.g., 'production', 'staging')
description: Description of the alias
Returns:
Alias details including agentAliasId
"""
try:
response = bedrock_agent.create_agent_alias(
agentId=agent_id,
agentAliasName=alias_name,
description=description or f"{alias_name} environment"
)
alias = response['agentAlias']
print(f"✓ Created alias: {alias['agentAliasName']}")
print(f" Alias ID: {alias['agentAliasId']}")
print(f" Status: {alias['agentAliasStatus']}")
return alias
except Exception as e:
print(f"✗ Failed to create alias: {str(e)}")
raise
alias = create_agent_alias(
agent_id=agent_id,
alias_name="production",
description="Production environment for customer support agent"
)
agent_alias_id = alias['agentAliasId']
def wait_for_alias_ready(agent_id: , alias_id: , max_wait: = ):
start_time = time.time()
time.time() - start_time < max_wait:
response = bedrock_agent.get_agent_alias(
agentId=agent_id,
agentAliasId=alias_id
)
status = response[][]
status == :
()
status == :
()
()
time.sleep()
wait_for_alias_ready(agent_id, agent_alias_id)
6. Invoke Agent (Runtime)
Invoke the agent with single-turn or multi-turn conversations.
Single-Turn Invocation:
import boto3
bedrock_agent_runtime = boto3.client('bedrock-agent-runtime', region_name='us-east-1')
def invoke_agent(
agent_id: str,
agent_alias_id: str,
session_id: str,
input_text: str,
enable_trace: bool = True
) -> dict:
"""
Invoke a Bedrock Agent with streaming response.
Args:
agent_id: ID of the agent
agent_alias_id: ID of the agent alias
session_id: Unique session ID for conversation tracking
input_text: User's input text
enable_trace: Enable trace for reasoning visibility
Returns:
Complete response with text, traces, and citations
"""
try:
response = bedrock_agent_runtime.invoke_agent(
agentId=agent_id,
agentAliasId=agent_alias_id,
sessionId=session_id,
inputText=input_text,
enableTrace=enable_trace
)
result = {
'completion': '',
'traces': [],
'citations': []
}
event_stream = response['completion']
for event in event_stream:
if 'chunk' in event:
chunk = event['chunk']
if 'bytes' in chunk:
text = chunk['bytes'].decode('utf-8')
result['completion'] += text
print(text, end='', flush=True)
elif event:
trace = event[][]
result[].append(trace)
enable_trace:
trace:
orch = trace[]
orch:
()
orch:
()
event:
result[] = event[]
()
result
Exception e:
()
uuid
session_id = (uuid.uuid4())
response = invoke_agent(
agent_id=agent_id,
agent_alias_id=agent_alias_id,
session_id=session_id,
input_text=,
enable_trace=
)
()
Multi-Turn Conversation:
def chat_with_agent(
agent_id: str,
agent_alias_id: str,
enable_trace: bool = False
):
"""
Interactive chat session with the agent.
Maintains conversation context across turns.
"""
session_id = str(uuid.uuid4())
print(f"Chat session started (ID: {session_id})")
print("Type 'exit' to end the conversation\n")
while True:
user_input = input("You: ").strip()
if user_input.lower() in ['exit', 'quit', 'bye']:
print("Ending conversation...")
break
if not user_input:
continue
print("\nAgent: ", end='', flush=True)
response = invoke_agent(
agent_id=agent_id,
agent_alias_id=agent_alias_id,
session_id=session_id,
input_text=user_input,
enable_trace=enable_trace
)
print()
Invoke with Session State:
def invoke_agent_with_state(
agent_id: str,
agent_alias_id: str,
session_id: str,
input_text: str,
session_state: dict = None
) -> dict:
"""
Invoke agent with custom session state.
Session state can include:
- promptSessionAttributes: Dynamic variables for prompts
- sessionAttributes: Persistent session data
"""
invoke_params = {
'agentId': agent_id,
'agentAliasId': agent_alias_id,
'sessionId': session_id,
'inputText': input_text
}
if session_state:
invoke_params['sessionState'] = session_state
response = bedrock_agent_runtime.invoke_agent(**invoke_params)
result = {'completion': ''}
for event in response['completion']:
if 'chunk' in event and 'bytes' in event['chunk']:
result['completion'] += event['chunk']['bytes'].decode('utf-8')
return result
response = invoke_agent_with_state(
agent_id=agent_id,
agent_alias_id=agent_alias_id,
session_id=session_id,
input_text="What's my order status?",
session_state={
'sessionAttributes': {
'customer_id': 'CUST-12345',
'customer_tier': ,
:
},
: {
: ,
:
}
}
)
7. Delete Agent Memory
Clear conversation memory for a session.
def delete_agent_memory(
agent_id: str,
agent_alias_id: str,
session_id: str
) -> dict:
"""
Delete conversation memory for a specific session.
Args:
agent_id: ID of the agent
agent_alias_id: ID of the agent alias
session_id: Session ID to clear
Returns:
Deletion confirmation
"""
try:
response = bedrock_agent_runtime.delete_agent_memory(
agentId=agent_id,
agentAliasId=agent_alias_id,
memoryId=session_id
)
print(f"✓ Deleted agent memory for session: {session_id}")
return response
except Exception as e:
print(f"✗ Failed to delete memory: {str(e)}")
raise
delete_agent_memory(agent_id, agent_alias_id, session_id)
8. Update Agent
Update agent configuration, instructions, or foundation model.
def update_agent(
agent_id: str,
agent_name: str = None,
instructions: str = None,
foundation_model: str = None,
description: str = None
) -> dict:
"""
Update agent configuration.
After update, must prepare agent again before use.
"""
try:
current = bedrock_agent.get_agent(agentId=agent_id)['agent']
update_params = {
'agentId': agent_id,
'agentName': agent_name or current['agentName'],
'foundationModel': foundation_model or current['foundationModel'],
'instruction': instructions or current.get('instruction', '')
}
if description:
update_params['description'] = description
response = bedrock_agent.update_agent(**update_params)
print(f"✓ Updated agent: {response['agent']['agentName']}")
print(" Note: Prepare agent before use")
return response['agent']
except Exception as e:
print(f"✗ Failed to update agent: {str(e)}")
update_agent(
agent_id=agent_id,
instructions=
)
prepare_agent(agent_id)
wait_for_agent_ready(agent_id)
9. List and Delete Agents
Manage agent lifecycle.
def list_agents() -> list:
"""List all agents in the account."""
try:
response = bedrock_agent.list_agents(maxResults=50)
agents = response.get('agentSummaries', [])
print(f"Found {len(agents)} agents:")
for agent in agents:
print(f"\n Name: {agent['agentName']}")
print(f" ID: {agent['agentId']}")
print(f" Status: {agent['agentStatus']}")
print(f" Updated: {agent['updatedAt']}")
return agents
except Exception as e:
print(f"✗ Failed to list agents: {str(e)}")
raise
def delete_agent(agent_id: str, skip_resource_in_use_check: bool = False) -> dict:
"""
Delete an agent and all associated resources.
Args:
agent_id: ID of the agent to delete
skip_resource_in_use_check: Skip check for aliases
"""
try:
response = bedrock_agent.delete_agent(
agentId=agent_id,
skipResourceInUseCheck=skip_resource_in_use_check
)
()
()
response
Exception e:
()
agents = list_agents()
Agent Patterns
Pattern 1: Single-Turn Request-Response
Simple question answering without multi-turn context.
def single_turn_query(agent_id: str, agent_alias_id: str, query: str) -> str:
"""
Execute a single query without conversation context.
Each invocation is independent.
"""
session_id = str(uuid.uuid4())
response = bedrock_agent_runtime.invoke_agent(
agentId=agent_id,
agentAliasId=agent_alias_id,
sessionId=session_id,
inputText=query,
enableTrace=False
)
completion = ''
for event in response['completion']:
if 'chunk' in event and 'bytes' in event['chunk']:
completion += event['chunk']['bytes'].decode('utf-8')
return completion
answer1 = single_turn_query(agent_id, agent_alias_id, "What's your refund policy?")
answer2 = single_turn_query(agent_id, agent_alias_id, "How long does shipping take?")
Pattern 2: Multi-Turn Conversation
Maintain context across multiple exchanges.
class ConversationSession:
"""Manage a multi-turn conversation with context."""
def __init__(self, agent_id: str, agent_alias_id: str):
self.agent_id = agent_id
self.agent_alias_id = agent_alias_id
self.session_id = str(uuid.uuid4())
self.history = []
def send_message(self, message: str) -> str:
"""Send a message and get response."""
response = bedrock_agent_runtime.invoke_agent(
agentId=self.agent_id,
agentAliasId=self.agent_alias_id,
sessionId=self.session_id,
inputText=message
)
completion = ''
for event in response['completion']:
if 'chunk' in event and 'bytes' in event['chunk']:
completion += event['chunk']['bytes'].decode('utf-8')
self.history.append({'user': message, 'agent': completion})
return completion
def reset(self):
"""Clear conversation memory."""
bedrock_agent_runtime.delete_agent_memory(
agentId=.agent_id,
agentAliasId=.agent_alias_id,
memoryId=.session_id
)
.history = []
conversation = ConversationSession(agent_id, agent_alias_id)
response1 = conversation.send_message()
response2 = conversation.send_message()
response3 = conversation.send_message()
Pattern 3: RAG with Knowledge Base
Agent retrieves information from knowledge base.
def rag_query(agent_id: str, agent_alias_id: str, query: str) -> dict:
"""
Query agent with RAG capabilities.
Returns response with citations from knowledge base.
"""
session_id = str(uuid.uuid4())
response = bedrock_agent_runtime.invoke_agent(
agentId=agent_id,
agentAliasId=agent_alias_id,
sessionId=session_id,
inputText=query,
enableTrace=True
)
result = {
'answer': '',
'citations': [],
'retrieved_docs': []
}
for event in response['completion']:
if 'chunk' in event and 'bytes' in event['chunk']:
result['answer'] += event['chunk']['bytes'].decode('utf-8')
elif 'trace' in event:
trace = event['trace']['trace']
if 'orchestrationTrace' in trace:
orch = trace['orchestrationTrace']
if 'observation' in orch:
obs = orch['observation']
if 'knowledgeBaseLookupOutput' in obs:
kb_output = obs[]
result[].extend(
kb_output.get(, [])
)
result
result = rag_query(
agent_id,
agent_alias_id,
)
()
()
doc result[]:
()
Pattern 4: Tool Use with Action Groups
Agent invokes tools to accomplish tasks.
def tool_use_example(agent_id: str, agent_alias_id: str, request: str):
"""
Agent uses action groups to invoke tools.
Demonstrates tool invocation flow.
"""
session_id = str(uuid.uuid4())
response = bedrock_agent_runtime.invoke_agent(
agentId=agent_id,
agentAliasId=agent_alias_id,
sessionId=session_id,
inputText=request,
enableTrace=True
)
for event in response['completion']:
if 'trace' in event:
trace = event['trace']['trace']
if 'orchestrationTrace' in trace:
orch = trace['orchestrationTrace']
if 'rationale' in orch:
print(f"[Planning] {orch['rationale']['text']}")
if 'invocationInput' in orch:
inv = orch['invocationInput']
if 'actionGroupInvocationInput' in inv:
action = inv['actionGroupInvocationInput']
print(f"[Tool Call] {action['actionGroupName']}")
()
()
orch:
obs = orch[]
obs:
output = obs[]
()
event event[]:
text = event[][].decode()
(text, end=, flush=)
tool_use_example(
agent_id,
agent_alias_id,
)
Best Practices
1. Agent Instructions
Write clear, specific instructions:
good_instructions = """You are a financial advisor agent for retail banking customers.
Your capabilities:
- Answer questions about savings accounts, checking accounts, and credit cards
- Look up account balances and transaction history using available tools
- Provide information about loan products from the knowledge base
- Help customers understand fees and policies
Your guidelines:
- Always verify customer identity before accessing account information
- Provide accurate information from official sources only
- If you cannot help with something, explain why and suggest alternatives
- Use professional but friendly language
- Cite sources when providing policy or fee information
Your limitations:
- You cannot make transactions or transfer money
- You cannot change account settings
- You cannot access accounts without proper verification
- For complex issues, escalate to a human advisor
Session context:
- Customer tier: {{customer_tier}}
- Customer since: {{customer_since}}
- Primary account: {{primary_account_type}}
"""
2. Session Management
Track sessions appropriately:
session_id = str(uuid.uuid4())
session_id = f"user-{user_id}-{datetime.now().strftime('%Y%m%d')}"
session_id = f"support-ticket-{ticket_id}-{int(time.time())}"
def cleanup_old_sessions(agent_id: str, agent_alias_id: str, session_ids: list):
"""Delete memory for expired sessions."""
for session_id in session_ids:
try:
bedrock_agent_runtime.delete_agent_memory(
agentId=agent_id,
agentAliasId=agent_alias_id,
memoryId=session_id
)
except:
pass
3. Error Handling
Handle streaming errors gracefully:
def invoke_agent_safe(agent_id: str, agent_alias_id: str, session_id: str, input_text: str):
"""Invoke agent with comprehensive error handling."""
try:
response = bedrock_agent_runtime.invoke_agent(
agentId=agent_id,
agentAliasId=agent_alias_id,
sessionId=session_id,
inputText=input_text
)
completion = ''
for event in response['completion']:
try:
if 'chunk' in event and 'bytes' in event['chunk']:
completion += event['chunk']['bytes'].decode('utf-8')
elif 'internalServerException' in event:
raise Exception("Internal server error")
elif 'validationException' in event:
raise ValueError("Validation error")
elif 'throttlingException' in event:
raise Exception("Rate limit exceeded")
except Exception as e:
print(f"Stream error: {str(e)}")
completion
bedrock_agent_runtime.exceptions.ResourceNotFoundException:
()
bedrock_agent_runtime.exceptions.AccessDeniedException:
()
Exception e:
()
4. Action Group Lambda Handler
Structure Lambda functions for action groups:
def lambda_handler(event, context):
"""
Handle action group invocations from Bedrock Agent.
Event structure:
{
'actionGroup': 'order-management',
'apiPath': '/orders/{order_id}',
'httpMethod': 'GET',
'parameters': [...]
}
"""
action_group = event['actionGroup']
api_path = event['apiPath']
http_method = event['httpMethod']
parameters = {}
for param in event.get('parameters', []):
parameters[param['name']] = param['value']
if api_path == '/orders/{order_id}' and http_method == 'GET':
return get_order_details(parameters.get('order_id'))
elif api_path == '/orders/search' and http_method == 'POST':
return search_orders(parameters)
else:
return {
'statusCode': 404,
'body': {'error': 'Operation not found'}
}
def get_order_details(order_id: str):
"""Get order details from database."""
order = {
'order_id': order_id,
'status': 'shipped',
: ,
: [
{: , : , : },
{: , : , : }
],
:
}
{
: ,
: order
}
5. Monitoring and Logging
Track agent performance:
import logging
logger = logging.getLogger()
logger.setLevel(logging.INFO)
def invoke_agent_with_metrics(agent_id: str, agent_alias_id: str, session_id: str, input_text: str):
"""Invoke agent with logging and metrics."""
start_time = time.time()
logger.info(f"Agent invocation started: {agent_id}")
logger.info(f"Session: {session_id}")
logger.info(f"Input: {input_text[:100]}...")
try:
response = bedrock_agent_runtime.invoke_agent(
agentId=agent_id,
agentAliasId=agent_alias_id,
sessionId=session_id,
inputText=input_text,
enableTrace=True
)
completion = ''
tool_calls = 0
kb_retrievals = 0
for event in response['completion']:
if 'chunk' in event and 'bytes' in event['chunk']:
completion += event['chunk']['bytes'].decode('utf-8')
elif 'trace' in event:
trace = event['trace']['trace']
if 'orchestrationTrace' in trace:
orch = trace['orchestrationTrace']
if orch:
inv = orch[]
inv:
tool_calls +=
orch:
obs = orch[]
obs:
kb_retrievals +=
duration = time.time() - start_time
logger.info()
logger.info()
logger.info()
logger.info()
logger.info()
{
: completion,
: {
: duration,
: tool_calls,
: kb_retrievals
}
}
Exception e:
duration = time.time() - start_time
logger.error()
Related Skills
- bedrock-agentcore: Amazon Bedrock AgentCore platform
- bedrock-agentcore-multi-agent: Multi-agent orchestration patterns
- bedrock-knowledge-bases: RAG with vector databases
- bedrock-guardrails: Content filtering and safety
- bedrock-inference: Direct model invocation
- boto3-lambda: Lambda integration for action groups
- anthropic-expert: Claude models and capabilities
References
