Write a custom Python service 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.

A custom Python service policy is a Python function registered in Unity Catalog that decides whether each request or response to an AI service is allowed. Use Python when a rule is hard to express in SQL, when it needs a Python library, or when it needs to call another service, such as a guardrail service you already use.

You attach a Python policy to a Model Service, Model Provider Service, or MCP Service, the same way you attach any service policy.

When to use a Python policy

Choose the simplest policy type that expresses your rule:

A built-in guardrail covers common risks, such as unsafe content and sensitive data, with no code.
A custom LLM-as-a-judge policy handles semantic rules you can describe in plain language.
A custom SQL policy handles simple deterministic checks on fields such as the tool name or the caller's identity.
An external service policy enforces a guardrail vendor that implements the Databricks external policy API.

Use a Python policy for logic the other types can't express. Common uses include:

Keyword and pattern filtering: block requests or responses that match regular expressions, such as internal hostnames, ticket IDs, or credential formats.
Custom sensitive data checks: detect identifiers specific to your organization, such as employee or customer numbers, that the built-in Sensitive Data Detection policy doesn't cover.
Tool argument validation: check MCP tool call arguments in detail, for example to allow a write only to approved repositories or paths.
Python packages: use a package from PyPI, such as a parser or a classifier. See Add Python packages.
Calls to a partner guardrail or internal service: call an external service, such as a guardrail, with a Unity Catalog service credential, then return its verdict. See Call an external service with a service credential.

Availability

Custom Python service policies are in Beta and are available in the following regions:

us-east-1
us-west-2

Custom Python service policies aren't available in other regions.

How a Python policy works

A Python policy is a Unity Catalog Python UDF with a fixed signature. It takes two VARIANT parameters and returns a VARIANT:

SQL
CREATE OR REPLACE FUNCTION <catalog>.<schema>.<function_name>(
  event VARIANT,
  config VARIANT
)
RETURNS VARIANT
LANGUAGE PYTHON
ENVIRONMENT (environment_version = '6')
AS $$
  <function body>
$$;

event is the interaction under evaluation. It has the same fields as the event a SQL policy receives, such as type (request or response), context.message (the latest user or assistant message), context.tool.name and context.tool.arguments on an MCP Service, and the caller's identity in context.actor. See Event fields.
config holds the options stored on the policy's attachment, or an empty object when it has none. Your function must declare it, even if it doesn't use it.
The function returns a decision: ALLOW, DENY, or ASK, with an optional reason. ASK holds an MCP tool call for human approval on the request phase. On an MCP tool result, ASK lets the result through, so return DENY to block it. On a Model Service or Model Provider Service, ASK blocks the interaction, and the caller sees a message that ASK isn't supported instead of your reason, so return DENY there.

Databricks runs the function on serverless compute that it manages, not on your SQL warehouse or cluster. The function runs at both evaluation points, the request (input) and the response (output), so branch on type to act on one phase. If the function raises an error, times out, or returns an unrecognized decision, Databricks denies the interaction.

Before you begin

The Unity Gateway Service Policies preview must be turned on for your workspace. A workspace admin turns it on from the Previews page. See Manage workspace-level previews.
The workspace that sends requests to the service must be in a supported region. In other regions, a Python policy denies every interaction it evaluates in Enforce mode. In Log mode, the interaction proceeds and the would-be denial is recorded.
The workspace that sends requests to the service must not use the compliance security profile, a serverless network policy that restricts internet access or blocks specific destinations, or a private network gateway. In these workspaces, a Python policy denies every interaction it evaluates in Enforce mode. In Log mode, the interaction proceeds and the would-be denial is recorded. The check applies to the workspace that sends the request, not the workspace where you attach the policy.
To create the function: USE CATALOG and USE SCHEMA on the target catalog and schema, and CREATE FUNCTION on the schema.
To attach the policy: MANAGE on the target service and EXECUTE on the function. Callers of the service don't need EXECUTE on the function.
To call an external cloud service from the function: a Unity Catalog service credential for that service, and ACCESS on the credential to create a function that declares it.
To read a secret from the function: a Unity Catalog secret.

