Custom Policies

When the builtin policies don't cover your use case, you have two options:

1. cel_policy — write a CEL expression directly in your config. No code to deploy. Start here.
2. Python policy — a full Python function registered on the server. Use this only when your logic requires state, external calls, or anything else CEL's non-Turing-complete expressions cannot express.

Using cel_policy

cel_policy is a builtin policy that evaluates a CEL expression against every event. CEL is a safe, sandboxed expression language — no loops, no I/O, no side effects — which makes it the right default for most custom rules.

The expression receives an event object with type and data fields (the same PolicyEvent every policy sees). Return true to deny the action.

Examples

Deny a specific tool:

policies:
  no_dangerous_tool:
    type: function
    handler: omnigent.policies.builtins.safety.cel_policy
    factory_params:
      expression: "event.type == 'tool_call' && event.data.name == 'dangerous_tool'"
      reason: "dangerous_tool is not allowed in this session."

Block writes outside a specific directory:

policies:
  confine_writes:
    type: function
    handler: omnigent.policies.builtins.safety.cel_policy
    factory_params:
      expression: >
        event.type == 'tool_call' &&
        event.data.name in ['sys_os_write', 'sys_os_edit'] &&
        !event.data.arguments.path.startsWith('/home/user/project/')
      reason: "Writes are only allowed inside /home/user/project/."

Track calls with state_updates:

Return a state_updates list to accumulate counters or flags in session state. The engine applies the mutations after the policy fires, so later events can read the updated values via event.session_state.

policies:
  track_tool_calls:
    type: function
    handler: omnigent.policies.builtins.safety.cel_policy
    factory_params:
      expression: >
        event.type == 'tool_call'
          ? {"result": "ALLOW",
             "state_updates": [
               {"key": "tool_count", "action": "increment", "value": 1},
               {"key": "last_tool",  "action": "set",       "value": event.data.name}
             ]}
          : {"result": "ALLOW"}

State persists across the whole session. A companion policy can read event.session_state.tool_count to gate on the accumulated total.

Use cel_policy whenever your rule can be expressed as a boolean over the event fields. You get the same policy lifecycle (ALLOW / ASK / DENY) without writing or deploying any code.

Python policies

Use a Python policy when cel_policy isn't enough — for example, when your rule needs accumulated state across calls, a database lookup, or control flow that CEL cannot express.

1. Write a policy function

A policy is a Python function that receives an event and returns ALLOW, ASK, DENY, or None (no opinion):

from omnigent.policies.schema import PolicyEvent, PolicyResponse
 
def my_policy(event: PolicyEvent) -> PolicyResponse | None:
    if event["type"] != "tool_call":
        return None
    if event["data"]["name"] == "dangerous_tool":
        return {"result": "DENY", "reason": "Blocked."}
    return {"result": "ALLOW"}

If your policy needs parameters, use the factory pattern. The function takes config and returns the evaluator:

def block_domains(blocked_domains: list[str]) -> callable:
    blocked = frozenset(d.lower() for d in blocked_domains)
 
    def evaluate(event: PolicyEvent) -> PolicyResponse | None:
        if event["type"] != "tool_call":
            return None
        url = event["data"]["arguments"].get("url", "")
        for domain in blocked:
            if domain in url.lower():
                return {"result": "DENY", "reason": f"Domain {domain} blocked."}
        return {"result": "ALLOW"}
 
    return evaluate

2. Register on the server

To make your policy discoverable by your Omnigent and visible in the UI, do two things:

Export a POLICY_REGISTRY from your module:

# myorg/policies.py
 
POLICY_REGISTRY = [
    {
        "handler": "myorg.policies.block_domains",
        "kind": "factory",
        "name": "Block Domains",
        "description": "Block web access to specific domains.",
        "params_schema": {
            "type": "object",
            "properties": {
                "blocked_domains": {
                    "type": "array",
                    "items": {"type": "string"},
                    "description": "Domains to block"
                }
            },
            "required": ["blocked_domains"]
        }
    }
]

Add the module to your server config:

# config.yaml
policy_modules:
  - myorg.policies

Then start the server with that config:

omnigent server -c config.yaml

The -c flag is short for --config.

Once registered, your custom policies appear alongside the builtins. Your Omnigent can select them when you ask it to add a policy in chat, and they show up in the UI settings panel.

3. Use it

Once registered, your custom policy works the same as any builtin. See Adding a policy for all the ways to apply it (chat, Omnigent YAML, or server config).

Reference: PolicyEvent

Every policy receives a PolicyEvent dict. Use the type field to filter which events you care about:

Return None for event types you don't handle.

Every event also carries:

• event["target"] — tool name on tool_call / tool_result; None otherwise.
• event["session_state"] — read-only dict of per-session key/value pairs accumulated by earlier policies. Mutate it via state_updates in the return value, not by modifying this dict.
• event["context"] — metadata about the session (see below).

EventContext

event["context"] is a dict with the following fields:

Reference: PolicyResponse

A policy callable returns None (abstain) or a dict with the following fields:

Returning None is treated as abstain (equivalent to {"result": "ALLOW"}).

# minimal
{"result": "ALLOW"}
 
# full
{
    "result": "DENY",
    "reason": "Blocked.",
    "data": <transformed-content>,
    "state_updates": [
        {"key": "call_count", "action": "increment", "value": 1},
    ],
    "set_labels": {"integrity": "0"},
}

Reference: StateUpdateEntry

Each entry in state_updates is a dict:

The engine applies updates to event["session_state"] so subsequent policy evaluations see the new values.

# overwrite
{"key": "status",     "action": "set",       "value": "escalated"}
 
# numeric delta (initializes to 0 if absent)
{"key": "call_count", "action": "increment",  "value": 1}
 
# append to a list (initializes to [] if absent)
{"key": "seen_tools", "action": "append",     "value": "sys_os_shell"}
 
# remove the key entirely
{"key": "temp_flag",  "action": "delete"}