Policy Integration Examples

This guide demonstrates practical patterns for integrating Policy with AgentCore Gateway, including complete working examples, common policy patterns, and best practices.

Complete Example: Refund Processing System

This end-to-end example shows how to create a policy-protected refund processing system.

Important: Action Name Format

Action names in Cedar policies use the format TargetName___tool_name with three underscores (___): - Format: AgentCore::Action::"<TargetName>___<tool_name>" - Example: AgentCore::Action::"RefundTarget___process_refund" - The target name from your gateway and the tool name are separated by triple underscores

Step 1: Create Policy Engine

from bedrock_agentcore_starter_toolkit.operations.policy import PolicyClient

# Initialize the policy client
policy_client = PolicyClient(region_name='us-west-2')

# Create a policy engine
policy_engine = policy_client.create_or_get_policy_engine(
    name='RefundPolicyEngine',
    description='Policy engine for refund processing authorization'
)

print(f"Policy Engine ARN: {policy_engine['policyEngineArn']}")

Step 2: Create Cedar Policies

# Policy 1: Allow refund-agent to process refunds under $500
refund_policy = """
permit(
  principal is AgentCore::OAuthUser,
  action == AgentCore::Action::"RefundTarget___process_refund",
  resource == AgentCore::Gateway::"arn:aws:bedrock-agentcore:us-west-2:123456789012:gateway/refund-gateway"
)
when {
  principal.hasTag("username") &&
  principal.getTag("username") == "refund-agent" &&
  context.input.amount < 500
};
"""

policy_client.create_policy(
    policy_engine_id=policy_engine['policyEngineId'],
    name='RefundUnder500Policy',
    definition={'cedar': {'statement': refund_policy}},
    validation_mode='FAIL_ON_ANY_FINDINGS'
)

# Policy 2: Emergency shutdown - forbid all refunds
emergency_policy = """
forbid(
  principal,
  action == AgentCore::Action::"RefundTarget___process_refund",
  resource == AgentCore::Gateway::"arn:aws:bedrock-agentcore:us-west-2:123456789012:gateway/refund-gateway"
);
"""

# Note: This policy is created but can be enabled/disabled as needed

Step 3: Create Gateway with Policy Engine

from bedrock_agentcore_starter_toolkit.operations.gateway.client import GatewayClient
import json

gateway_client = GatewayClient(region_name='us-west-2')

# Define Lambda refund tool
lambda_config = {
    "arn": "arn:aws:lambda:us-west-2:123456789012:function:RefundProcessor",
    "tools": [
        {
            "name": "process_refund",
            "description": "Process customer refund",
            "inputSchema": {
                "type": "object",
                "properties": {
                    "orderId": {"type": "string"},
                    "amount": {"type": "integer"},
                    "reason": {"type": "string"}
                },
                "required": ["orderId", "amount"]
            }
        }
    ]
}

# Create Gateway with OAuth and Policy Engine
cognito = gateway_client.create_oauth_authorizer_with_cognito("refund-processor")

# Create the Gateway
gateway = gateway_client.create_mcp_gateway(
    name="refund-gateway",
    role_arn=None,  # Auto-creates IAM role
    authorizer_config=cognito['authorizer_config'],
    enable_semantic_search=False
)

# Add Lambda target
lambda_target = gateway_client.create_mcp_gateway_target(
    gateway=gateway,
    name="RefundTool",
    target_type="lambda",
    target_payload={
        "lambdaArn": "arn:aws:lambda:us-west-2:123456789012:function:RefundProcessor",
        "toolSchema": {
            "inlinePayload": [
                {
                    "name": "process_refund",
                    "description": "Process customer refund",
                    "inputSchema": {
                        "type": "object",
                        "properties": {
                            "orderId": {"type": "string"},
                            "amount": {"type": "integer"},
                            "reason": {"type": "string"}
                        },
                        "required": ["orderId", "amount"]
                    }
                }
            ]
        }
    },
    credentials=None
)

# Attach Policy Engine to Gateway
gateway_client.update_gateway_policy_engine(
    gateway_identifier=gateway["gatewayId"],
    policy_engine_arn=policy_engine['policyEngineArn'],
    mode="ENFORCE"
)

