Amazon Bedrock Agents examples using SDK for Python (Boto3) - AWS SDK Code Examples

There are more AWS SDK examples available in the AWS Doc SDK Examples GitHub repo.

Amazon Bedrock Agents examples using SDK for Python (Boto3)

The following code examples show you how to perform actions and implement common scenarios by using the AWS SDK for Python (Boto3) with Amazon Bedrock Agents.

Actions are code excerpts from larger programs and must be run in context. While actions show you how to call individual service functions, you can see actions in context in their related scenarios.

Scenarios are code examples that show you how to accomplish specific tasks by calling multiple functions within a service or combined with other AWS services.

Each example includes a link to the complete source code, where you can find instructions on how to set up and run the code in context.

Actions

The following code example shows how to use CreateAgent.

SDK for Python (Boto3)
Note

There's more on GitHub. Find the complete example and learn how to set up and run in the AWS Code Examples Repository.

Create an agent.

def create_agent(self, agent_name, foundation_model, role_arn, instruction): """ Creates an agent that orchestrates interactions between foundation models, data sources, software applications, user conversations, and APIs to carry out tasks to help customers. :param agent_name: A name for the agent. :param foundation_model: The foundation model to be used for orchestration by the agent. :param role_arn: The ARN of the IAM role with permissions needed by the agent. :param instruction: Instructions that tell the agent what it should do and how it should interact with users. :return: The response from Amazon Bedrock Agents if successful, otherwise raises an exception. """ try: response = self.client.create_agent( agentName=agent_name, foundationModel=foundation_model, agentResourceRoleArn=role_arn, instruction=instruction, ) except ClientError as e: logger.error(f"Error: Couldn't create agent. Here's why: {e}") raise else: return response["agent"]
  • For API details, see CreateAgent in AWS SDK for Python (Boto3) API Reference.

The following code example shows how to use CreateAgentActionGroup.

SDK for Python (Boto3)
Note

There's more on GitHub. Find the complete example and learn how to set up and run in the AWS Code Examples Repository.

Create an agent action group.

def create_agent_action_group( self, name, description, agent_id, agent_version, function_arn, api_schema ): """ Creates an action group for an agent. An action group defines a set of actions that an agent should carry out for the customer. :param name: The name to give the action group. :param description: The description of the action group. :param agent_id: The unique identifier of the agent for which to create the action group. :param agent_version: The version of the agent for which to create the action group. :param function_arn: The ARN of the Lambda function containing the business logic that is carried out upon invoking the action. :param api_schema: Contains the OpenAPI schema for the action group. :return: Details about the action group that was created. """ try: response = self.client.create_agent_action_group( actionGroupName=name, description=description, agentId=agent_id, agentVersion=agent_version, actionGroupExecutor={"lambda": function_arn}, apiSchema={"payload": api_schema}, ) agent_action_group = response["agentActionGroup"] except ClientError as e: logger.error(f"Error: Couldn't create agent action group. Here's why: {e}") raise else: return agent_action_group

The following code example shows how to use CreateAgentAlias.

SDK for Python (Boto3)
Note

There's more on GitHub. Find the complete example and learn how to set up and run in the AWS Code Examples Repository.

Create an agent alias.

def create_agent_alias(self, name, agent_id): """ Creates an alias of an agent that can be used to deploy the agent. :param name: The name of the alias. :param agent_id: The unique identifier of the agent. :return: Details about the alias that was created. """ try: response = self.client.create_agent_alias( agentAliasName=name, agentId=agent_id ) agent_alias = response["agentAlias"] except ClientError as e: logger.error(f"Couldn't create agent alias. {e}") raise else: return agent_alias
  • For API details, see CreateAgentAlias in AWS SDK for Python (Boto3) API Reference.

The following code example shows how to use DeleteAgent.

SDK for Python (Boto3)
Note

There's more on GitHub. Find the complete example and learn how to set up and run in the AWS Code Examples Repository.

