govern() — Two-Line Governance Wrapper for Any Callable

govern() is the fastest path to governed agents. It wraps any Python callable — a tool function, an agent’s run() method, or a LangChain tool class — so that every invocation is checked against a policy, logged to the audit trail, and either permitted or blocked before a single line of the wrapped function executes.

from agentmesh.governance import govern

safe_tool = govern(my_tool, policy="policy.yaml")   # every call checked, logged, enforced

That is the entire integration. safe_tool behaves identically to my_tool — same signature, same return type — except that it raises GovernanceDenied if the policy denies the call.

Import

from agentmesh.governance import govern
from agentmesh.governance import GovernanceDenied, GovernanceConfig, GovernedCallable

Signature

def govern(
    fn: Callable,
    *,
    policy: str | Policy,
    agent_id: str = "*",
    audit: bool = True,
    on_deny: Callable[[PolicyDecision], Any] | None = None,
    approval_handler: ApprovalHandler | None = None,
    advisory: AdvisoryCheck | None = None,
    conflict_strategy: str = "deny_overrides",
    ring: ExecutionRing | None = None,
    session_id: str = "",
    approval_coordinator: ApprovalCoordinator | None = None,
    approval_chain_id: str | None = None,
    approval_ttl_seconds: float = 300.0,
) -> GovernedCallable

Parameters

Callable

required

The function, tool, or agent callable to govern. The returned GovernedCallable preserves the original function’s __name__, __doc__, and signature via functools.update_wrapper.

policy

str | Policy

required

The policy to enforce. Accepts:

A file path to a YAML policy file (supports extends for inheritance)
An inline YAML string
A Policy object constructed programmatically

# File path
safe_tool = govern(my_tool, policy="policies/production.yaml")

# Inline YAML
safe_tool = govern(my_tool, policy="""
name: inline-policy
rules:
  - name: block-writes
    condition: {field: action.type, operator: eq, value: write}
    action: deny
""")

agent_id

str

default:"*"

Agent identifier passed to the policy engine during evaluation. Use a specific agent ID when policies contain agent-scoped rules. Defaults to "*" (wildcard, matches all agent patterns).

audit

bool

default:"True"

When True, every call to the governed function is logged to an in-memory AuditLog. The log is accessible via safe_tool.audit_log. Set to False in unit tests where you don’t need audit overhead.

on_deny

Callable[[PolicyDecision], Any] | None

default:"None"

Optional callback invoked when the policy denies a call. Receives the PolicyDecision. When None (default), a GovernanceDenied exception is raised. Use a callback to return a graceful fallback instead:

def fallback(decision):
    return {"error": decision.reason, "allowed": False}

safe_tool = govern(my_tool, policy="policy.yaml", on_deny=fallback)

approval_handler

ApprovalHandler | None

default:"None"

Optional human-approval handler for require_approval policy actions. Built-in handlers include ConsoleApproval (stdin prompt), WebhookApproval (HTTP callback), and CallbackApproval (custom function). Defaults to AutoRejectApproval (auto-rejects unapproved calls fail-closed).

advisory

AdvisoryCheck | None

default:"None"

Optional advisory layer (e.g., a semantic classifier or LLM-based check). Advisory checks run only after the deterministic policy permits a call. They can escalate to a block, but never override a deny. Fails open — an advisory error does not block the call.

conflict_strategy

str

default:"deny_overrides"

Policy conflict resolution strategy. Options: "deny_overrides" (any deny wins — recommended for enterprise), "allow_overrides", "priority_first_match", "most_specific_wins". See Conflict Resolution for details.

ring

ExecutionRing | None

default:"None"

Optional privilege ring assignment for ring-level resource constraint enforcement. When set, ring enforcement runs before policy evaluation. Shared ring breach detectors accumulate violations across all governed callables for the same (agent_id, session_id) pair.

session_id

str

default:""

Session identifier used by the ring breach detector to count violations per session. Defaults to an empty string (treated as "default").

approval_coordinator

ApprovalCoordinator | None

default:"None"

When combined with approval_chain_id, routes require_approval decisions through the action-bound approval protocol (ADR-0030). Provides cryptographic binding, audit linkage, and fail-closed timeout semantics.

