![[アップデート] Amazon Bedrock AgentCore にPolicy機能(プレビュー)が追加されました #AWSreInvent](https://devio2024-media.developers.io/image/upload/f_auto,q_auto,w_3840/v1764709867/user-gen-eyecatch/luaqxeood29sjmqwzemm.png)
[アップデート] Amazon Bedrock AgentCore にPolicy機能(プレビュー)が追加されました #AWSreInvent
はじめに
こんにちは、コンサルティング部の神野です!
re:Invent 2025 の Keynote で Amazon Bedrock AgentCoreにPolicy機能(Amazon Bedorock AgentCore Policy)がプレビューで追加されることが発表されました!
何が嬉しいかというと今まではRuntime→Gateway→ツールを実行する際に、誰がどのツールを使えるかなどのきめ細かい認可制御は直近リリースされた、Interceptorといった別途Lambda関数を作ることでしかできませんでした。
それが今回のアップデートではこのPolicy機能で「誰が」、「どの行動を」、「何に対して」、ある条件の時に許可および拒否することを可能にするCedarといったポリシー言語で認可制御が可能になりました。(CedarはAmazon Verified Permissionsといった認可サービスでも使用されています)
Cedarは下記のように書けるポリシー言語です。下記ポリシーなら認証を受けたユーザーがprocess_refundツールをあるGateway経由で使用する時を対象にしています。また条件として、ロールはrefund-agentが対象で、amountが200未満の場合といった具合に書ける言語です。
permit(
principal is AgentCore::OAuthUser,
action == AgentCore::Action::"RefundTool__process_refund",
resource == AgentCore::Gateway::"<GATEWAY_ARN>"
)
when {
principal.hasTag("role") &&
principal.getTag("role") == "refund-agent" &&
context.input.amount < 200
};
ただこのポリシー作るの難しそう?と思った方にも朗報です。自然言語でCedarを作ることも可能です。自然言語でコンソール上で依頼すると、Gatewayが紐づいているMCP ServerやLambda関数のツール情報を読み取って下記のように適切なポリシーを作成することができます。
注文情報は参照可能で、更新や作成は不可にして欲しいです。

permitは参照系が許可されていて、forbidは作成や更新が不可のポリシーが出来上がっていました。
どこまで自然言語で対応できるかは不明ですが、叩きを作る上で便利ですね。
今回は速報レベルでポリシー評価機能を早速試してみたので共有していきたいと思います!
前提
今回は公式ドキュメントに記載があった、Getting started with AgentCore Policyを実施して挙動を確認していきます。
やってみる
まずはディレクトリを作って、仮想環境を有効にします。
mkdir agentcore-policy-quickstart
cd agentcore-policy-quickstart
python3 -m venv .venv
source .venv/bin/activate
その後は依存関係をインストールしていきます。
bedrock-agentcore-starter-toolkitはバージョンが0.21でPolicyに対応しているので、こちらを使用します。
pip install boto3
pip install bedrock-agentcore-starter-toolkit=0.21
pip install requests
次にGateway、CognitoとPolicyを作成するスクリプトをsetup_policy.pyで記載します。
"""
Setup script to create Gateway with Policy Engine
Run this first: python setup_policy.py
"""
from bedrock_agentcore_starter_toolkit.operations.gateway.client import GatewayClient
from bedrock_agentcore_starter_toolkit.operations.policy.client import PolicyClient
from bedrock_agentcore_starter_toolkit.utils.lambda_utils import create_lambda_function
import boto3
import json
import logging
import time
def setup_policy():
# Configuration
region = "us-west-2"
refund_limit = 1000
print("🚀 Setting up AgentCore Gateway with Policy Engine...")
print(f"Region: {region}\n")
# Initialize clients
gateway_client = GatewayClient(region_name=region)
gateway_client.logger.setLevel(logging.INFO)
policy_client = PolicyClient(region_name=region)
policy_client.logger.setLevel(logging.INFO)
# Step 1: Create OAuth authorizer
print("Step 1: Creating OAuth authorization server...")
cognito_response = gateway_client.create_oauth_authorizer_with_cognito("PolicyGateway")
print("✓ Authorization server created\n")
# Step 2: Create Gateway
print("Step 2: Creating Gateway...")
gateway = gateway_client.create_mcp_gateway(
name=None,
role_arn=None,
authorizer_config=cognito_response["authorizer_config"],
enable_semantic_search=False,
)
print(f"✓ Gateway created: {gateway['gatewayUrl']}\n")
# Fix IAM permissions
gateway_client.fix_iam_permissions(gateway)
print("⏳ Waiting 30s for IAM propagation...")
time.sleep(30)
print("✓ IAM permissions configured\n")
# Step 3: Create Lambda function with refund tool
print("Step 3: Creating Lambda function with refund tool...")
refund_lambda_code = """
def lambda_handler(event, context):
amount = event.get('amount', 0)
return {
"status": "success",
"message": f"Refund of ${amount} processed successfully",
"amount": amount
}
"""
session = boto3.Session(region_name=region)
lambda_arn = create_lambda_function(
session=session,
logger=gateway_client.logger,
function_name=f"RefundTool-{int(time.time())}",
lambda_code=refund_lambda_code,
runtime="python3.13",
handler="lambda_function.lambda_handler",
gateway_role_arn=gateway["roleArn"],
description="Refund tool for policy demo",
)
print("✓ Lambda function created\n")
# Step 4: Add Lambda target with refund tool schema
print("Step 4: Adding Lambda target with refund tool schema...")
lambda_target = gateway_client.create_mcp_gateway_target(
gateway=gateway,
name="RefundTarget",
target_type="lambda",
target_payload={
"lambdaArn": lambda_arn,
"toolSchema": {
"inlinePayload": [
{
"name": "process_refund",
"description": "Process a customer refund",
"inputSchema": {
"type": "object",
"properties": {
"amount": {
"type": "integer",
"description": "Refund amount in dollars"
}
},
"required": ["amount"],
},
}
]
},
},
credentials=None,
)
print("✓ Lambda target added\n")
# Step 5: Create Policy Engine
print("Step 5: Creating Policy Engine...")
engine = policy_client.create_or_get_policy_engine(
name="RefundPolicyEngine",
description="Policy engine for refund governance"
)
print(f"✓ Policy Engine: {engine['policyEngineId']}\n")
# Step 6: Create Cedar policy
print(f"Step 6: Creating Cedar policy (refund limit: ${refund_limit})...")
cedar_statement = (
f"permit(principal, "
f'action == AgentCore::Action::"RefundTarget___process_refund", '
f'resource == AgentCore::Gateway::"{gateway["gatewayArn"]}") '
f"when {{ context.input.amount < {refund_limit} }};"
)
policy = policy_client.create_or_get_policy(
policy_engine_id=engine["policyEngineId"],
name="refund_limit_policy",
description=f"Allow refunds under ${refund_limit}",
definition={"cedar": {"statement": cedar_statement}},
)
print(f"✓ Policy: {policy['policyId']}\n")
# Step 7: Attach Policy Engine to Gateway
print("Step 7: Attaching Policy Engine to Gateway (ENFORCE mode)...")
gateway_client.update_gateway_policy_engine(
gateway_identifier=gateway["gatewayId"],
policy_engine_arn=engine["policyEngineArn"],
mode="ENFORCE"
)
print("✓ Policy Engine attached to Gateway\n")
# Step 8: Save configuration
config = {
"gateway_url": gateway["gatewayUrl"],
"gateway_id": gateway["gatewayId"],
"gateway_arn": gateway["gatewayArn"],
"policy_engine_id": engine["policyEngineId"],
"policy_engine_arn": engine["policyEngineArn"],
"policy_id": policy["policyId"],
"region": region,
"client_info": cognito_response["client_info"],
"refund_limit": refund_limit
}
with open("config.json", "w") as f:
json.dump(config, f, indent=2)
print("=" * 60)
print("✅ Setup complete!")
print(f"Gateway URL: {gateway['gatewayUrl']}")
print(f"Policy Engine ID: {engine['policyEngineId']}")
print(f"Refund limit: ${refund_limit}")
print("\nConfiguration saved to: config.json")
print("\nNext step: Run 'python test_policy.py' to test your Policy")
print("=" * 60)
return config
if __name__ == "__main__":
setup_policy()
ここで作成しているリソースは下記となります。
- Cognito
- MCPツール化するLambda関数
- Cedarポリシーの作成
- AgentCore Gateway
- Inbound Authの設定(Cognito)
- Lambda関数をターゲットに紐付け
作成したポリシーは下記となります。テスト用のためか、誰でも実行可能で、使用できるツールはprocess_refundに限定、また対象のGatewayは今回作成するGatewayのみに絞る形です。
特徴としては条件が付与されていて、refund_limitがamountよりも下回った際しかこのアクションは実行されません。返金上限を超えた額を指定されても認めないイメージかと思います。
permit(
principal,
action == AgentCore::Action::"RefundTarget___process_refund",
resource == AgentCore::Gateway::"{gateway["gatewayArn"]}"
)
when {
context.input.amount < {refund_limit}
};
補足ですが、今回PolicyのEnforcement modeはENFORCEが指定されていますがこれはポリシーに適していないリクエストだった場合は拒否するかどうかの選択肢となります。ENFORCEなので拒否する形になります。
コンソール上にはこういった記載がありました。

Enabling enforcement: Consider testing and validating policies in log only mode before enabling enforcement to avoid unintended denials or adversely affecting production traffic.
適用を有効にする前に、ログのみのモードでポリシーを確認してくださいとあります。設計が意図通りになっているかは必ず事前にチェックするのが大事ですね。本番運用する前は必ず注意したいポイントです。
仮にログのみを使用したい場合はLOG_ONLYを指定します。
今回はENFORCEでそのまま実行してみます。
python setup_policy.py
🚀 Setting up AgentCore Gateway with Policy Engine...
Region: us-west-2
Step 1: Creating OAuth authorization server...
2025-12-03 05:25:23,089 - bedrock_agentcore.gateway - INFO - Starting EZ Auth setup: Creating Cognito resources...
2025-12-03 05:25:24,275 - bedrock_agentcore.gateway - INFO - ✓ Created User Pool: us-west-2_1B02xRJuB
2025-12-03 05:25:25,389 - bedrock_agentcore.gateway - INFO - ✓ Created domain: agentcore-2aa795bb
2025-12-03 05:25:25,390 - bedrock_agentcore.gateway - INFO - ⏳ Waiting for domain to be available...
2025-12-03 05:25:25,567 - bedrock_agentcore.gateway - INFO - ✓ Domain is active
2025-12-03 05:25:26,122 - bedrock_agentcore.gateway - INFO - ✓ Created resource server: PolicyGateway
2025-12-03 05:25:26,526 - bedrock_agentcore.
・・・
============================================================
✅ Setup complete!
Gateway URL: https://testgateway48856501-3oq3878x1q.gateway.bedrock-agentcore.us-west-2.amazonaws.com/mcp
Policy Engine ID: RefundPolicyEngine-dxtmb2vcin
Refund limit: $1000
Configuration saved to: config.json
Next step: Run 'python test_policy.py' to test your Policy
============================================================
実行してSetup completeと表示されていればOKです!上限金額は$1000で設定されていますね。
今度はテスト用のスクリプトを作成して、$1000 未満、$1000以上の金額を指定して検証してみます。
test_policy.pyを作成します。
"""
Test Policy Engine with direct HTTP calls to Gateway
Run after setup: python test_policy.py
"""
import json
import sys
import requests
from bedrock_agentcore_starter_toolkit.operations.gateway.client import GatewayClient
def test_refund(gateway_url, bearer_token, amount):
"""Test a refund request - print raw response"""
response = requests.post(
gateway_url,
headers={
"Content-Type": "application/json",
"Authorization": f"Bearer {bearer_token}",
},
json={
"jsonrpc": "2.0",
"id": 1,
"method": "tools/call",
"params": {
"name": "RefundTarget___process_refund",
"arguments": {"amount": amount}
},
},
)
print(f"Status Code: {response.status_code}")
print(f"Response Body: {json.dumps(response.json(), indent=2)}")
return response
def main():
print("=" * 60)
print("🧪 Testing Policy Engine")
print("=" * 60 + "\n")
# Load configuration
try:
with open("config.json", "r") as f:
config = json.load(f)
except FileNotFoundError:
print("❌ Error: config.json not found!")
print("Please run 'python setup_policy.py' first.")
sys.exit(1)
gateway_url = config["gateway_url"]
refund_limit = config["refund_limit"]
print(f"Gateway: {gateway_url}")
print(f"Refund limit: ${refund_limit}\n")
# Get access token
print("🔑 Getting access token...")
gateway_client = GatewayClient(region_name=config["region"])
token = gateway_client.get_access_token_for_cognito(config["client_info"])
print("✅ Token obtained\n")
# Test 1: Refund $500 (should be allowed)
print(f"📝 Test 1: Refund $500 (Expected: ALLOW)")
print("-" * 40)
test_refund(gateway_url, token, 500)
print()
# Test 2: Refund $2000 (should be denied)
print(f"📝 Test 2: Refund $2000 (Expected: DENY)")
print("-" * 40)
test_refund(gateway_url, token, 2000)
print()
print("=" * 60)
print("✅ Testing complete!")
print("=" * 60)
if __name__ == "__main__":
main()
アクセストークンを取得し、$500と$2000で返金処理を実行するイメージですね。試してみます。
python test_policy.py
Gateway: https://xxxx-xxx.gateway.bedrock-agentcore.us-west-2.amazonaws.com/mcp
Refund limit: $1000
🔑 Getting access token...
2025-12-03 05:28:36,473 - bedrock_agentcore.gateway - INFO - Fetching test token from Cognito...
2025-12-03 05:28:36,473 - bedrock_agentcore.gateway - INFO - Attempting to connect to token endpoint: https://agentcore-xxx.auth.us-west-2.amazoncognito.com/oauth2/token
2025-12-03 05:28:37,048 - bedrock_agentcore.gateway - INFO - ✓ Got test token successfully
✅ Token obtained
📝 Test 1: Refund $500 (Expected: ALLOW)
----------------------------------------
Status Code: 200
Response Body: {
"jsonrpc": "2.0",
"id": 1,
"result": {
"isError": false,
"content": [
{
"type": "text",
"text": "{\"status\":\"success\",\"message\":\"Refund of $500 processed successfully\",\"amount\":500}"
}
]
}
}
📝 Test 2: Refund $2000 (Expected: DENY)
----------------------------------------
Status Code: 200
Response Body: {
"jsonrpc": "2.0",
"id": 1,
"error": {
"code": -32002,
"message": "Tool Execution Denied: Tool call not allowed due to policy enforcement [Policy evaluation denied due to no determining policies]"
}
}
============================================================
✅ Testing complete!
============================================================
おおお、しっかりと$2000の場合はツール利用が拒否されましたね!Tool Execution Denied: Tool call not allowed due to policy enforcement [Policy evaluation denied due to no determining policies]と表示されています!
ツールの呼び出し時のポリシーエラーは下記形式で返ってくる形となります。
{
"jsonrpc": "2.0",
"id": 2,
"result": {
"content": [
{
"type": "text",
"text": "AuthorizeActionException - Tool Execution Denied: Tool call not allowed due to policy enforcement [Policy evaluation denied due to no determining policies]"
}
],
"isError": true
}
}
おわりに
今回はプレビューリリースされたPolicy機能を速報レベルで試してみました!
実際に自前のGatewayやツールをポリシーで検証して、適切に動くか試していきたいですね。
引き続き検証した結果はブログに記載します!
本記事が少しでも参考になりましたら幸いです!最後までご覧いただきありがとうございましたー!!








