@governed decorator

The single-function entry point. Decorate any callable with @governed and the policy gate runs before the function body executes. If the call is denied, GovernanceError is raised and the function never runs.

from kitelogik import OPAClient, PolicyGate, SessionContext, governed

gate = PolicyGate(opa_client=OPAClient())
context = SessionContext(
    session_id="sess_001",
    user_role="support_agent",
    session_scopes=["read_customer", "approve_refund_under_100"],
)

@governed(gate=gate, context=context)
async def approve_refund(customer_id: str, amount: float) -> str:
    return f"Refunded ${amount:.2f} to {customer_id}"

What happens at call time

1. The decorator binds the call's positional + keyword args into a single dict using inspect.signature(fn) — that dict becomes input.args for OPA.
2. A ToolCallInput(action, tool_name, args) is built. By default action == tool_name == fn.__name__.
3. gate.evaluate_tool_call(tool_call, context) runs.
4. If the decision isn't allow=True, GovernanceError is raised with the full PolicyDecision on exc.decision.
5. On allow, the function body executes. String return values are run through the prompt-injection sanitiser by default.

Sync and async, both work

@governed(gate=gate, context=context)
def lookup(customer_id: str) -> str:        # sync
    return db.fetch(customer_id)

@governed(gate=gate, context=context)
async def search(query: str) -> str:        # async
    return await client.search(query)

The decorator inspects the function with inspect.iscoroutinefunction(fn) and produces the matching sync/async wrapper. Sync wrappers bridge to the async gate via _run_coroutine_sync, which works inside Jupyter and FastAPI loops too.

Constructor parameters

governed(
    gate: PolicyGate,
    context: SessionContext,
    action: str | None = None,    # OPA action name; defaults to fn.__name__
    sanitize: bool = True,         # scan str returns for injection
)

Param	Default	Purpose
action	fn.__name__	Override when the Python function name differs from the policy action name (e.g. partial_refund calls policies that match on approve_refund)
sanitize	True	Run the prompt-injection sanitiser on string return values. Disable only if you sanitise downstream.

Distinguishing failure modes

GovernanceError is raised on any non-allow decision — hard deny, soft deny, and requires_hitl=True. Inspect exc.decision to distinguish:

try:
    await approve_refund(customer_id="cust_001", amount=5000.0)
except GovernanceError as exc:
    if exc.decision.deny:
        ... # hard block — model cannot override; security-critical
    elif exc.decision.requires_hitl:
        ... # HITL escalation pending — see HITL queue patterns
    else:
        ... # soft deny — agent could adjust args and retry

Three distinct error messages are produced internally so the exception's str is also human-readable for logs:

Decision	Exception message
deny=True	Tool '<name>' hard blocked by security policy: <reason>
requires_hitl=True	Tool '<name>' requires human approval (risk tier: <tier>)
allow=False (other)	Tool '<name>' denied by policy: <reason>

When NOT to use @governed

• In a framework adapter agent loop. Adapters (OpenAI, LangChain, CrewAI, …) convert denials into structured tool results so the loop continues. @governed raises — which is what you want for plain Python callers, not for the agent loop.
• For tool collections that the agent dispatches by name. Reach for GovernedToolbox instead.

Comparison with GovernedToolbox

Need	Use
Wrap one specific function	@governed
Register many tools, call by name	GovernedToolbox
Wire into OpenAI / LangChain / etc.	Framework adapter (built on GovernedToolbox)

Source

kitelogik/governed.py on GitHub. Runnable example: examples/01_decorator.py.