print(f"Gateway URL: {gateway['gatewayUrl']}")

Step 4: Test the Policy

import httpx
import asyncio

async def test_refund_policy():
    # Get OAuth token (from Cognito)
    token = gateway_client.get_access_token_for_cognito(cognito['client_info'])

    gateway_url = gateway['gatewayUrl']

    # Test 1: Valid refund (under $500)
    response1 = await httpx.AsyncClient().post(
        gateway_url,
        headers={"Authorization": f"Bearer {token}"},
        json={
            "jsonrpc": "2.0",
            "id": 1,
            "method": "tools/call",
            "params": {
                "name": "RefundTool__process_refund",
                "arguments": {
                    "orderId": "12345",
                    "amount": 450,
                    "reason": "Defective product"
                }
            }
        }
    )
    print(f"Test 1 (amount=450): {response1.json()}")  # Should ALLOW

    # Test 2: Invalid refund (over $500)
    response2 = await httpx.AsyncClient().post(
        gateway_url,
        headers={"Authorization": f"Bearer {token}"},
        json={
            "jsonrpc": "2.0",
            "id": 2,
            "method": "tools/call",
            "params": {
                "name": "RefundTool__process_refund",
                "arguments": {
                    "orderId": "12346",
                    "amount": 750,
                    "reason": "Defective product"
                }
            }
        }
    )
    print(f"Test 2 (amount=750): {response2.json()}")  # Should DENY

asyncio.run(test_refund_policy())

Common Policy Patterns

Amount-Based Restrictions

Limit operations based on monetary amounts:

# Natural language
nl_policy = "Allow users with scope payment:process to transfer funds when the amount is less than $10,000"

# Converts to Cedar
cedar_policy = """
permit(
  principal is AgentCore::OAuthUser,
  action == AgentCore::Action::"PaymentTarget___transfer_funds",
  resource == AgentCore::Gateway::"arn:aws:bedrock-agentcore:us-west-2:123456789012:gateway/payment"
)
when {
  principal.hasTag("scope") &&
  principal.getTag("scope") like "*payment:process*" &&
  context.input.amount < 10000
};
"""

User Tier-Based Access

Different limits for different user tiers:

# Premium users: transfers up to $50,000
premium_policy = """
permit(
  principal is AgentCore::OAuthUser,
  action == AgentCore::Action::"PaymentTarget___transfer_funds",
  resource == AgentCore::Gateway::"arn:aws:bedrock-agentcore:us-west-2:123456789012:gateway/payment"
)
when {
  principal.hasTag("tier") &&
  principal.getTag("tier") == "premium" &&
  context.input.amount < 50000
};
"""

# Standard users: transfers up to $10,000
standard_policy = """
permit(
  principal is AgentCore::OAuthUser,
  action == AgentCore::Action::"PaymentTarget___transfer_funds",
  resource == AgentCore::Gateway::"arn:aws:bedrock-agentcore:us-west-2:123456789012:gateway/payment"
)
when {
  principal.hasTag("tier") &&
  principal.getTag("tier") == "standard" &&
  context.input.amount < 10000
};
"""

Regional Restrictions

Restrict operations to specific regions:

# Allow only for US, CA, UK regions
regional_policy = """
permit(
  principal is AgentCore::OAuthUser,
  action == AgentCore::Action::"ShippingTarget___calculate_rate",
  resource == AgentCore::Gateway::"arn:aws:bedrock-agentcore:us-west-2:123456789012:gateway/shipping"
)
when {
  context.input has region &&
  ["US", "CA", "UK"].contains(context.input.region)
};
"""

Role-Based Access Control

Control access based on user roles:

# Allow managers to approve high-value decisions
manager_policy = """
permit(
  principal is AgentCore::OAuthUser,
  action == AgentCore::Action::"DecisionTarget___approve_decision",
  resource == AgentCore::Gateway::"arn:aws:bedrock-agentcore:us-west-2:123456789012:gateway/decision"
)
when {
  principal.hasTag("role") &&
  ["manager", "director"].contains(principal.getTag("role")) &&
  context.input.amount > 100000
};
"""

Required Field Validation

Enforce that optional parameters are provided:

