How do I use the AWS CDK to create an Amazon Bedrock agent?

Resolution

Note: If you receive errors when you run AWS Command Line Interface (AWS CLI) commands, then see Troubleshooting errors for the AWS CLI. Also, make sure that you're using the most recent AWS CLI version.

Use the AWS CDK to define, deploy, and manage Bedrock agents as infrastructure-as-code to guarantee consistency, repeatability, and version control.

Note: To use the AWS CDK to update an Amazon Bedrock Agent, see How do I use the AWS CDK to update an Amazon Bedrock Agent?

Prerequisites: Make sure that you meet the following prerequisites:

• You have an AWS Account with the appropriate AWS Identity and Access Management (IAM) permissions.
• Install AWS CDK, Node.js, and Python 3.8 or later.
• Configure AWS CLI.
• You can access Amazon Bedrock.

Create an AWS CDK project

To create an AWS CDK project that uses Python, run the following command:

mkdir bedrock-agent-cdk
cd bedrock-agent-cdk
cdk init app --language python

Install the required dependencies

To install the required dependencies, run the following command:

pip install aws-cdk-lib constructs

Define the Bedrock Agent in your CDK stack

Open the main stack file bedrock_agent_cdk/bedrock_agent_cdk_stack.py. Then, replace the configurations to create an agent based on your needs.

To create an Amazon Bedrock agent, add the following configuration to your main stack file:

from aws_cdk import (
    Stack,
    aws_bedrock as bedrock,
    aws_iam as iam,
    aws_s3 as s3,
    CfnOutput,
)
from constructs import Construct

class BedrockAgentCdkStack(Stack):
    def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
        super().__init__(scope, construct_id, **kwargs)

        # IAM role for agent
        agent_role = iam.Role(
            self,
            "BedrockAgentRole",
            assumed_by=iam.ServicePrincipal("bedrock.amazonaws.com"),
        )

        agent_role.add_to_policy(
            iam.PolicyStatement(
                actions=["bedrock:InvokeModel"],
                resources=[
                    f"arn:aws:bedrock:{self.region}::foundation-model/anthropic.claude-v2",
                    f"arn:aws:bedrock:{self.region}::foundation-model/anthropic.claude-3-sonnet-20240229-v1:0"
                ],
            )
        )

        # Create Bedrock Agent
        agent = bedrock.CfnAgent(
            self,
            "BedrockAgent",
            agent_name="my-bedrock-agent",
            agent_resource_role_arn=agent_role.role_arn,
            foundation_model="anthropic.claude-3-sonnet-20240229-v1:0",
            instruction="You are an HR agent, helping employees understand HR policies and manage vacation time",
            description="Agent for providing HR assistance to reserve vacation time off",
        )

        # Create alias
        agent_alias = bedrock.CfnAgentAlias(
            self,
            "AgentAlias",
            agent_alias_name="prod",
            agent_id=agent.attr_agent_id,
        )

        CfnOutput(self, "AgentId", value=agent.attr_agent_id)
        CfnOutput(self, "AgentAliasId", value=agent_alias.attr_agent_alias_id)

To create an agent with a knowledge base, add the following configuration to your main stack file:

from aws_cdk import (
    Stack,
    aws_bedrock as bedrock,
    aws_iam as iam,
    aws_s3 as s3,
    CfnOutput,
)
from constructs import Construct