Create a Python policy in the policy form

You can write, check, and test the function without leaving the policy form.

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.
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 Create new. In Compute for creating the function and testing, select a SQL warehouse. Databricks uses it only to create the function and run tests, not to evaluate live traffic.
Select Python. To start from a template, select one, such as Block unsafe output or Block prompt injection.
In Function name, enter the full name, for example main.governance.block_internal_hosts.
In Dependencies (optional), list the Python packages the function needs that aren't already in the environment, one per line, for example phonenumbers. Databricks adds them to the function's ENVIRONMENT clause.
(Optional) To use a secret or a service credential, expand Secrets and credentials (optional):
- In Secrets (optional), list the Unity Catalog secrets the function reads, one per line, for example main.default.api_key.
- In Default credential (optional), select the service credential the function uses. To add more credentials, click Add, then select each credential and give it an Alias.
Databricks adds them to the function's SECRETS and CREDENTIALS clauses. See Use a secret and Call an external service with a service credential.
Write the function body in the editor, then click Check syntax.
Expand Test function. Under event, insert a sample event. On a Model Service or Model Provider Service, click Request or Response. On an MCP Service, select a tool in Select tool to insert a tool call with that tool's arguments. Edit the event to match the case you want to test. Leave config as {}. Click Run test. Returns shows the decision, and the reason when there is one.
Click Create policy.

The function is created in Unity Catalog and the policy appears on the service's Policies tab. The function's dependencies install the first time the policy runs, and the policy denies interactions until the installation finishes. See Performance. To reuse a function you already created, select Select existing under Function instead of Create new.

Create a Python policy with SQL

You can also create the function with CREATE FUNCTION in a notebook or the SQL editor, then attach it with Select existing in the policy form.

The following function runs on a Model Service or Model Provider Service. It blocks responses that contain an internal hostname or an access key pattern. It reads the message from context.message and returns a decision built by the policy_decision helper:

SQL
CREATE OR REPLACE FUNCTION main.governance.block_internal_hosts(
  event VARIANT,
  config VARIANT
)
RETURNS VARIANT
LANGUAGE PYTHON
ENVIRONMENT (environment_version = '6')
AS $$
import json
import re

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

event_value = event.toPython() if hasattr(event, "toPython") else event
if event_value.get("type") != "response":
    return policy_decision("ALLOW")

message = str(event_value.get("context", {}).get("message", ""))
patterns = {
    "an internal hostname": r"\b[a-z0-9-]+\.corp\.example\.com\b",
    "an access key": r"\bAKIA[0-9A-Z]{16}\b",
}
for label, pattern in patterns.items():
    if re.search(pattern, message, flags=re.IGNORECASE):
        return policy_decision("DENY", f"The response contains {label}.")
return policy_decision("ALLOW")
$$;

To attach it, follow Create a Python policy in the policy form, but select Select existing under Function and choose the function.

Write the function

Read the event

The event and config parameters are VARIANT values. Convert event to a Python dictionary with toPython(), then read fields with get(), so a missing field doesn't raise an error:

Python
event_value = event.toPython() if hasattr(event, "toPython") else event
phase = event_value.get("type")                                  # "request" or "response"
message = event_value.get("context", {}).get("message", "")      # latest user or assistant message
tool = event_value.get("context", {}).get("tool", {})            # MCP Services only

For the full list of fields, see Event fields.

Return a decision

Return a VARIANT that holds a decision object with a result of ALLOW, DENY, or ASK and an optional reason. The result is case-insensitive. Build the return value from a dictionary with type(event).parseJson():

Python
import json

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

return policy_decision("DENY", "The response contains an internal hostname.")

If the function returns DENY without a reason, or with an empty one, the caller sees the default reason: Access denied: this request is not permitted by a policy on this service.

When the policy denies an interaction on a Model Service or Model Provider Service, the caller receives the reason in the databricks_service_policy object. On an MCP Service, the caller receives a JSON-RPC error with the reason in error.message. See Policy decisions.

Add Python packages