# Require description for all claims
required_description_policy = """
forbid(
  principal is AgentCore::OAuthUser,
  action == AgentCore::Action::"InsuranceTarget___file_claim",
  resource == AgentCore::Gateway::"arn:aws:bedrock-agentcore:us-west-2:123456789012:gateway/insurance"
)
unless {
  context.input has description
};
"""

Emergency Shutdown Patterns

Disable All Tools

emergency_shutdown = """
forbid(
  principal,
  action,
  resource
);
"""

Disable Specific Tool

disable_tool = """
forbid(
  principal,
  action == AgentCore::Action::"PaymentTarget___transfer_funds",
  resource == AgentCore::Gateway::"arn:aws:bedrock-agentcore:us-west-2:123456789012:gateway/payment"
);
"""

Block Specific User

block_user = """
forbid(
  principal is AgentCore::OAuthUser,
  action,
  resource
)
when {
  principal.hasTag("username") &&
  principal.getTag("username") == "suspended-user"
};
"""

Natural Language Policy Authoring

Using the Policy Authoring Service

# Generate Cedar policy from natural language (automatic polling & fetching)
# Note: name must match pattern ^[A-Za-z][A-Za-z0-9_]*$ (use underscores, not hyphens)
result = policy_client.generate_policy(
    policy_engine_id=policy_engine['policyEngineId'],
    name='refund_policy',  # Use underscores, not hyphens
    resource={'arn': gateway['gatewayArn']},
    content={'rawText': 'Allow refunds for amounts less than $500'},
    fetch_assets=True  # Automatically fetches generated policies
)

# Display generated Cedar policies
for policy_asset in result.get('generatedPolicies', []):
    cedar_statement = policy_asset['definition']['cedar']['statement']
    print(f"Generated Cedar Policy:\n{cedar_statement}")

Natural Language Examples

# Example 1: Simple amount restriction
nl1 = "Allow users to process payments when the amount is less than $1000"

# Example 2: Role and amount combined
nl2 = "Allow users with role manager to approve expenses when the amount exceeds $5000"

# Example 3: Regional restrictions
nl3 = "Allow all users to ship packages when the destination country is US, CA, or UK"

# Example 4: Required fields
nl4 = "Block users from filing claims unless a description and priority are provided"

# Example 5: Scope-based
nl5 = "Allow users with scope admin:write to update user profiles when the account is verified"

Important: When using generate_policy() or start_policy_generation, the name parameter must follow these rules: - Only letters, numbers, and underscores allowed - Must start with a letter - No hyphens, dots, or special characters - Pattern: ^[A-Za-z][A-Za-z0-9_]*$

Testing and Debugging Policies

LOG_ONLY Mode for Testing

# Create gateway in LOG_ONLY mode for testing
test_cognito = gateway_client.create_oauth_authorizer_with_cognito("test-gateway")
test_gateway = gateway_client.create_mcp_gateway(
    name="test-gateway",
    role_arn=None,
    authorizer_config=test_cognito['authorizer_config'],
    enable_semantic_search=False
)

# Add your Lambda target
test_target = gateway_client.create_mcp_gateway_target(
    gateway=test_gateway,
    name="TestTarget",
    target_type="lambda",
    target_payload={"lambdaArn": "arn:aws:lambda:us-west-2:123456789012:function:TestFunction"},
    credentials=None
)

# Attach Policy Engine in LOG_ONLY mode
gateway_client.update_gateway_policy_engine(
    gateway_identifier=test_gateway["gatewayId"],
    policy_engine_arn=policy_engine['policyEngineArn'],
    mode="LOG_ONLY"  # Test without enforcing
)

# All requests will be allowed, but policy decisions are logged

Policy Validation

# Policies are automatically validated on creation
try:
    policy_client.create_policy(
        policy_engine_id=policy_engine['policyEngineId'],
        name='TestPolicy',
        definition={'cedar': {'statement': invalid_cedar}},
        validation_mode='FAIL_ON_ANY_FINDINGS'
    )
except Exception as e:
    print(f"Validation failed: {e}")
    # Fix the policy and try again

Common Pitfalls and Solutions

Pitfall 1: Invalid Generation Name

Problem: Using hyphens or special characters in policy generation names