class BedrockAgentCdkStack(Stack):
    def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
        super().__init__(scope, construct_id, **kwargs)

        # Create S3 bucket for knowledge base data
        kb_bucket = s3.Bucket(
            self,
            "KnowledgeBaseBucket",
            versioned=True,
            encryption=s3.BucketEncryption.S3_MANAGED,
        )

        # IAM role for knowledge base
        kb_role = f"arn:aws:iam::{self.account}:role/service-role/role-name"

        # Create Knowledge Base
        knowledge_base = bedrock.CfnKnowledgeBase(
            self,
            "KnowledgeBase",
            name="my-knowledge-base",
            role_arn=kb_role,
            knowledge_base_configuration=bedrock.CfnKnowledgeBase.KnowledgeBaseConfigurationProperty(
                type="VECTOR",
                vector_knowledge_base_configuration=bedrock.CfnKnowledgeBase.VectorKnowledgeBaseConfigurationProperty(
                    embedding_model_arn=f"arn:aws:bedrock:{self.region}::foundation-model/amazon.titan-embed-text-v2:0"
                ),
            ),
            storage_configuration=bedrock.CfnKnowledgeBase.StorageConfigurationProperty(
                type="OPENSEARCH_SERVERLESS",
                opensearch_serverless_configuration=bedrock.CfnKnowledgeBase.OpenSearchServerlessConfigurationProperty(
                    collection_arn=f"arn:aws:aoss:{self.region}:{self.account}:collection/your-collection-id",
                    vector_index_name="bedrock-knowledge-base-index",
                    field_mapping=bedrock.CfnKnowledgeBase.OpenSearchServerlessFieldMappingProperty(
                        vector_field="bedrock-knowledge-base-default-vector",
                        text_field="AMAZON_BEDROCK_TEXT",
                        metadata_field="AMAZON_BEDROCK_METADATA",
                    ),
                ),
            ),
        )

        # Create Data Source
        data_source = bedrock.CfnDataSource(
            self,
            "DataSource",
            name="s3-data-source",
            knowledge_base_id=knowledge_base.attr_knowledge_base_id,
            data_source_configuration=bedrock.CfnDataSource.DataSourceConfigurationProperty(
                type="S3",
                s3_configuration=bedrock.CfnDataSource.S3DataSourceConfigurationProperty(
                    bucket_arn=kb_bucket.bucket_arn,
                ),
            ),
        )

        # Create Agent with Knowledge Base
        agent_role = iam.Role(
            self,
            "AgentRole",
            assumed_by=iam.ServicePrincipal("bedrock.amazonaws.com"),
        )

        agent_role.add_to_policy(
            iam.PolicyStatement(
                actions=["bedrock:InvokeModel", "bedrock:Retrieve"],
                resources=["*"],
            )
        )

        agent = bedrock.CfnAgent(
            self,
            "Agent",
            agent_name="kb-agent",
            agent_resource_role_arn=agent_role.role_arn,
            foundation_model="anthropic.claude-3-sonnet-20240229-v1:0",
            instruction="You are an assistant with access to a knowledge base. Use it to answer questions accurately.",
            knowledge_bases=[
                bedrock.CfnAgent.AgentKnowledgeBaseProperty(
                    knowledge_base_id=knowledge_base.attr_knowledge_base_id,
                    description="Company documentation knowledge base",
                    knowledge_base_state="ENABLED",
                )
            ],
        )

        # Create alias
        agent_alias = bedrock.CfnAgentAlias(
            self,
            "AgentAlias",
            agent_alias_name="prod",
            agent_id=agent.attr_agent_id,
        )

        CfnOutput(self, "KnowledgeBaseId", value=knowledge_base.attr_knowledge_base_id)
        CfnOutput(self, "BucketName", value=kb_bucket.bucket_name)
        CfnOutput(self, "AgentId", value=agent.attr_agent_id)
        CfnOutput(self, "AgentAliasId", value=agent_alias.attr_agent_alias_id)

Note: For more information, see Augment response generation for your agent with knowledge base.

To create an agent with action groups, add the following configuration to your main stack file:

from aws_cdk import (
    Stack,
    aws_bedrock as bedrock,
    aws_iam as iam,
    aws_s3 as s3,
    aws_lambda as lambda_,
    CfnOutput,
)
from constructs import Construct