Delete an agent.

def delete_agent(self, agent_id): """ Deletes an Amazon Bedrock agent. :param agent_id: The unique identifier of the agent to delete. :return: The response from Amazon Bedrock Agents if successful, otherwise raises an exception. """ try: response = self.client.delete_agent( agentId=agent_id, skipResourceInUseCheck=False ) except ClientError as e: logger.error(f"Couldn't delete agent. {e}") raise else: return response
  • For API details, see DeleteAgent in AWS SDK for Python (Boto3) API Reference.

The following code example shows how to use DeleteAgentAlias.

SDK for Python (Boto3)
Note

There's more on GitHub. Find the complete example and learn how to set up and run in the AWS Code Examples Repository.

Delete an agent alias.

def delete_agent_alias(self, agent_id, agent_alias_id): """ Deletes an alias of an Amazon Bedrock agent. :param agent_id: The unique identifier of the agent that the alias belongs to. :param agent_alias_id: The unique identifier of the alias to delete. :return: The response from Amazon Bedrock Agents if successful, otherwise raises an exception. """ try: response = self.client.delete_agent_alias( agentId=agent_id, agentAliasId=agent_alias_id ) except ClientError as e: logger.error(f"Couldn't delete agent alias. {e}") raise else: return response
  • For API details, see DeleteAgentAlias in AWS SDK for Python (Boto3) API Reference.

The following code example shows how to use GetAgent.

SDK for Python (Boto3)
Note

There's more on GitHub. Find the complete example and learn how to set up and run in the AWS Code Examples Repository.

Get an agent.

def get_agent(self, agent_id, log_error=True): """ Gets information about an agent. :param agent_id: The unique identifier of the agent. :param log_error: Whether to log any errors that occur when getting the agent. If True, errors will be logged to the logger. If False, errors will still be raised, but not logged. :return: The information about the requested agent. """ try: response = self.client.get_agent(agentId=agent_id) agent = response["agent"] except ClientError as e: if log_error: logger.error(f"Couldn't get agent {agent_id}. {e}") raise else: return agent
  • For API details, see GetAgent in AWS SDK for Python (Boto3) API Reference.

The following code example shows how to use ListAgentActionGroups.

SDK for Python (Boto3)
Note

There's more on GitHub. Find the complete example and learn how to set up and run in the AWS Code Examples Repository.

List the action groups for an agent.

def list_agent_action_groups(self, agent_id, agent_version): """ List the action groups for a version of an Amazon Bedrock Agent. :param agent_id: The unique identifier of the agent. :param agent_version: The version of the agent. :return: The list of action group summaries for the version of the agent. """ try: action_groups = [] paginator = self.client.get_paginator("list_agent_action_groups") for page in paginator.paginate( agentId=agent_id, agentVersion=agent_version, PaginationConfig={"PageSize": 10}, ): action_groups.extend(page["actionGroupSummaries"]) except ClientError as e: logger.error(f"Couldn't list action groups. {e}") raise else: return action_groups

The following code example shows how to use ListAgentKnowledgeBases.

SDK for Python (Boto3)
Note

There's more on GitHub. Find the complete example and learn how to set up and run in the AWS Code Examples Repository.

List the knowledge bases associated with an agent.

def list_agent_knowledge_bases(self, agent_id, agent_version): """ List the knowledge bases associated with a version of an Amazon Bedrock Agent. :param agent_id: The unique identifier of the agent. :param agent_version: The version of the agent. :return: The list of knowledge base summaries for the version of the agent. """ try: knowledge_bases = [] paginator = self.client.get_paginator("list_agent_knowledge_bases") for page in paginator.paginate( agentId=agent_id, agentVersion=agent_version, PaginationConfig={"PageSize": 10}, ): knowledge_bases.extend(page["agentKnowledgeBaseSummaries"]) except ClientError as e: logger.error(f"Couldn't list knowledge bases. {e}") raise else: return knowledge_bases

