Amazon Bedrock AgentCore Construct Library
---The APIs of higher level constructs in this module are experimental and under active development. They are subject to non-backward compatible changes or removal in any future version. These are not subject to the Semantic Versioning model and breaking changes will be announced in the release notes. This means that while you may use them, you may need to update your source code when upgrading to a newer version of this package.
The Policy submodule is the only submodule that remains in alpha. All other constructs have graduated to stable in
aws-cdk-lib/aws-bedrockagentcoreand we recommend migrating to the stable versions.
| Language | Package |
|---|---|
TypeScript |
@aws-cdk/aws-bedrock-agentcore-alpha |
Migration to Stable
All constructs except Policy have moved to aws-cdk-lib/aws-bedrockagentcore:
# Before
import aws_cdk.aws_bedrock_agentcore_alpha as agentcore
# After (for all non-Policy constructs)
import aws_cdk.aws_bedrockagentcore as agentcore
The following constructs are now in stable:
Runtime:
Runtime,RuntimeEndpoint,AgentRuntimeArtifact,NetworkConfiguration,ObservabilityGateway:
Gateway,GatewayTarget,GatewayAuthorizer,GatewayCredentialProvider,InterceptorTools:
BrowserCustom,CodeInterpreterCustomMemory:
Memory,MemoryStrategyEvaluation:
OnlineEvaluationConfig,Evaluator,EvaluatorSelectorIdentity:
OAuth2CredentialProvider,ApiKeyCredentialProvider,WorkloadIdentity
What Remains in Alpha
The Policy submodule remains experimental:
PolicyEnginePolicyPolicyStatementPolicyValidationModePolicyEngineMode
Policy Engine
A policy engine is a collection of policies that evaluates and authorizes agent tool calls. When associated with a gateway, the policy engine intercepts all agent requests and determines whether to allow or deny each action based on the defined policies.
For more information, see the Policy in Amazon Bedrock AgentCore documentation.
PolicyEngine Properties
| Name | Type | Required | Description |
|---|---|---|---|
policyEngineName |
string |
No | The name of the policy engine. Valid characters: a-z, A-Z, 0-9, _ (underscore). Must start with a letter, 1-48 characters. If not provided, a unique name will be auto-generated |
description |
string |
No | Optional description for the policy engine (max 4,096 characters). Default: no description |
kmsKey |
IKey |
No | Custom KMS key for encryption. IMPORTANT: Once set, cannot be changed (requires replacement). Must be symmetric ENCRYPT_DECRYPT key. If key becomes inaccessible, all authorization decisions will be DENIED. Default: AWS owned key |
tags |
{ [key: string]: string } |
No | Tags for the policy engine (max 50 tags). Default: no tags |
Understanding Cedar Policies in AgentCore
Policies are constructed using Cedar language, an open source language for writing and enforcing authorization policies. Cedar policies in AgentCore follow a specific structure with three main components: Principal, Action, and Resource. Understanding how these components work together is critical for writing effective policies.
Policy Structure
Every Cedar policy has this basic structure:
permit( // or forbid
principal, // Who is making the request
action, // What operation they want to perform
resource // What Gateway/tool they want to access
)
when { // Optional conditions
// Additional constraints
};
Example Policy
permit(
principal,
action == AgentCore::Action::"ApplicationToolTarget___create_application",
resource == AgentCore::Gateway::"<gateway-arn>"
) when {
context.input.coverage_amount <= 1000000
};
Basic PolicyEngine and Policy Creation
Create a policy engine and add policies to it.
Policy Engine Mode
When associating a policy engine with a gateway, you can control the enforcement behavior using PolicyEngineMode:
PolicyEngineMode.LOG_ONLY(default) — evaluates actions and adds traces but does not enforce decisions. Use this mode for testing and validation before enabling enforcement.PolicyEngineMode.ENFORCE— actively allows or denies agent operations based on Cedar policy evaluation.
# Create a Policy engine
policy_engine = agentcore.PolicyEngine(self, "MyPolicyEngine",
policy_engine_name="my_policy_engine",
description="Policy engine for access control"
)
gateway = agentcore.Gateway(self, "MyGateway",
gateway_name="my-gateway",
policy_engine_configuration=agentcore.GatewayPolicyEngineConfig(
policy_engine=policy_engine,
mode=agentcore.PolicyEngineMode.ENFORCE
)
)
# Add policy to policy engine
policy_engine.add_policy("AllowAllActions",
definition=f"""
permit(
principal,
action,
resource == AgentCore::Gateway::\"{gateway.gatewayArn}\"
);
""",
description="Allow all actions on specific gateway (development)",
validation_mode=agentcore.PolicyValidationMode.IGNORE_ALL_FINDINGS
)
# you can add multiple policies to the policy engine
policy_engine.add_policy("SpecificToolPolicy",
definition=f"""
permit(
principal is AgentCore::OAuthUser,
action == AgentCore::Action::\"WeatherTool__get_forecast\",
resource == AgentCore::Gateway::\"{gateway.gatewayArn}\"
);
""",
description="Allow specific weather tool access",
validation_mode=agentcore.PolicyValidationMode.FAIL_ON_ANY_FINDINGS
)
Type-Safe Policy Builder
For a more type-safe approach, use the PolicyStatement builder instead of writing raw Cedar syntax.
gateway = agentcore.Gateway(self, "MyGateway",
gateway_name="my-gateway"
)
policy_engine = agentcore.PolicyEngine(self, "MyPolicyEngine",
policy_engine_name="my_policy_engine"
)
allow_all_policy = agentcore.Policy(self, "AllowAllPolicy",
policy_engine=policy_engine,
policy_name="allow_all",
statement=agentcore.PolicyStatement.permit().for_all_principals().on_all_actions().on_resource("AgentCore::Gateway", gateway.gateway_arn),
description="Allow all actions on specific gateway (development only)",
validation_mode=agentcore.PolicyValidationMode.IGNORE_ALL_FINDINGS
)
Policy with Specific Actions
# policy_engine: agentcore.PolicyEngine
# gateway: agentcore.Gateway
# Allow specific tool actions on specific gateway
# Action names follow pattern: "ToolName__operation"
policy_engine.add_policy("SpecificToolPolicy",
statement=agentcore.PolicyStatement.permit().for_principal("AgentCore::OAuthUser::your-client-id").on_actions(["AgentCore::Action::WeatherTool__get_forecast", "AgentCore::Action::WeatherTool__get_current"
]).on_resource("AgentCore::Gateway", gateway.gateway_arn),
description="Allow specific weather tool operations",
validation_mode=agentcore.PolicyValidationMode.FAIL_ON_ANY_FINDINGS
)
Policy with Conditions
Use when clauses to add advanced conditions based on principal tags (from OAuth token) or context:
# policy_engine: agentcore.PolicyEngine
# gateway: agentcore.Gateway
# Policy with when conditions using principal tags
conditional_policy = agentcore.Policy(self, "ConditionalPolicy",
policy_engine=policy_engine,
policy_name="conditional_access",
statement=agentcore.PolicyStatement.permit().for_principal("AgentCore::OAuthUser").on_all_actions().on_resource("AgentCore::Gateway", gateway.gateway_arn).when().principal_attribute("department").equal_to("Engineering").and().context_attribute("input.priority").equal_to("high").done(),
description="Allow engineers for high-priority requests",
validation_mode=agentcore.PolicyValidationMode.FAIL_ON_ANY_FINDINGS
)
Policy with Exclusions (unless)
Use unless clauses to exclude specific conditions from a policy. The policy applies when the unless conditions are NOT met:
# policy_engine: agentcore.PolicyEngine
# gateway: agentcore.Gateway
# Allow access unless the user is suspended
policy_with_unless = agentcore.Policy(self, "UnlessPolicy",
policy_engine=policy_engine,
policy_name="unless_suspended",
statement=agentcore.PolicyStatement.permit().for_principal("AgentCore::OAuthUser").on_all_actions().on_resource("AgentCore::Gateway", gateway.gateway_arn).unless().principal_attribute("suspended").equal_to(True).done(),
description="Allow all actions unless user is suspended",
validation_mode=agentcore.PolicyValidationMode.FAIL_ON_ANY_FINDINGS
)
You can combine when and unless clauses in the same policy:
# policy_engine: agentcore.PolicyEngine
# gateway: agentcore.Gateway
# Allow engineers unless they are on probation
policy_engine.add_policy("CombinedConditions",
statement=agentcore.PolicyStatement.permit().for_principal("AgentCore::OAuthUser").on_all_actions().on_resource("AgentCore::Gateway", gateway.gateway_arn).when().principal_attribute("department").equal_to("Engineering").done().unless().principal_attribute("status").equal_to("probation").done(),
description="Allow engineers unless on probation",
validation_mode=agentcore.PolicyValidationMode.FAIL_ON_ANY_FINDINGS
)
Forbid (Deny) Policy
Use forbid to explicitly deny access. Forbid policies override permit policies.
# policy_engine: agentcore.PolicyEngine
# gateway: agentcore.Gateway
# Explicitly deny dangerous tool operations
policy_engine.add_policy("DenyDangerous",
statement=agentcore.PolicyStatement.forbid().for_all_principals().on_action("AgentCore::Action::DeleteTool__delete_all").on_resource("AgentCore::Gateway", gateway.gateway_arn),
description="Forbid delete_all operation for all users",
validation_mode=agentcore.PolicyValidationMode.FAIL_ON_ANY_FINDINGS
)
Raw Cedar for Advanced Cases
For advanced Cedar features not supported by the builder, use raw Cedar strings:
# policy_engine: agentcore.PolicyEngine
# Option 1: Using definition property
advanced_policy = agentcore.Policy(self, "AdvancedPolicy",
policy_engine=policy_engine,
definition="permit(principal, action, resource) when { context.custom > 10 };",
description="Advanced policy with custom Cedar logic"
)
# Option 2: Using fromCedar() with statement property
policy_engine.add_policy("CustomPolicy",
statement=agentcore.PolicyStatement.from_cedar("forbid(principal, action, resource) when { resource.confidential == true };"),
description="Custom policy from Cedar string"
)
Note: You must specify either definition (raw Cedar string) or statement (PolicyStatement builder), but not both.
Accessing Policies on PolicyEngine
You can access the list of policies added to a PolicyEngine using policyEngine.policies.
PolicyEngine with KMS Encryption
Encrypt policy data with a custom KMS key.
# Create a custom KMS key
policy_key = kms.Key(self, "PolicyEngineKey",
enable_key_rotation=True,
description="KMS key for policy engine encryption"
)
# Create policy engine with encryption
policy_engine = agentcore.PolicyEngine(self, "EncryptedEngine",
policy_engine_name="encrypted_engine",
description="Policy engine with KMS encryption",
kms_key=policy_key
)
Importing Existing PolicyEngine
Import an existing policy engine from its ARN:
imported_engine = agentcore.PolicyEngine.from_policy_engine_attributes(self, "ImportedEngine",
policy_engine_arn="policy-engine-arn",
kms_key_arn="kms-arn"
)
# Use the imported engine
policy = agentcore.Policy(self, "PolicyForImportedEngine",
policy_engine=imported_engine,
definition="permit(principal, action, resource);"
)
Importing Existing Policy
Import an existing policy from its ARN:
imported_engine = agentcore.PolicyEngine.from_policy_engine_attributes(self, "ImportedEngine",
policy_engine_arn="policy-engine/my-engine-id"
)
imported_policy = agentcore.Policy.from_policy_attributes(self, "ImportedPolicy",
policy_arn="my-policy-arn",
policy_engine=imported_engine
)
# Grant permissions to the imported policy
role = iam.Role(self, "PolicyRole",
assumed_by=iam.ServicePrincipal("lambda.amazonaws.com")
)
imported_policy.grant_read(role)
PolicyEngine IAM Permissions
Grant various levels of access to policy engines:
policy_engine = agentcore.PolicyEngine(self, "MyEngine",
policy_engine_name="my_engine"
)
lambda_role = iam.Role(self, "LambdaRole",
assumed_by=iam.ServicePrincipal("lambda.amazonaws.com")
)
# Grant read permissions
policy_engine.grant_read(lambda_role)
# Grant evaluation permissions
policy_engine.grant_evaluate(lambda_role)
Using Policy with Stable Gateway
Since Gateway is now in aws-cdk-lib/aws-bedrockagentcore but Policy remains in alpha, use the L1 escape hatch to associate a policy engine with a stable gateway:
Proper L2 integration will be added in a future update.
import aws_cdk.aws_bedrockagentcore as agentcore
import aws_cdk.aws_bedrock_agentcore_alpha as agentcore_alpha
# Create policy engine (alpha)
policy_engine = agentcore_alpha.PolicyEngine(self, "Engine",
policy_engine_name="my_engine"
)
# Create gateway (stable)
gateway = agentcore.Gateway(self, "Gateway",
gateway_name="my-gateway"
)
# Wire policy engine to gateway via the L1 construct
cfn_gateway = gateway.node.default_child
cfn_gateway.policy_engine_configuration = agentcore.CfnGateway.GatewayPolicyEngineConfigurationProperty(
arn=policy_engine.policy_engine_arn,
mode=agentcore_alpha.PolicyEngineMode.ENFORCE.value
)
# Grant evaluate permissions to the gateway role
gateway.role.add_to_principal_policy(iam.PolicyStatement(
actions=["bedrock-agentcore:GetPolicyEngine"],
resources=[policy_engine.policy_engine_arn]
))
gateway.role.add_to_principal_policy(iam.PolicyStatement(
actions=["bedrock-agentcore:AuthorizeAction", "bedrock-agentcore:PartiallyAuthorizeActions"],
resources=[policy_engine.policy_engine_arn, gateway.gateway_arn]
))
TypeScript