class BedrockAgentCdkStack(Stack):
    def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
        super().__init__(scope, construct_id, **kwargs)

        # Create Lambda function for action group
        action_lambda = lambda_.Function(
            self,
            "AgentActionFunction",
            runtime=lambda_.Runtime.PYTHON_3_11,
            handler="index.handler",
            code=lambda_.Code.from_inline("""
def handler(event, context):
    import json
    
    action = event.get('actionGroup', '')
    api_path = event.get('apiPath', '')
    parameters = event.get('parameters', [])
    
    # Process the action
    if api_path == '/getWeather':
        city = next((p['value'] for p in parameters if p['name'] == 'city'), 'Unknown')
        result = {
            'city': city,
            'temperature': 72,
            'condition': 'Sunny'
        }
    else:
        result = {'error': 'Unknown action'}
    
    return {
        'messageVersion': '1.0',
        'response': {
            'actionGroup': action,
            'apiPath': api_path,
            'httpMethod': event.get('httpMethod', 'GET'),
            'httpStatusCode': 200,
            'responseBody': {
                'application/json': {
                    'body': json.dumps(result)
                }
            }
        }
    }
            """),
        )

        # IAM role for agent
        agent_role = iam.Role(
            self,
            "BedrockAgentRole",
            assumed_by=iam.ServicePrincipal("bedrock.amazonaws.com"),
        )

        agent_role.add_to_policy(
            iam.PolicyStatement(
                actions=["bedrock:InvokeModel"],
                resources=[
                    f"arn:aws:bedrock:{self.region}::foundation-model/anthropic.claude-v2",
                    f"arn:aws:bedrock:{self.region}::foundation-model/anthropic.claude-3-sonnet-20240229-v1:0"
                ],
            )
        )

        # Allow agent to invoke Lambda
        action_lambda.grant_invoke(agent_role)

        # Create Bedrock Agent
        agent = bedrock.CfnAgent(
            self,
            "BedrockAgent",
            agent_name="my-bedrock-agent",
            agent_resource_role_arn=agent_role.role_arn,
            foundation_model="anthropic.claude-3-sonnet-20240229-v1:0",
            instruction="You are a weather assistant that can provide weather information for cities.",
            description="Weather information agent",
        )

        # Create Action Group
        action_group = bedrock.CfnAgent.AgentActionGroupProperty(
            action_group_name="weather-actions",
            description="Actions for getting weather information",
            action_group_state="ENABLED",
            action_group_executor=bedrock.CfnAgent.ActionGroupExecutorProperty(
                lambda_=action_lambda.function_arn
            ),
            api_schema=bedrock.CfnAgent.APISchemaProperty(
                payload="""
{
  "openapi": "3.0.0",
  "info": {
    "title": "Weather API",
    "version": "1.0.0",
    "description": "API for getting weather information"
  },
  "paths": {
    "/getWeather": {
      "get": {
        "summary": "Get weather for a city",
        "description": "Returns current weather information for the specified city",
        "operationId": "getWeather",
        "parameters": [
          {
            "name": "city",
            "in": "query",
            "description": "Name of the city",
            "required": true,
            "schema": {
              "type": "string"
            }
          }
        ],
        "responses": {
          "200": {
            "description": "Successful response",
            "content": {
              "application/json": {
                "schema": {
                  "type": "object",
                  "properties": {
                    "city": {"type": "string"},
                    "temperature": {"type": "number"},
                    "condition": {"type": "string"}
                  }
                }
              }
            }
          }
        }
      }
    }
  }
}
                """
            ),
        )

        # Update agent with action groups
        agent.action_groups = [action_group]

        # Create alias
        agent_alias = bedrock.CfnAgentAlias(
            self,
            "AgentAlias",
            agent_alias_name="prod",
            agent_id=agent.attr_agent_id,
        )
        CfnOutput(self, "AgentId", value=agent.attr_agent_id)
        CfnOutput(self, "AgentAliasId", value=agent_alias.attr_agent_alias_id)

Deploy the stack

To synthesize your AWS CDK stack, run the following command:

cdk synth

To bootstrap your AWS CDK stack, run the following command:

# Bootstrap CDK (first time only)
cdk bootstrap

To deploy the stack, run the following command:

cdk deploy

Remove your resources

To destroy the stack and remove unnecessary resources, run the following command to destroy the stack:

cdk destroy

Related information

Agents for Amazon Bedrock Runtime