The following code example shows how to use ListAgents.

SDK for Python (Boto3)
Note

There's more on GitHub. Find the complete example and learn how to set up and run in the AWS Code Examples Repository.

List the agents belonging to an account.

def list_agents(self): """ List the available Amazon Bedrock Agents. :return: The list of available bedrock agents. """ try: all_agents = [] paginator = self.client.get_paginator("list_agents") for page in paginator.paginate(PaginationConfig={"PageSize": 10}): all_agents.extend(page["agentSummaries"]) except ClientError as e: logger.error(f"Couldn't list agents. {e}") raise else: return all_agents
  • For API details, see ListAgents in AWS SDK for Python (Boto3) API Reference.

The following code example shows how to use PrepareAgent.

SDK for Python (Boto3)
Note

There's more on GitHub. Find the complete example and learn how to set up and run in the AWS Code Examples Repository.

Prepare an agent for internal testing.

def prepare_agent(self, agent_id): """ Creates a DRAFT version of the agent that can be used for internal testing. :param agent_id: The unique identifier of the agent to prepare. :return: The response from Amazon Bedrock Agents if successful, otherwise raises an exception. """ try: prepared_agent_details = self.client.prepare_agent(agentId=agent_id) except ClientError as e: logger.error(f"Couldn't prepare agent. {e}") raise else: return prepared_agent_details
  • For API details, see PrepareAgent in AWS SDK for Python (Boto3) API Reference.

Scenarios

The following code example shows how to:

  • Create an execution role for the agent.

  • Create the agent and deploy a DRAFT version.

  • Create a Lambda function that implements the agent's capabilities.

  • Create an action group that connects the agent to the Lambda function.

  • Deploy the fully configured agent.

  • Invoke the agent with user-provided prompts.

  • Delete all created resources.

SDK for Python (Boto3)
Note

There's more on GitHub. Find the complete example and learn how to set up and run in the AWS Code Examples Repository.

Create and invoke an agent.

