Governance with the Agent Governance Toolkit

The reference architecture's Chapter 10 on Governance asks four questions a model can't answer for you:

1. What data is each agent allowed to access?
2. Who is accountable for what an agent produces?
3. What's the scope of agent action?
4. How is the model's lifecycle managed?

MAF doesn't ship a governance primitive — by design. Governance is policy, and policy is not a model concern. You have to enforce it in application code. The question is which application-code primitive to reach for.

For this project I reached for the Agent Governance Toolkit. It's an OSS toolkit aligned to the OWASP Agentic Top 10, with YAML policies, a one-line wrapper, and a programmatic evaluator. The framing line from its README is honest:

Prompt-level safety ("please follow the rules") is not a control surface. It is a polite request to a stochastic system.

That's the entire premise. Every tool call, message send, and delegation is intercepted in deterministic application code before the model's intent reaches the wire.

Two API surfaces

AGT exposes two paths into the same policy engine. You use both.

Surface 1 — govern_tool(fn, policy=...) — one-line wrapper

The project's governance/agt.py wraps it:

def govern_tool(fn, *, policy=None, agent_id="*", audit=True):
    from agentmesh.governance import govern
    policy_path = str(Path(policy) if policy else DEFAULT_POLICY_PATH)
    return govern(fn, policy=policy_path, agent_id=agent_id, audit=audit)

You hand it a callable and a YAML path; it gives you back a callable that checks policy on every invocation. The first time a denied call happens, you get a GovernanceDenied. The audit log is written automatically.

Surface 2 — PolicyEvaluator — programmatic evaluator

When you want to run a policy check without wrapping a function — for instance to gate orchestrator decisions before they happen — you use the evaluator directly:

def build_policy_evaluator(rules, *, policy_dir=None):
    from agent_os.policies import (
        PolicyAction, PolicyCondition, PolicyDefaults,
        PolicyDocument, PolicyEvaluator, PolicyOperator, PolicyRule,
    )
    # rules is a list of dicts in AGT's rule shape
    rule_models = [
        PolicyRule(
            name=r["name"],
            condition=PolicyCondition(
                field=r["condition"]["field"],
                operator=PolicyOperator(r["condition"]["operator"]),
                value=r["condition"]["value"],
            ),
            action=PolicyAction(r["action"]),
            priority=r.get("priority", 100),
        )
        for r in rules or []
    ]
    doc = PolicyDocument(
        name="programmatic-policy", version="1.0",
        defaults=PolicyDefaults(action=PolicyAction.ALLOW),
        rules=rule_models,
    )
    return PolicyEvaluator(policies=[doc])

A live check:

ev = build_policy_evaluator(rules=[
    {"name": "block-shell", "condition":
        {"field": "tool_name", "operator": "in",
         "value": ["shell_exec", "subprocess"]},
     "action": "deny", "priority": 100},
])

decision = ev.evaluate({"tool_name": "web_search"})
# decision.allowed = True

decision = ev.evaluate({"tool_name": "shell_exec"})
# decision.allowed = False
# decision.matched_rule = "block-shell"
# decision.reason = "Matched rule 'block-shell'"

The decision carries allowed, matched_rule, reason, and an audit_entry with a timestamp and the context snapshot. That's the input both for "do I let this call happen?" and for "what just happened, audit-trail-wise?"

The YAML

config/policy.yaml is the project's shipped policy:

apiVersion: governance.toolkit/v1
name: multi-agent-maf-policy
version: "1.0"
default_action: allow

rules:
  - name: block-destructive-actions
    condition: "action.type in ['drop', 'delete', 'truncate', 'rm', 'shutdown']"
    action: deny
    priority: 100

  - name: require-approval-for-send
    condition: "tool_name in ['send_email', 'post_message', 'http_post']"
    action: require_approval
    approvers: ["security-team"]
    priority: 90

  - name: log-web-search
    condition: "tool_name == 'web_search'"
    action: log
    priority: 50

Three rules covering destructive-action blocking, outbound-communication approval gating, and external-lookup logging. The action values AGT accepts in this version are allow | deny | warn | require_approval | log — audit is not one of them (a mistake I made in my first version, easy to spot from the Pydantic validation error).

Wiring into the agents

The project's Researcher is the only agent with tools, so it's the natural place to apply governance. Rather than wrapping in the agent file, the project has a separate tools/governed.py module that exports governed versions of every tool, with graceful fallback if AGT or the policy isn't available:

from agent_framework import tool

from multi_agent.tools.calculator import calculator as _calc_tool
from multi_agent.tools.search import web_search as _search_tool


def _build_governed():
    if not _POLICY_PATH.exists():
        log.info("config/policy.yaml not found — falling back to ungoverned tools.")
        return _calc_tool, _search_tool
    try:
        from multi_agent.governance import make_governed_tool
    except ImportError:
        log.warning("agent-governance-toolkit not installed — falling back.")
        return _calc_tool, _search_tool

    raw_calc, raw_search = _calc_tool.func, _search_tool.func
    gc = make_governed_tool(raw_calc,  policy=_POLICY_PATH, tool_name="calculator")
    gs = make_governed_tool(raw_search, policy=_POLICY_PATH, tool_name="web_search")
    return tool(approval_mode="never_require")(gc), tool(approval_mode="never_require")(gs)


