はじめに
2026年3月3日、 Policy in Amazon Bedrock AgentCore が東京リージョンを含む13リージョンでGA(一般提供)されました。
AIエージェントは複数のツールを自律的に呼び出して複雑なタスクを遂行できます。
しかし、その柔軟性は同時にリスクでもあります。エージェントがビジネスルールを誤解釈したり、意図しない操作を実行したりする可能性があるためです。
Policy in AgentCoreは、エージェントのコードの外側でツール呼び出しを制御する仕組みです。
エージェントがツールを呼び出すたびにリクエストを傍受し、定義されたポリシーに基づいて許可/拒否を判定します。
本記事ではPolicy機能の概要を解説し、ハンズオンで実際にポリシーを作成・適用する手順を紹介します。
Policyとは
Policy in AgentCoreは、エージェントとツールの間にAgentCore Gatewayを配置し、すべてのツール呼び出しをポリシーで評価・認可する仕組みです。
主要コンポーネント
| コンポーネント | 役割 |
|---|---|
| AgentCore Gateway | エージェントとツール間のトラフィックを傍受するプロキシ |
| Policy Engine | Cedarポリシーを格納・評価するエンジン |
| Cedarポリシー | 許可/拒否ルールを定義する認可ポリシー |
アーキテクチャ
ENFORCEモードの動作原理
Policy EngineをGatewayにENFORCEモードでアタッチすると、以下のルールで動作します。
-
デフォルト全拒否: 明示的に
permitされていないアクションはすべて拒否 - permitで許可: 条件を満たすリクエストを明示的に許可
-
forbidが優先:
permitとforbidが両方マッチした場合、forbidが勝つ(forbid-winsセマンティクス) - CloudWatch連携: すべてのポリシー評価結果がログ・メトリクスとして記録
主要機能
Cedar言語によるポリシー定義
CedarはAWSが開発したオープンソースの認可ポリシー言語です。
シンプルな構文でありながら、きめ細かなアクセス制御を表現できます。
// $1000未満の返金のみ許可
permit(
principal,
action == AgentCore::Action::"RefundTarget___process_refund",
resource == AgentCore::Gateway::"<gateway-arn>"
)
when {
context.input.amount < 1000
};
ポリシーの構成要素:
-
permit/forbid: 許可または拒否を宣言 -
principal: リクエストを行うユーザー(OAuth認証済み) -
action: 対象のツール呼び出し(ターゲット名___ツール名の形式) -
resource: ポリシーが適用されるGatewayインスタンス -
when条件: ツール入力パラメータやユーザー属性に基づく追加条件
自然言語によるポリシー作成
Cedar構文を知らなくても、英語の自然言語プロンプトからポリシーを自動生成できます。
生成されたポリシーはツールスキーマに対して検証され、さらに自動推論によって以下の問題を検出します。
- 過度に許容的なポリシー
- 過度に制限的なポリシー
- 条件が絶対に満たされないポリシー
CloudWatch連携
すべてのポリシー評価結果がCloudWatchメトリクス・ログとして記録されます。
セキュリティチームや監査チームがエージェントの挙動を可視化・検証できます。
ハンズオン: 返金ツールにポリシーを適用する
公式チュートリアルをベースに、$1000未満の返金のみ許可するポリシーを作成します。
前提条件
- AWSアカウントとクレデンシャル設定済み
- Python 3.10以上
- IAM権限:
bedrock-agentcore:*、Lambda作成権限、IAMロール作成権限
Step 1: 環境セットアップ
mkdir agentcore-policy-quickstart
cd agentcore-policy-quickstart
python3 -m venv .venv
source .venv/bin/activate
pip install boto3 bedrock-agentcore-starter-toolkit requests
Step 2: セットアップスクリプト作成
setup_policy.pyを作成します。このスクリプトはGateway、Lambda(返金ツール)、Policy Engineを一括作成します。
"""
Gateway + Policy Engineのセットアップスクリプト
実行: 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():
region = "ap-northeast-1" # 東京リージョン
refund_limit = 1000
# クライアント初期化
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: OAuth認証サーバー作成
print("Step 1: OAuth認証サーバーを作成中...")
cognito_response = gateway_client.create_oauth_authorizer_with_cognito(
"PolicyGateway"
)
# Step 2: Gateway作成
print("Step 2: 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 URL: {gateway['gatewayUrl']}")
gateway_client.fix_iam_permissions(gateway)
print("IAM権限の反映を待機中(30秒)...")
time.sleep(30)
# Step 3: Lambda関数(返金ツール)作成
print("Step 3: Lambda関数を作成中...")
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",
)
# Lambda関数の準備完了を待機
print("Lambda関数の準備を待機中(10秒)...")
time.sleep(10)
# Step 4: Lambdaターゲット追加
print("Step 4: Lambdaターゲットを追加中...")
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,
)
# Step 5: Policy Engine作成
print("Step 5: Policy Engineを作成中...")
engine = policy_client.create_or_get_policy_engine(
name="RefundPolicyEngine",
description="Policy engine for refund governance",
)
print(f"Policy Engine ID: {engine['policyEngineId']}")
# Step 6: Cedarポリシー作成
print(f"Step 6: Cedarポリシーを作成中(返金上限: ${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 ID: {policy['policyId']}")
# Step 7: Policy EngineをGatewayにアタッチ(ENFORCEモード)
print("Step 7: Policy EngineをGatewayにアタッチ中(ENFORCEモード)...")
gateway_client.update_gateway_policy_engine(
gateway_identifier=gateway["gatewayId"],
policy_engine_arn=engine["policyEngineArn"],
mode="ENFORCE",
)
# 設定情報を保存
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("=" * 50)
print("セットアップ完了!")
print(f"Gateway URL: {gateway['gatewayUrl']}")
print(f"返金上限: ${refund_limit}")
print("設定ファイル: config.json")
print("次のステップ: python test_policy.py")
print("=" * 50)
return config
if __name__ == "__main__":
setup_policy()
処理の流れをまとめると以下のようになります。
Step 3: セットアップ実行
python setup_policy.py
完了まで2〜3分かかります。
Step 4: ポリシーのテスト
test_policy.pyを作成し、ポリシーが正しく動作するか検証します。
"""
Policy Engineのテストスクリプト
実行: 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):
"""返金リクエストをテスト"""
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: {response.status_code}")
print(f"Response: {json.dumps(response.json(), indent=2)}")
return response
def main():
# 設定ファイル読み込み
try:
with open("config.json", "r") as f:
config = json.load(f)
except FileNotFoundError:
print("config.jsonが見つかりません。先にsetup_policy.pyを実行してください。")
sys.exit(1)
gateway_url = config["gateway_url"]
refund_limit = config["refund_limit"]
print(f"Gateway: {gateway_url}")
print(f"返金上限: ${refund_limit}\n")
# アクセストークン取得
gateway_client = GatewayClient(region_name=config["region"])
token = gateway_client.get_access_token_for_cognito(config["client_info"])
# テスト1: $500の返金(許可されるはず)
print(f"Test 1: $500の返金 (期待値: 許可)")
print("-" * 40)
test_refund(gateway_url, token, 500)
print()
# テスト2: $2000の返金(拒否されるはず)
print(f"Test 2: $2000の返金 (期待値: 拒否)")
print("-" * 40)
test_refund(gateway_url, token, 2000)
if __name__ == "__main__":
main()
python test_policy.py
テスト結果
$500の返金(許可):
{
"jsonrpc": "2.0",
"id": 1,
"result": {
"isError": false,
"content": [
{
"type": "text",
"text": "{\"status\":\"success\",\"message\":\"Refund of $500 processed successfully\",\"amount\":500}"
}
]
}
}
$2000の返金(拒否):
{
"jsonrpc": "2.0",
"id": 2,
"error": {
"code": -32002,
"message": "Tool Execution Denied: Tool call not allowed due to policy enforcement [No policy applies to the request (denied by default).]"
}
}
amount < 1000の条件を満たさないリクエストはポリシーにマッチせず、デフォルト拒否(deny by default)されます。
エージェントのコードには一切手を加えていないのに、ポリシーだけでアクセス制御が実現できています。
クリーンアップ
"""リソース削除スクリプト: python cleanup_policy.py"""
from bedrock_agentcore_starter_toolkit.operations.gateway.client import GatewayClient
from bedrock_agentcore_starter_toolkit.operations.policy.client import PolicyClient
import json
import time
def cleanup():
with open("config.json", "r") as f:
config = json.load(f)
region = config["region"]
policy_client = PolicyClient(region_name=region)
gateway_client = GatewayClient(region_name=region)
# 1. Policyを削除し、完了を待機
policy_client.delete_policy(
policy_engine_id=config["policy_engine_id"],
policy_id=config["policy_id"],
)
print("Policy削除の完了を待機中...")
time.sleep(15)
# 2. Policy Engineを削除
policy_client.delete_policy_engine(
policy_engine_id=config["policy_engine_id"]
)
# 3. Gatewayを削除(ターゲットも自動削除される)
gateway_client.delete_gateway(
gateway_identifier=config["gateway_id"]
)
print("クリーンアップ完了")
if __name__ == "__main__":
cleanup()
ポリシー管理のCLI操作
作成済みのポリシーはaws bedrock-agentcore-controlコマンドで管理できます。
# Policy Engine一覧
aws bedrock-agentcore-control list-policy-engines
# ポリシー一覧
aws bedrock-agentcore-control list-policies --policy-engine-id <engine-id>
# ポリシー詳細(Cedar文を確認)
aws bedrock-agentcore-control get-policy \
--policy-engine-id <engine-id> \
--policy-id <policy-id>
# ポリシー更新
aws bedrock-agentcore-control update-policy \
--policy-id <policy-id> \
--policy-engine-id <engine-id> \
--definition '{"cedar": {"statement": "permit(principal, action, resource);"}}'
まとめ
Policy in Amazon Bedrock AgentCoreは、AIエージェントのツール利用に対してコードの外側から決定論的なアクセス制御を適用できる機能です。
| 特徴 | 内容 |
|---|---|
| 制御の粒度 | ユーザーID、ツール入力パラメータレベル |
| ポリシー言語 | Cedar(OSS) + 自然言語からの自動生成 |
| 適用方式 | Gateway境界での傍受(エージェントコード変更不要) |
| 監査 | CloudWatchメトリクス・ログ連携 |
| リージョン | 東京を含む13リージョン |
エージェントの自律性を活かしつつ、ガバナンスを確保する手段として有効です。
特にエンタープライズ環境でのエージェント運用において、セキュリティチームとの責任分界を明確にできる点が大きなメリットです。