To use a package that isn't in the environment, list it in Dependencies (optional) in the policy form, or declare it in the dependencies option of the ENVIRONMENT clause when you create the function with SQL. Keep environment_version = '6':

SQL
ENVIRONMENT (
  dependencies = '["phonenumbers"]',
  environment_version = '6'
)

Packages that are already in environment version 6, such as boto3 and requests, don't need to be listed. Dependencies are installed from PyPI. See Extend UDFs using custom dependencies.

Call an external service with a service credential

To call a cloud service, declare a Unity Catalog service credential in the CREDENTIALS clause and use it through databricks.service_credentials.getServiceCredentialsProvider():

SQL
CREDENTIALS (`my-credential` DEFAULT)
ENVIRONMENT (environment_version = '6')

For example, the following code creates an Amazon Bedrock client from the credential. boto3 is already in the environment:

Python
import boto3
from databricks.service_credentials import getServiceCredentialsProvider

session = boto3.Session(botocore_session=getServiceCredentialsProvider("my-credential"))
client = session.client("bedrock-runtime", region_name="us-east-1")

A function that declares a service credential must set environment_version = '6'. If you omit the ENVIRONMENT clause, the function is created, but the policy fails with a CREDENTIALS_UNSUPPORTED_ENVIRONMENT error when it runs. For credential requirements, see Use a service credential in a Python UDF.

For a complete example, see Tutorial: Enforce Amazon Bedrock Guardrails with a custom Python policy.

Use a secret

To use a value such as an API key, store it as a Unity Catalog secret. Declare the secret's three-part name in the SECRETS clause, or list it in Secrets (optional) in the policy form, then read it in the function body with databricks.secrets.get(). The function must set environment_version = '6':

SQL
SECRETS (main.default.api_key)
ENVIRONMENT (environment_version = '6')

Python
from databricks.secrets import get

api_key = get(catalog="main", schema="default", key="api_key")

The function can read only the secrets declared in its SECRETS clause. Don't include a secret value in the reason you return, because the reason is returned to the caller. See Use secrets in a Python UDF.

Handle errors

A Python policy fails closed, so an unhandled exception denies the interaction with the generic reason Access denied: a policy on this service could not be evaluated. To return a clearer reason, catch exceptions around calls that can fail, such as network calls, and return DENY with your own explanation.

Test a policy function in a notebook

A Python policy fails closed, so a coding error in a live policy shows up only as a denied interaction with a generic reason. To find errors before you attach the policy, test the function body in a notebook. The following code runs the body the same way a Python policy does, as a function of event and config, and checks the following:

Code: The body compiles, and its imports load.
Contract: The return value is a VARIANT that holds a decision object with a result of ALLOW, DENY, or ASK.
Verdict: The decision for a sample request and a sample response matches the one you expect.

Create a notebook and attach it to serverless compute with environment version 6, the environment that Python policies use.
If the function uses packages that aren't in the environment, install them in the first cell with the same versions as the function's dependencies, for example %pip install phonenumbers.

In the next cell, paste the function body into POLICY. You can also paste the whole CREATE FUNCTION statement. Then edit REQUEST, RESPONSE, and EXPECTED for the cases you want to test. The cell is pre-filled with the function from Create a Python policy with SQL:

Python
# Paste the function body (the code between AS $$ and $$), or the whole CREATE FUNCTION statement.
POLICY = r'''
import json
import re

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

event_value = event.toPython() if hasattr(event, "toPython") else event
if event_value.get("type") != "response":
    return policy_decision("ALLOW")

message = str(event_value.get("context", {}).get("message", ""))
patterns = {
    "an internal hostname": r"\b[a-z0-9-]+\.corp\.example\.com\b",
    "an access key": r"\bAKIA[0-9A-Z]{16}\b",
}
for label, pattern in patterns.items():
    if re.search(pattern, message, flags=re.IGNORECASE):
        return policy_decision("DENY", f"The response contains {label}.")
return policy_decision("ALLOW")
'''