governed_calculator, governed_web_search = _build_governed()

That's the entire researcher-side change. The agent's tools=[...] list points at governed_calculator, governed_web_search instead of the raw versions. AGT does the rest, every call.

Bridging into OpenTelemetry

make_governed_tool is govern_tool plus a metrics bridge:

def make_governed_tool(fn, *, policy=None, tool_name=None, agent_id="*"):
    import functools
    from agentmesh.governance import GovernanceDenied
    from multi_agent.observability import get_workflow_metrics

    governed = govern_tool(fn, policy=policy, agent_id=agent_id)
    label_tool = tool_name or getattr(fn, "__name__", "tool")
    metrics = get_workflow_metrics()

    @functools.wraps(fn)
    def wrapper(*args, **kwargs):
        try:
            result = governed(*args, **kwargs)
        except GovernanceDenied as exc:
            metrics.policy_decisions.add(
                1, {"tool": label_tool, "action": "deny",
                    "rule": getattr(exc, "rule_name", None) or "unknown"},
            )
            raise
        metrics.policy_decisions.add(
            1, {"tool": label_tool, "action": "allow", "rule": "none"},
        )
        return result

    return wrapper

Now every allow and every deny shows up in Prometheus as multi_agent_policy_decisions_total{tool, action, rule}. The Grafana dashboard ships a panel for it:

=== Live Prometheus output ===
multi_agent_policy_decisions_total{action='allow', rule='none', tool='calculator'} = 3
multi_agent_policy_decisions_total{action='allow', rule='none', tool='web_search'} = 2

The deny case is the high-signal one — a non-zero deny rate is an investigation. The Grafana dashboard's "Policy deny rate (15m)" stat goes orange at 1% and red at 10%, which are the thresholds I'd want to see at a glance on a deploy.

The flow, in a diagram

sequenceDiagram
    autonumber
    actor User
    participant Agent
    participant Govern as govern_tool wrapper
    participant Policy as PolicyEvaluator
    participant Tool as Underlying @tool
    participant OTel as OTel counter

    User->>Agent: prompt
    Agent->>Govern: invoke wrapped tool(args)
    Govern->>Policy: evaluate({tool_name, args, agent_id})
    alt allowed
        Policy-->>Govern: allowed
        Govern->>Tool: forward call
        Tool-->>Govern: result
        Govern->>OTel: action=allow
        Govern-->>Agent: result
    else denied
        Policy-->>Govern: denied
        Govern->>OTel: action=deny, rule
        Govern-->>Agent: GovernanceDenied
    end

The dotted line at the bottom — the OTel counter — is the bit that turns a one-off blocked call into a deployable signal. If your deny rate climbs after a model change, you want to know in your dashboard rather than in a customer ticket.

What this isn't

AGT is not a substitute for content filtering, mTLS, identity, RBAC, or model-side guardrails. It's a policy enforcement layer that operates between the model's decision and the tool's execution. The OWASP Agentic Top 10 has ten threats; AGT addresses a specific subset (excessive agency, tool misuse, data exfiltration via tool call) extremely well, and others (prompt injection, model poisoning) not at all.

The right framing is: AGT is the place to put the rules you want to enforce deterministically. Don't put the rules you want to suggest to the model there. Don't put a "please be kind" instruction in config/policy.yaml. The model never sees it.

Tests offline

The deterministic surface — PolicyEvaluator — is testable without API keys:

def test_policy_evaluator_deny_rule():
    ev = build_policy_evaluator(rules=[
        {"name": "block-shell", "condition":
            {"field": "tool_name", "operator": "in",
             "value": ["shell_exec", "subprocess"]},
         "action": "deny", "priority": 100},
    ])
    assert ev.evaluate({"tool_name": "web_search"}).allowed
    assert not ev.evaluate({"tool_name": "shell_exec"}).allowed


def test_policy_evaluator_matches_operator():
    ev = build_policy_evaluator(rules=[
        {"name": "block-sql-tools", "condition":
            {"field": "tool_name", "operator": "matches", "value": r".*sql.*"},
         "action": "deny", "priority": 90},
    ])
    assert ev.evaluate({"tool_name": "search"}).allowed
    assert not ev.evaluate({"tool_name": "run_sql"}).allowed

Four AGT tests run in 0.5 seconds in CI. The metrics bridge has its own test using OpenTelemetry's InMemoryMetricReader, in the same suite.

Sequence summary

You have three things to do to ship AGT alongside MAF:

1. Drop a config/policy.yaml next to your agent code. Three rules is a perfectly defensible starting point.
2. Wrap every tool that touches the outside world with make_governed_tool. Don't wrap deterministic helpers — they're not the threat model.
3. Add the policy_decisions panel to your Grafana dashboard so deny-rate spikes are visible at a glance.

The whole AGT integration in this project is one Python file (governance/agt.py), one YAML file (config/policy.yaml), one wrapper file (tools/governed.py), and 70 lines of test. The value is governance that can't be bypassed by the model. That's the whole point.