approval_chain_id

str | None

default:"None"

Approval chain ID for the action-bound protocol. Required when approval_coordinator is set.

approval_ttl_seconds

float

default:"300.0"

Time-to-live for approval requests in seconds. Requests that expire before approval is received are automatically rejected (fail-closed).

Returns

GovernedCallable

A callable wrapper that behaves identically to fn but with governance enforcement on every call. Exposes two properties:

Show GovernedCallable properties

engine

PolicyEngine

The underlying PolicyEngine instance. Access for advanced programmatic policy manipulation.

audit_log

AuditLog | None

The in-memory AuditLog used to record decisions. None when audit=False.

GovernanceDenied Exception

GovernanceDenied is raised when the policy engine denies an action and no on_deny callback is registered.

from agentmesh.governance import GovernanceDenied

try:
    result = safe_tool(action="drop", table="users")
except GovernanceDenied as e:
    print(e)              # "Action denied by policy rule 'block-destructive': ..."
    print(e.decision)     # PolicyDecision object
    print(e.decision.matched_rule)   # "block-destructive"
    print(e.decision.reason)         # "Destructive operations require human approval"
    print(e.decision.action)         # "deny"
    print(e.decision.allowed)        # False

decision

PolicyDecision

The full PolicyDecision that triggered the denial. Contains matched_rule, reason, action, allowed, and audit_entry.

The string representation of the exception follows: "Action denied by policy rule '{rule_name}': {reason}".

How Context Is Built

govern() builds the policy evaluation context automatically from the wrapped function’s keyword arguments:

If a kwarg named action is passed, it becomes context["action"]. If it’s a dict, it’s used as-is; otherwise it’s wrapped as {"type": str(value)}.
All other kwargs are added directly to the context dict. Dicts are passed through; scalars are wrapped as {"value": scalar}.
Positional arguments are mapped to context["action"]["type"] when no action kwarg is present.

# Call:
safe_tool(action="delete", table="users", requester="alice")

# Resulting context sent to policy engine:
{
    "action": {"type": "delete"},
    "table": {"value": "users"},
    "requester": {"value": "alice"},
}

# Policy rule that would match:
# condition:
#   field: action.type     ← access nested fields with dot notation
#   operator: eq
#   value: delete

Complete Code Examples

Quickstart — two lines

from agentmesh.governance import govern

def my_tool(action: str, table: str) -> dict:
    return {"table": table, "rows": 42}

safe_tool = govern(my_tool, policy="policy.yaml")

# Allowed call
result = safe_tool(action="read", table="users")
# {"table": "users", "rows": 42}

# Denied call
safe_tool(action="drop", table="users")
# GovernanceDenied: Action denied by policy rule 'block-destructive':
#   Destructive operations require human approval

# policy.yaml
apiVersion: governance.toolkit/v1
name: production-policy
default_action: allow
rules:
  - name: block-destructive
    condition: "action.type in ['drop', 'delete', 'truncate']"
    action: deny
    description: "Destructive operations require human approval"

Graceful degradation with on_deny

from agentmesh.governance import govern, GovernanceDenied

def fallback_response(decision):
    return {
        "error": "governance_denied",
        "reason": decision.reason,
        "rule": decision.matched_rule,
    }

safe_tool = govern(
    my_tool,
    policy="policy.yaml",
    on_deny=fallback_response,
)

# Returns the fallback dict instead of raising
result = safe_tool(action="drop", table="users")
# {"error": "governance_denied", "reason": "...", "rule": "block-destructive"}

Inspect the audit log

from agentmesh.governance import govern

safe_tool = govern(my_tool, policy="policy.yaml", agent_id="agent-001")

safe_tool(action="read", table="users")
safe_tool(action="read", table="products")

# Inspect decisions
for entry in safe_tool.audit_log._chain._entries:
    print(entry.event_type, entry.outcome, entry.data)

Human approval for sensitive operations

from agentmesh.governance import govern
from agentmesh.governance import ConsoleApproval

safe_tool = govern(
    send_email_tool,
    policy="email-policy.yaml",
    agent_id="email-agent",
    approval_handler=ConsoleApproval(),
)