# Sample events. Edit them to test the cases you care about.
REQUEST = {"type": "request", "context": {"message": "Where is the build server?"}}
RESPONSE = {"type": "response", "context": {"message": "The build server is build-01.corp.example.com."}}
# For an MCP Service, test a tool call instead, for example:
# REQUEST = {"type": "request", "context": {"tool": {"name": "push_files", "arguments": {"owner": "acme", "repo": "website"}}}}
CONFIG = {}

# The verdict you expect for each event: "ALLOW", "DENY", or "ASK". Use None to skip the check.
EXPECTED = {"request": "ALLOW", "response": "DENY"}

Paste the following code into the last cell and run it:

Python
import ast
import json
import linecache
import re
import textwrap
import traceback

from pyspark.sql.types import VariantVal


def load_policy(source):
    """Compile the body as the body of policy(event, config), the way a Python policy runs it."""
    match = re.search(r"\bAS\s*\$\$(.*?)\$\$", source, flags=re.DOTALL | re.IGNORECASE)
    body = textwrap.dedent(match.group(1) if match else source).strip("\n") + "\n"
    if not body.strip():
        print("Code: FAIL. The function body is empty.")
        return None
    try:
        wrapper = ast.parse("def policy(event, config):\n    pass").body[0]
        wrapper.body = ast.parse(body, filename="<policy>").body
        code = compile(ast.Module(body=[wrapper], type_ignores=[]), "<policy>", "exec")
    except SyntaxError as error:
        print(f"Code: FAIL. Syntax error on line {error.lineno} of the function body: {error.msg}")
        print(f"  {(error.text or '').strip()}")
        return None
    # Show the body's own lines in tracebacks.
    linecache.cache["<policy>"] = (len(body), None, body.splitlines(True), "<policy>")
    namespace = {}
    exec(code, namespace)
    print("Code: PASS. The function body compiles.")
    return namespace["policy"]


def check_contract(value):
    """Return (result, reason, problem). problem is None when the value follows the decision contract."""
    if not isinstance(value, VariantVal):
        return None, None, "The function must return a VARIANT. Build it with type(event).parseJson(json.dumps(...))."
    data = value.toPython()
    decision = data.get("decision") if isinstance(data, dict) else None
    if not isinstance(decision, dict):
        return None, None, 'The return value must hold a "decision" object.'
    result, reason = decision.get("result"), decision.get("reason", "")
    if not isinstance(result, str) or result.upper() not in ("ALLOW", "DENY", "ASK"):
        return None, None, f'"decision.result" must be ALLOW, DENY, or ASK, not {result!r}.'
    if not isinstance(reason, str):
        return None, None, '"decision.reason" must be a string.'
    return result.upper(), reason, None


def run_case(policy, name, event, config, expected):
    """Run the policy on one sample event. Return True when every check passes."""
    try:
        value = policy(VariantVal.parseJson(json.dumps(event)), VariantVal.parseJson(json.dumps(config)))
    except ModuleNotFoundError as error:
        print(f"{name}: FAIL. Module {error.name!r} isn't installed. Install it with %pip install, "
              "and list it in the function's dependencies.")
        return False
    except Exception:
        print(f"{name}: FAIL. The function raised an error, so the policy would deny the interaction:")
        traceback.print_exc()
        return False
    result, reason, problem = check_contract(value)
    if problem:
        print(f"{name}: FAIL. {problem} The policy would deny the interaction.")
        return False
    print(f"{name}: {result}" + (f" ({reason})" if reason else ""))
    if result == "ASK" and "tool" not in event.get("context", {}):
        print("  Note: Model Services and Model Provider Services can't ask for approval, so here ASK blocks the interaction.")
    elif result == "ASK" and event.get("type") != "request":
        print("  Note: on an MCP tool result, ASK lets the result through. Return DENY to block it.")
    if expected and result != expected.upper():
        print(f"  FAIL. Expected {expected.upper()}.")
        return False
    return True


policy = load_policy(POLICY)
if policy:
    checks = [run_case(policy, name, event, CONFIG, EXPECTED.get(name))
              for name, event in [("request", REQUEST), ("response", RESPONSE)]]
    print("All checks passed." if all(checks) else "Some checks failed.")