Solution: Use only letters, numbers, and underscores; must start with a letter

# ❌ Wrong: Contains hyphens
policy_client.generate_policy(
    name='refund-policy-v1',  # ValidationException
    ...
)

# ✅ Correct: Uses underscores
policy_client.generate_policy(
    name='refund_policy_v1',
    ...
)

Pitfall 2: Forgetting Default Deny

Problem: Expecting actions to be allowed without a permit policy

Solution: Always create explicit permit policies for allowed actions

# ❌ Wrong: No permit policy, everything denied
forbid_policy = """
forbid(principal, action, resource)
when { context.input.amount > 1000 };
"""

# ✅ Correct: Explicit permit for valid cases
permit_policy = """
permit(principal, action, resource)
when { context.input.amount <= 1000 };
"""

Pitfall 3: Vague Conditions

Problem: Using subjective terms in conditions

Solution: Use precise, testable conditions

# ❌ Wrong: Vague term "reasonable"
"Allow transfers when the amount is reasonable"

# ✅ Correct: Specific threshold
"Allow transfers when the amount is less than $10,000"

Pitfall 4: Missing Tag Checks

Problem: Accessing tags without checking if they exist

Solution: Always use hasTag() before getTag()

# ❌ Wrong: May fail if tag doesn't exist
when { principal.getTag("role") == "admin" }

# ✅ Correct: Check existence first
when {
  principal.hasTag("role") &&
  principal.getTag("role") == "admin"
}

Pitfall 5: Incorrect Resource Scope

Problem: Using type check with specific actions

Solution: Use specific Gateway ARN when specifying tools

# ❌ Wrong: Type check with specific action
resource is AgentCore::Gateway,
action == AgentCore::Action::"SpecificTarget___specific_tool"

# ✅ Correct: Specific Gateway ARN with specific action
resource == AgentCore::Gateway::"arn:aws:bedrock-agentcore:region:account:gateway/id",
action == AgentCore::Action::"SpecificTarget___specific_tool"

Policy Management Best Practices

Version Control Your Policies

# Store policies in version control
policy_definitions = {
    'refund_under_500': {
        'version': '1.0.0',
        'cedar': refund_policy,
        'description': 'Allow refunds under $500'
    },
    'emergency_shutdown': {
        'version': '1.0.0',
        'cedar': emergency_policy,
        'description': 'Emergency refund shutdown'
    }
}

# Deploy from version control
for name, config in policy_definitions.items():
    policy_client.create_or_get_policy(
        policy_engine_id=policy_engine['policyEngineId'],
        name=name,
        definition={'cedar': {'statement': config['cedar']}},
        description=f"{config['description']} (v{config['version']})"
    )

Organize Policies by Purpose

# Group policies by functionality
policies = {
    'authentication': [...],   # Identity-based policies
    'authorization': [...],    # Role/permission-based policies
    'business_rules': [...],   # Amount limits, regional restrictions
    'emergency': [...]         # Shutdown and incident response
}

Test Before Deploying

# 1. Create test gateway with LOG_ONLY
# 2. Run test suite
# 3. Review CloudWatch logs
# 4. Switch to ENFORCE mode

def test_policy_suite(gateway_url, test_cases):
    """Run comprehensive policy tests"""
    results = []
    for test in test_cases:
        response = call_gateway_tool(
            gateway_url,
            test['token'],
            test['tool'],
            test['args']
        )
        results.append({
            'test': test['name'],
            'expected': test['expected'],
            'actual': 'ALLOW' if not response.get('isError') else 'DENY',
            'passed': (not response.get('isError')) == (test['expected'] == 'ALLOW')
        })
    return results

Cleanup

# Delete policies
for policy in policy_client.list_policies(policy_engine['policyEngineId']):
    policy_client.delete_policy(
        policy_engine_id=policy_engine['policyEngineId'],
        policy_id=policy['policyId']
    )

# Delete policy engine (must be detached from gateways first)
policy_client.delete_policy_engine(policy_engine['policyEngineId'])

Next Steps

• Policy Overview - Understand Policy concepts
• Policy Quickstart - Get started quickly
• Cedar Documentation - Learn Cedar language
• Gateway Integration - More Gateway examples