# An end-to-end example showing how to create and invoke Amazon Bedrock Agents using an AWS SDK Create and invoke an agent 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. ------ #### [ Python ] **SDK for Python (Boto3)** There's more on GitHub. Find the complete example and learn how to set up and run in the [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/bedrock-agent#code-examples). 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}") ``` + For API details, see the following topics in *AWS SDK for Python (Boto3) API Reference*. + [CreateAgent](https://docs.aws.amazon.com/goto/boto3/bedrock-agent-2023-12-12/CreateAgent) + [CreateAgentActionGroup](https://docs.aws.amazon.com/goto/boto3/bedrock-agent-2023-12-12/CreateAgentActionGroup) + [CreateAgentAlias](https://docs.aws.amazon.com/goto/boto3/bedrock-agent-2023-12-12/CreateAgentAlias) + [DeleteAgent](https://docs.aws.amazon.com/goto/boto3/bedrock-agent-2023-12-12/DeleteAgent) + [DeleteAgentAlias](https://docs.aws.amazon.com/goto/boto3/bedrock-agent-2023-12-12/DeleteAgentAlias) + [GetAgent](https://docs.aws.amazon.com/goto/boto3/bedrock-agent-2023-12-12/GetAgent) + [ListAgentActionGroups](https://docs.aws.amazon.com/goto/boto3/bedrock-agent-2023-12-12/ListAgentActionGroups) + [ListAgentKnowledgeBases](https://docs.aws.amazon.com/goto/boto3/bedrock-agent-2023-12-12/ListAgentKnowledgeBases) + [ListAgents](https://docs.aws.amazon.com/goto/boto3/bedrock-agent-2023-12-12/ListAgents) + [PrepareAgent](https://docs.aws.amazon.com/goto/boto3/bedrock-agent-2023-12-12/PrepareAgent) ------ For a complete list of AWS SDK developer guides and code examples, see [Using Amazon Bedrock with an AWS SDK](sdk-general-information-section.md). This topic also includes information about getting started and details about previous SDK versions.