Tutorial: Enforce Amazon Bedrock Guardrails with a custom Python policy

Beta

This feature is in Beta. Workspace admins can control access to this feature from the Previews page by opting into the Unity Gateway Service Policies preview. See Manage workspace-level previews.

This tutorial shows how to enforce a guardrail you configured in Amazon Bedrock Guardrails on traffic through Unity Gateway. A custom Python service policy sends each message to the Bedrock ApplyGuardrail API and blocks the interaction when Bedrock intervenes.

Bedrock authenticates with AWS Identity and Access Management (IAM), so the policy reaches it through a Unity Catalog service credential that assumes an IAM role in your AWS account.

In this tutorial, you:

Create an IAM role and a Unity Catalog service credential that can call your Bedrock guardrail.
Create the Python policy function.
Attach the policy to a service and test it.

Requirements

The workspace that sends requests to the service is in a region that supports custom Python service policies. See Availability.
The workspace that sends requests to the service doesn't use the compliance security profile, a serverless network policy that restricts internet access or blocks specific destinations, or a private network gateway. See Before you begin.
A guardrail in Amazon Bedrock with a published version, and its guardrail ID and version number. See Create a guardrail in the AWS documentation. Set the guardrail's actions to Block. A Python policy can't redact content, so a guardrail that masks sensitive information denies the whole interaction instead.
Permission in your AWS account to create an IAM role and policy.
The CREATE SERVICE CREDENTIAL privilege on the metastore, or a metastore admin who can create the service credential for you.
USE CATALOG, USE SCHEMA, and CREATE FUNCTION on the catalog and schema where you create the function. This tutorial uses main.governance.
MANAGE on the Model Service, Model Provider Service, or MCP Service you want to govern.

Step 1: Create a service credential for Bedrock

Follow Create a service credential that references an AWS IAM role. Name the credential bedrock-guardrail.

In the IAM role's permissions policy, allow the role to apply your guardrail. Replace the Region, account ID, and guardrail ID with your values:

JSON
{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": ["bedrock:ApplyGuardrail", "bedrock:GetGuardrail"],
      "Resource": "arn:aws:bedrock:<region>:<account-id>:guardrail/<guardrail-id>"
    }
  ]
}

If someone else creates the function in Step 2, grant them ACCESS on the credential:
SQL
```
GRANT ACCESS ON SERVICE CREDENTIAL `bedrock-guardrail` TO `<user-or-group>`;
```

Step 2: Create the policy function

Run the following SQL in a notebook or the SQL editor. Replace the Region, guardrail ID, and version with your guardrail's values. Use a published version, so the guardrail doesn't change while the policy uses it.

On a Model Service or Model Provider Service, the function sends the latest message to Bedrock as INPUT on the request phase and as OUTPUT on the response phase. On an MCP Service, it sends the tool name and arguments as INPUT before the tool runs. It doesn't check MCP tool results. It allows the interaction when Bedrock doesn't intervene. When Bedrock intervenes, it denies the interaction and names the denied topics, content filters, and sensitive information types that blocked it.

The function fails closed. It denies the interaction when the event or Bedrock's result is unexpected, or when the call to Bedrock fails. It waits up to 3 seconds to connect and 5 seconds for a response, and doesn't retry, so a slow guardrail call fails quickly instead of holding up the request.

SQL
CREATE OR REPLACE FUNCTION main.governance.bedrock_guardrail(
  event VARIANT,
  config VARIANT
)
RETURNS VARIANT
LANGUAGE PYTHON
CREDENTIALS (`bedrock-guardrail` DEFAULT)
ENVIRONMENT (environment_version = '6')
AS $$
import json
import boto3
from botocore.config import Config
from databricks.service_credentials import getServiceCredentialsProvider

REGION = "<region>"                        # for example, "us-east-1"
GUARDRAIL_ID = "<guardrail-id>"
GUARDRAIL_VERSION = "<guardrail-version>"  # a published version, for example "1"

def policy_decision(result, reason=""):
    return type(event).parseJson(json.dumps({"decision": {"result": result, "reason": reason}}))

def blocked_policies(response):
    names = set()
    for assessment in response.get("assessments", []):
        for topic in assessment.get("topicPolicy", {}).get("topics", []):
            if topic.get("action") == "BLOCKED" and topic.get("name"):
                names.add(topic["name"])
        for content_filter in assessment.get("contentPolicy", {}).get("filters", []):
            if content_filter.get("action") == "BLOCKED" and content_filter.get("type"):
                names.add(content_filter["type"])
        for pii in assessment.get("sensitiveInformationPolicy", {}).get("piiEntities", []):
            if pii.get("action") == "BLOCKED" and pii.get("type"):
                names.add(pii["type"])
    return ", ".join(sorted(names))

event_value = event.toPython() if hasattr(event, "toPython") else event
if not isinstance(event_value, dict):
    return policy_decision("DENY", "The policy received an invalid event.")