With the pre-filled function and events, the output is:

Text
Code: PASS. The function body compiles.
request: ALLOW
response: DENY (The response contains an internal hostname.)
All checks passed.

When a check fails, the output names the check. For an error in the function, it also shows the Python traceback, with line numbers that match the function body you pasted.

The notebook runs the function body directly, not through a policy, so it doesn't test the function's CREDENTIALS clause or calls that use a service credential. Test those with Test function in the policy form.

Performance

A Python policy adds latency to every request and response it evaluates, so test it on realistic traffic before you enforce it. Expect the following latency:

First evaluations: Databricks installs the function's dependencies the first time the policy runs, not when you attach it. Until the installation finishes, the policy denies each interaction it evaluates with Access denied: the policy function could not start in time. Try the request again. How long the installation takes depends on the dependencies. With a small package, a retry soon succeeds. A large package can take much longer. If you change the function's dependencies, they install again.
Later evaluations: After the dependencies are installed, an evaluation typically takes 1 to 2 seconds, depending on the function and its dependencies.
External calls: A call to an external service adds that service's latency.

To check whether the policy is ready, send a request that it should deny, for example from AI Playground, and retry until the policy returns its own verdict. For a policy in Log mode, check the would-be verdict in the unified trace table.

To reduce the impact, keep functions small, limit dependencies, and attach a Python policy only to the services that need it. Databricks recommends starting in Log mode to measure latency and review decisions before you switch to Enforce.

Limitations

The following limitations apply:

Regions: Available only in the regions listed in Availability, based on the region of the workspace that sends the request.
Workspaces: Not available in workspaces that 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.
Function type: Only scalar Unity Catalog Python UDFs are supported. Batch Unity Catalog Python UDFs aren't supported.
Environment version: Only environment_version = '6' is supported.
Timeout: Each evaluation times out after 30 seconds, including the time to start the function's environment. You can't change this timeout.
Memory: A policy function can use up to 500 MiB of memory. If it uses more, the policy denies the interaction.
Dependencies: Dependencies from PyPI are supported. Dependencies from files in Unity Catalog volumes aren't supported.
Unity Catalog connections: Functions can't use Unity Catalog connections. Use a service credential to call cloud services.
Content types: The function receives text content only. Images, audio, and other multimodal content aren't evaluated.
Decisions: A Python policy returns ALLOW, DENY, or ASK. It can't redact or rewrite content.
Response evaluation: Databricks evaluates the complete response before it returns the response to the caller, so output isn't released until the check finishes.

Troubleshooting

Symptom	Likely cause
The policy fails with `CREDENTIALS_UNSUPPORTED_ENVIRONMENT`.	The function declares a service credential but doesn't set `environment_version = '6'`. Recreate the function with `ENVIRONMENT (environment_version = '6')`.
Interactions are denied with `Access denied: a policy on this service could not be evaluated.`	The function raised an unhandled exception, imported a package that isn't installed, or returned `NULL`, so the policy fails closed. Run the function in Test function with a sample event, or test it in a notebook, to see the error. Catch exceptions around external calls, and list each package that isn't in the environment in the function's dependencies.
Interactions are denied with `Access denied: the policy function did not return a value that could be interpreted as a decision.`	The return value doesn't follow the decision format: it isn't a `VARIANT` that holds a `decision` object, or its `result` isn't `ALLOW`, `DENY`, or `ASK`. Build the return value with `type(event).parseJson()`. See Return a decision.
Callers see the function's `reason` on requests it should allow.	The function returns a misspelled `result`, such as `ALOW`, together with a `reason`. When `result` isn't a valid decision but `reason` is a string, the policy denies the interaction and returns that reason. Check the spelling of `result`.
Calls to an external service fail.	The service credential isn't declared in the `CREDENTIALS` clause, the function doesn't use `getServiceCredentialsProvider()`, or the credential's role doesn't allow the action. Check the credential and its permissions.
In Enforce mode, every interaction is denied with `Python service policies are not available yet in this region.`	The workspace that sent the request isn't in a supported region. Call the service from a workspace in a supported region, or detach the Python policy from the service. In Log mode, the interaction proceeds and the would-be denial is recorded.
In Enforce mode, every interaction is denied with `Python service policies are not yet available in workspaces ...`.	The workspace that sent the request uses the compliance security profile, a serverless network policy that restricts internet access or blocks specific destinations, or a private network gateway. The reason names which one. After you change the workspace setting, the change can take up to 2 minutes to take effect. See Before you begin.
Interactions are denied with `Access denied: the policy function could not start in time. Try the request again.`	The function's environment wasn't ready in time. This is expected the first time the policy runs, and after you change its dependencies, while they install. Retry the request. See Performance.
Interactions are denied with `Access denied: the policy function did not complete within the 30s timeout.`	The function, including the time to start its environment, ran longer than 30 seconds. Set timeouts on external calls and keep the function fast.
Interactions are denied with `Access denied: the policy function ran out of memory.`	The function used more than 500 MiB of memory. Reduce the function's memory use, for example by loading less data or using smaller dependencies.