REGION = "us-east-1" ROLE_POLICY_NAME = "agent_permissions" class BedrockAgentScenarioWrapper: """Runs a scenario that shows how to get started using Amazon Bedrock Agents.""" def __init__( self, bedrock_agent_client, runtime_client, lambda_client, iam_resource, postfix ): self.iam_resource = iam_resource self.lambda_client = lambda_client self.bedrock_agent_runtime_client = runtime_client self.postfix = postfix self.bedrock_wrapper = BedrockAgentWrapper(bedrock_agent_client) self.agent = None self.agent_alias = None self.agent_role = None self.prepared_agent_details = None self.lambda_role = None self.lambda_function = None def run_scenario(self): print("=" * 88) print("Welcome to the Amazon Bedrock Agents demo.") print("=" * 88) # Query input from user print("Let's start with creating an agent:") print("-" * 40) name, foundation_model = self._request_name_and_model_from_user() print("-" * 40) # Create an execution role for the agent self.agent_role = self._create_agent_role(foundation_model) # Create the agent self.agent = self._create_agent(name, foundation_model) # Prepare a DRAFT version of the agent self.prepared_agent_details = self._prepare_agent() # Create the agent's Lambda function self.lambda_function = self._create_lambda_function() # Configure permissions for the agent to invoke the Lambda function self._allow_agent_to_invoke_function() self._let_function_accept_invocations_from_agent() # Create an action group to connect the agent with the Lambda function self._create_agent_action_group() # If the agent has been modified or any components have been added, prepare the agent again components = [self._get_agent()] components += self._get_agent_action_groups() components += self._get_agent_knowledge_bases() latest_update = max(component["updatedAt"] for component in components) if latest_update > self.prepared_agent_details["preparedAt"]: self.prepared_agent_details = self._prepare_agent() # Create an agent alias self.agent_alias = self._create_agent_alias() # Test the agent self._chat_with_agent(self.agent_alias) print("=" * 88) print("Thanks for running the demo!\n") if q.ask("Do you want to delete the created resources? [y/N] ", q.is_yesno): self._delete_resources() print("=" * 88) print( "All demo resources have been deleted. Thanks again for running the demo!" ) else: self._list_resources() print("=" * 88) print("Thanks again for running the demo!") def _request_name_and_model_from_user(self): existing_agent_names = [ agent["agentName"] for agent in self.bedrock_wrapper.list_agents() ] while True: name = q.ask("Enter an agent name: ", self.is_valid_agent_name) if name.lower() not in [n.lower() for n in existing_agent_names]: break print( f"Agent {name} conflicts with an existing agent. Please use a different name." ) models = ["anthropic.claude-instant-v1", "anthropic.claude-v2"] model_id = models[ q.choose("Which foundation model would you like to use? ", models) ] return name, model_id def _create_agent_role(self, model_id): role_name = f"AmazonBedrockExecutionRoleForAgents_{self.postfix}" model_arn = f"arn:aws:bedrock:{REGION}::foundation-model/{model_id}*" print("Creating an an execution role for the agent...") try: role = self.iam_resource.create_role( RoleName=role_name, AssumeRolePolicyDocument=json.dumps( { "Version": "2012-10-17", "Statement": [ { "Effect": "Allow", "Principal": {"Service": "bedrock.amazonaws.com"}, "Action": "sts:AssumeRole", } ], } ), ) role.Policy(ROLE_POLICY_NAME).put( PolicyDocument=json.dumps( { "Version": "2012-10-17", "Statement": [ { "Effect": "Allow", "Action": "bedrock:InvokeModel", "Resource": model_arn, } ], } ) ) except ClientError as e: logger.error(f"Couldn't create role {role_name}. Here's why: {e}") raise return role def _create_agent(self, name, model_id): print("Creating the agent...") instruction = """ You are a friendly chat bot. You have access to a function called that returns information about the current date and time. When responding with date or time, please make sure to add the timezone UTC. """ agent = self.bedrock_wrapper.create_agent( agent_name=name, foundation_model=model_id, instruction=instruction, role_arn=self.agent_role.arn, ) self._wait_for_agent_status(agent["agentId"], "NOT_PREPARED") return agent def _prepare_agent(self): print("Preparing the agent...") agent_id = self.agent["agentId"] prepared_agent_details = self.bedrock_wrapper.prepare_agent(agent_id) self._wait_for_agent_status(agent_id, "PREPARED") return prepared_agent_details def _create_lambda_function(self): print("Creating the Lambda function...") function_name = f"AmazonBedrockExampleFunction_{self.postfix}" self.lambda_role = self._create_lambda_role() try: deployment_package = self._create_deployment_package(function_name) lambda_function = self.lambda_client.create_function( FunctionName=function_name, Description="Lambda function for Amazon Bedrock example", Runtime="python3.11", Role=self.lambda_role.arn, Handler=f"{function_name}.lambda_handler", Code={"ZipFile": deployment_package}, Publish=True, ) waiter = self.lambda_client.get_waiter("function_active_v2") waiter.wait(FunctionName=function_name) except ClientError as e: logger.error( f"Couldn't create Lambda function {function_name}. Here's why: {e}" ) raise return lambda_function def _create_lambda_role(self): print("Creating an execution role for the Lambda function...") role_name = f"AmazonBedrockExecutionRoleForLambda_{self.postfix}" try: role = self.iam_resource.create_role( RoleName=role_name, AssumeRolePolicyDocument=json.dumps( { "Version": "2012-10-17", "Statement": [ { "Effect": "Allow", "Principal": {"Service": "lambda.amazonaws.com"}, "Action": "sts:AssumeRole", } ], } ), ) role.attach_policy( PolicyArn="arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole" ) print(f"Created role {role_name}") except ClientError as e: logger.error(f"Couldn't create role {role_name}. Here's why: {e}") raise print("Waiting for the execution role to be fully propagated...") wait(10) return role def _allow_agent_to_invoke_function(self): policy = self.iam_resource.RolePolicy( self.agent_role.role_name, ROLE_POLICY_NAME ) doc = policy.policy_document doc["Statement"].append( { "Effect": "Allow", "Action": "lambda:InvokeFunction", "Resource": self.lambda_function["FunctionArn"], } ) self.agent_role.Policy(ROLE_POLICY_NAME).put(PolicyDocument=json.dumps(doc)) def _let_function_accept_invocations_from_agent(self): try: self.lambda_client.add_permission( FunctionName=self.lambda_function["FunctionName"], SourceArn=self.agent["agentArn"], StatementId="BedrockAccess", Action="lambda:InvokeFunction", Principal="bedrock.amazonaws.com", ) except ClientError as e: logger.error( f"Couldn't grant Bedrock permission to invoke the Lambda function. Here's why: {e}" ) raise def _create_agent_action_group(self): print("Creating an action group for the agent...") try: with open("./scenario_resources/api_schema.yaml") as file: self.bedrock_wrapper.create_agent_action_group( name="current_date_and_time", description="Gets the current date and time.", agent_id=self.agent["agentId"], agent_version=self.prepared_agent_details["agentVersion"], function_arn=self.lambda_function["FunctionArn"], api_schema=json.dumps(yaml.safe_load(file)), ) except ClientError as e: logger.error(f"Couldn't create agent action group. Here's why: {e}") raise def _get_agent(self): return self.bedrock_wrapper.get_agent(self.agent["agentId"]) def _get_agent_action_groups(self): return self.bedrock_wrapper.list_agent_action_groups( self.agent["agentId"], self.prepared_agent_details["agentVersion"] ) def _get_agent_knowledge_bases(self): return self.bedrock_wrapper.list_agent_knowledge_bases( self.agent["agentId"], self.prepared_agent_details["agentVersion"] ) def _create_agent_alias(self): print("Creating an agent alias...") agent_alias_name = "test_agent_alias" agent_alias = self.bedrock_wrapper.create_agent_alias( agent_alias_name, self.agent["agentId"] ) self._wait_for_agent_status(self.agent["agentId"], "PREPARED") return agent_alias def _wait_for_agent_status(self, agent_id, status): while self.bedrock_wrapper.get_agent(agent_id)["agentStatus"] != status: wait(2) def _chat_with_agent(self, agent_alias): print("-" * 88) print("The agent is ready to chat.") print("Try asking for the date or time. Type 'exit' to quit.") # Create a unique session ID for the conversation session_id = uuid.uuid4().hex while True: prompt = q.ask("Prompt: ", q.non_empty) if prompt == "exit": break response = asyncio.run(self._invoke_agent(agent_alias, prompt, session_id)) print(f"Agent: {response}") async def _invoke_agent(self, agent_alias, prompt, session_id): response = self.bedrock_agent_runtime_client.invoke_agent( agentId=self.agent["agentId"], agentAliasId=agent_alias["agentAliasId"], sessionId=session_id, inputText=prompt, ) completion = "" for event in response.get("completion"): chunk = event["chunk"] completion += chunk["bytes"].decode() return completion def _delete_resources(self): if self.agent: agent_id = self.agent["agentId"] if self.agent_alias: agent_alias_id = self.agent_alias["agentAliasId"] print("Deleting agent alias...") self.bedrock_wrapper.delete_agent_alias(agent_id, agent_alias_id) print("Deleting agent...") agent_status = self.bedrock_wrapper.delete_agent(agent_id)["agentStatus"] while agent_status == "DELETING": wait(5) try: agent_status = self.bedrock_wrapper.get_agent( agent_id, log_error=False )["agentStatus"] except ClientError as err: if err.response["Error"]["Code"] == "ResourceNotFoundException": agent_status = "DELETED" if self.lambda_function: name = self.lambda_function["FunctionName"] print(f"Deleting function '{name}'...") self.lambda_client.delete_function(FunctionName=name) if self.agent_role: print(f"Deleting role '{self.agent_role.role_name}'...") self.agent_role.Policy(ROLE_POLICY_NAME).delete() self.agent_role.delete() if self.lambda_role: print(f"Deleting role '{self.lambda_role.role_name}'...") for policy in self.lambda_role.attached_policies.all(): policy.detach_role(RoleName=self.lambda_role.role_name) self.lambda_role.delete() def _list_resources(self): print("-" * 40) print(f"Here is the list of created resources in '{REGION}'.") print("Make sure you delete them once you're done to avoid unnecessary costs.") if self.agent: print(f"Bedrock Agent: {self.agent['agentName']}") if self.lambda_function: print(f"Lambda function: {self.lambda_function['FunctionName']}") if self.agent_role: print(f"IAM role: {self.agent_role.role_name}") if self.lambda_role: print(f"IAM role: {self.lambda_role.role_name}") @staticmethod def is_valid_agent_name(answer): valid_regex = r"^[a-zA-Z0-9_-]{1,100}$" return ( answer if answer and len(answer) <= 100 and re.match(valid_regex, answer) else None, "I need a name for the agent, please. Valid characters are a-z, A-Z, 0-9, _ (underscore) and - (hyphen).", ) @staticmethod def _create_deployment_package(function_name): buffer = io.BytesIO() with zipfile.ZipFile(buffer, "w") as zipped: zipped.write( "./scenario_resources/lambda_function.py", f"{function_name}.py" ) buffer.seek(0) return buffer.read() if __name__ == "__main__": logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s") postfix = "".join( random.choice(string.ascii_lowercase + "0123456789") for _ in range(8) ) scenario = BedrockAgentScenarioWrapper( bedrock_agent_client=boto3.client( service_name="bedrock-agent", region_name=REGION ), runtime_client=boto3.client( service_name="bedrock-agent-runtime", region_name=REGION ), lambda_client=boto3.client(service_name="lambda", region_name=REGION), iam_resource=boto3.resource("iam"), postfix=postfix, ) try: scenario.run_scenario() except Exception as e: logging.exception(f"Something went wrong with the demo. Here's what: {e}")