# When policy returns require_approval, operator is prompted at the console
safe_tool(action="send_email", to="cto@company.com", subject="Budget Report")
# [Governance] Approve? send_email → cto@company.com [y/N]: y

Async Tools

govern() wraps synchronous callables. For async tools, use asyncio.run inside the wrapped function or use the async-compatible approach:

import asyncio
from agentmesh.governance import govern

def async_safe_wrapper(**kwargs):
    """Synchronous shim for governed async tool."""
    return asyncio.run(my_async_tool(**kwargs))

safe_tool = govern(async_safe_wrapper, policy="policy.yaml")

LangChain Tool Integration

Wrap a LangChain tool by governing its _run method:

from langchain.tools import BaseTool
from agentmesh.governance import govern

class DatabaseTool(BaseTool):
    name = "database"
    description = "Query the company database"

    def _run(self, query: str) -> str:
        return execute_query(query)

tool = DatabaseTool()

# Govern the underlying callable
governed_run = govern(
    tool._run,
    policy="db-policy.yaml",
    agent_id="langchain-agent",
)

# Monkey-patch for transparent integration
tool._run = governed_run

Or govern at the agent level by wrapping the agent’s invoke:

from langchain.agents import AgentExecutor
from agentmesh.governance import govern

agent_executor = AgentExecutor(agent=agent, tools=tools)

safe_invoke = govern(
    agent_executor.invoke,
    policy="agent-policy.yaml",
    agent_id="langchain-agent-001",
)

result = safe_invoke({"input": "What is the current inventory?"})

GovernanceConfig Dataclass

For multi-step configuration, build a GovernanceConfig and pass it directly to GovernedCallable:

from agentmesh.governance import GovernanceConfig, GovernedCallable

config = GovernanceConfig(
    policy="policies/production.yaml",
    agent_id="research-agent-42",
    audit=True,
    audit_file="logs/audit.jsonl",   # persist audit log to disk
    conflict_strategy="deny_overrides",
    approval_ttl_seconds=120.0,
)

safe_tool = GovernedCallable(my_tool, config)

policy

str | Policy

required

Policy file path, inline YAML string, or Policy object.

agent_id

str

default:"*"

Agent identifier for policy evaluation.

audit

bool

default:"True"

Enable audit logging.

audit_file

str | None

default:"None"

Path for a file-based audit log. None = in-memory only.

on_deny

Callable | None

default:"None"

Callback on denial; default raises GovernanceDenied.

approval_handler

ApprovalHandler | None

default:"None"

Optional human-approval handler for require_approval policy actions.

advisory

AdvisoryCheck | None

default:"None"

Optional advisory layer. Runs only after the deterministic policy permits a call. Fails open — an advisory error does not block the call.

conflict_strategy

str

default:"deny_overrides"

Policy conflict resolution strategy.

approval_ttl_seconds

float

default:"300.0"

Time-to-live for approval requests.

Python SDK

Other SDKs

Specifications

govern() — Two-Line Governance Wrapper for Any Callable

Import

Signature

Parameters

Returns

GovernanceDenied Exception

How Context Is Built

Complete Code Examples

Quickstart — two lines

Graceful degradation with on_deny

Inspect the audit log

Human approval for sensitive operations

Async Tools

LangChain Tool Integration

GovernanceConfig Dataclass

See Also

Build docs developers (and LLMs) love

Python SDK

Other SDKs

Specifications

Documentation Index

​Import

​Signature

​Parameters

​Returns

​GovernanceDenied Exception

​How Context Is Built

​Complete Code Examples

​Quickstart — two lines

​Graceful degradation with on_deny

​Inspect the audit log

​Human approval for sensitive operations

​Async Tools

​LangChain Tool Integration

​GovernanceConfig Dataclass

​See Also

Build docs developers (and LLMs) love

Import

Signature

Parameters

Returns

GovernanceDenied Exception

How Context Is Built

Complete Code Examples

Quickstart — two lines

Graceful degradation with on_deny

Inspect the audit log

Human approval for sensitive operations

Async Tools

LangChain Tool Integration

GovernanceConfig Dataclass

See Also