Symptom	Likely cause
The policy fails with `CREDENTIALS_UNSUPPORTED_ENVIRONMENT`.	The function declares a service credential but doesn't set `environment_version = '6'`. Recreate the function with `ENVIRONMENT (environment_version = '6')`.
Interactions are denied with `Access denied: a policy on this service could not be evaluated.`	The function raised an unhandled exception, imported a package that isn't installed, or returned `NULL`, so the policy fails closed. Run the function in Test function with a sample event, or test it in a notebook, to see the error. Catch exceptions around external calls, and list each package that isn't in the environment in the function's dependencies.
Interactions are denied with `Access denied: the policy function did not return a value that could be interpreted as a decision.`	The return value doesn't follow the decision format: it isn't a `VARIANT` that holds a `decision` object, or its `result` isn't `ALLOW`, `DENY`, or `ASK`. Build the return value with `type(event).parseJson()`. See Return a decision.
Callers see the function's `reason` on requests it should allow.	The function returns a misspelled `result`, such as `ALOW`, together with a `reason`. When `result` isn't a valid decision but `reason` is a string, the policy denies the interaction and returns that reason. Check the spelling of `result`.
Calls to an external service fail.	The service credential isn't declared in the `CREDENTIALS` clause, the function doesn't use `getServiceCredentialsProvider()`, or the credential's role doesn't allow the action. Check the credential and its permissions.
In Enforce mode, every interaction is denied with `Python service policies are not available yet in this region.`	The workspace that sent the request isn't in a supported region. Call the service from a workspace in a supported region, or detach the Python policy from the service. In Log mode, the interaction proceeds and the would-be denial is recorded.
In Enforce mode, every interaction is denied with `Python service policies are not yet available in workspaces ...`.	The workspace that sent the request uses the compliance security profile, a serverless network policy that restricts internet access or blocks specific destinations, or a private network gateway. The reason names which one. After you change the workspace setting, the change can take up to 2 minutes to take effect. See Before you begin.
Interactions are denied with `Access denied: the policy function could not start in time. Try the request again.`	The function's environment wasn't ready in time. This is expected the first time the policy runs, and after you change its dependencies, while they install. Retry the request. See Performance.
Interactions are denied with `Access denied: the policy function did not complete within the 30s timeout.`	The function, including the time to start its environment, ran longer than 30 seconds. Set timeouts on external calls and keep the function fast.
Interactions are denied with `Access denied: the policy function ran out of memory.`	The function used more than 500 MiB of memory. Reduce the function's memory use, for example by loading less data or using smaller dependencies.

Additional resources

Tutorial: Enforce Amazon Bedrock Guardrails with a custom Python policy

When to use a Python policy​

Availability​

How a Python policy works​

Before you begin​

Create a Python policy in the policy form​

Create a Python policy with SQL​

Write the function​

Read the event​

Return a decision​

Add Python packages​

Call an external service with a service credential​

Use a secret​

Handle errors​

Test a policy function in a notebook​

Performance​

Limitations​

Troubleshooting​

Additional resources​