The following code example shows how to build and orchestrate generative AI applications with Amazon Bedrock and Step Functions.

SDK for Python (Boto3)

The Amazon Bedrock Serverless Prompt Chaining scenario demonstrates how AWS Step Functions, Amazon Bedrock, and https://docs.aws.amazon.com/bedrock/latest/userguide/agents.html can be used to build and orchestrate complex, serverless, and highly scalable generative AI applications. It contains the following working examples:

  • Write an analysis of a given novel for a literature blog. This example illustrates a simple, sequential chain of prompts.

  • Generate a short story about a given topic. This example illustrates how the AI can iteratively process a list of items that it previously generated.

  • Create an itinerary for a weekend vacation to a given destination. This example illustrates how to parallelize multiple distinct prompts.

  • Pitch movie ideas to a human user acting as a movie producer. This example illustrates how to parallelize the same prompt with different inference parameters, how to backtrack to a previous step in the chain, and how to include human input as part of the workflow.

  • Plan a meal based on ingredients the user has at hand. This example illustrates how prompt chains can incorporate two distinct AI conversations, with two AI personas engaging in a debate with each other to improve the final outcome.

  • Find and summarize today's highest trending GitHub repository. This example illustrates chaining multiple AI agents that interact with external APIs.

For complete source code and instructions to set up and run, see the full project on GitHub.

Services used in this example
  • Amazon Bedrock

  • Amazon Bedrock Runtime

  • Amazon Bedrock Agents

  • Amazon Bedrock Agents Runtime

  • Step Functions