phase = event_value.get("type")
source = {"request": "INPUT", "response": "OUTPUT"}.get(phase)
if source is None:
    return policy_decision("DENY", "The policy received an unexpected event type.")

context = event_value.get("context") or {}
tool = context.get("tool")
if tool:
    # MCP Service: check the tool name and arguments before the tool runs.
    if phase != "request":
        return policy_decision("ALLOW")
    text = f"{tool.get('name', '')} {json.dumps(tool.get('arguments', {}))}"
else:
    # Model Service or Model Provider Service: check the latest message.
    text = str(context.get("message", ""))
if not text.strip():
    return policy_decision("ALLOW")

try:
    session = boto3.Session(botocore_session=getServiceCredentialsProvider("bedrock-guardrail"))
    client = session.client(
        "bedrock-runtime",
        region_name=REGION,
        config=Config(connect_timeout=3, read_timeout=5, retries={"total_max_attempts": 1}),
    )
    response = client.apply_guardrail(
        guardrailIdentifier=GUARDRAIL_ID,
        guardrailVersion=GUARDRAIL_VERSION,
        source=source,
        content=[{"text": {"text": text}}],
    )
except Exception as error:
    return policy_decision("DENY", f"Could not reach the Bedrock guardrail: {type(error).__name__}.")

action = response.get("action")
if action == "NONE":
    return policy_decision("ALLOW")
if action != "GUARDRAIL_INTERVENED":
    return policy_decision("DENY", "The Bedrock guardrail returned an unexpected result.")
blocked = blocked_policies(response)
return policy_decision("DENY", f"Blocked by the Bedrock guardrail ({blocked})." if blocked else "Blocked by the Bedrock guardrail.")
$$;

When the call fails, the reason names only the error type. The reason is returned to the caller, and an AWS error message can include your account ID and IAM role name.

The function must set environment_version = '6' because it declares a service credential. Without it, the function is created, but the policy fails with a CREDENTIALS_UNSUPPORTED_ENVIRONMENT error when it runs.

You can also create the function in the policy form instead of with SQL. In Step 3, under Function, select Create new and Python. Enter the function name. Expand Secrets and credentials (optional) and select bedrock-guardrail in Default credential (optional). Then paste the function body, the code between the $$ markers, into the editor.

Step 3: Attach and test the policy

In the workspace sidebar, click Unity Gateway.
Select the service to govern: a model service on the Models tab, a model provider service on the Providers tab, or an MCP service on the MCPs tab.
Open the Policies tab, then click New policy.
Enter a Name for the policy, such as bedrock_guardrail.
In Guardrail type, select Custom.
Set the Rank to control evaluation order. The lowest rank runs first on the request and last on the response.
Under Implementation, select Custom function.
Under Function, select Select existing, then select main.governance.bedrock_guardrail. In Compute for creating the function and testing, select a SQL warehouse.
Expand Test function. Under event, insert a sample event with text your guardrail should block. On a model service or model provider service, click Request and replace its message. On an MCP service, select a tool in Select tool and set one of its arguments. Click Run test, and confirm that Returns shows deny with a reason that names the Bedrock policy that intervened.
(Optional) Expand Advanced options and set Mode to Log to record decisions without blocking while you review them.
Click Create policy.

The first time the policy runs, the function's environment starts, and the policy can deny interactions until it's ready. See Python policy performance. To check that the policy is ready, send a request to the service that your guardrail should block, for example from AI Playground, and retry until the policy returns its verdict. In Enforce mode, Databricks blocks it and returns a successful (HTTP 200) response. On a model service or model provider service, the databricks_service_policy object carries the reason. On an MCP service, the response is a JSON-RPC error with the reason in error.message. See Policy decisions.

note

Test function runs on the SQL warehouse you select. To call Bedrock from a test on a serverless SQL warehouse, enable the Enable networking for isolated workloads in Serverless SQL Warehouses preview on your workspace's Previews page. Live policy evaluation doesn't use the warehouse.

Performance considerations

Each evaluation calls Bedrock, so the policy adds Bedrock's latency to the Python policy's own evaluation time, on every request and response it checks. Bedrock also charges for each ApplyGuardrail call. To limit latency and cost:

Attach the policy only to the services that need it.
Create the guardrail in the AWS Region closest to your workspace. A call to another Region adds network latency.
To check only one direction, return ALLOW early for the other phase. For example, to check only requests, return ALLOW when phase is response.

Clean up

Delete the policy from the service's Policies tab.

Drop the function:

SQL
DROP FUNCTION IF EXISTS main.governance.bedrock_guardrail;

If you no longer need it, delete the bedrock-guardrail service credential and the IAM role.

Requirements​

Step 1: Create a service credential for Bedrock​

Step 2: Create the policy function​

Step 3: Attach and test the policy​

Performance considerations​

Clean up​

Additional resources​