Tutorial 28 — Building Custom Governance Integrations¶

┌──────────────────────────────────────────────────────────┐
│                    Your Application                      │
├────────────────────────┬─────────────────────────────────┤
│   Trust Integration    │       Kernel Adapter            │
│   (who can act?)       │       (what can they do?)       │
│                        │                                 │
│   AgentProfile         │       GovernancePolicy          │
│   ActionGuard          │       BaseIntegration           │
│   TrustTracker         │       ToolCallInterceptor       │
│                        │       pre_execute / post_execute│
├────────────────────────┴─────────────────────────────────┤
│                   Agent Framework                        │
│              (OpenAI, LangChain, CrewAI, …)              │
└──────────────────────────────────────────────────────────┘

myframework-agentmesh/
  myframework_agentmesh/
    __init__.py       # Public API
    trust.py          # Core implementation
  tests/
    test_trust.py     # Test suite
  pyproject.toml
  README.md

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

[project]
name = "myframework_agentmesh"
version = "0.1.0"
description = "AgentMesh trust layer for MyFramework"
readme = "README.md"
license = {text = "MIT"}
requires-python = ">=3.10"
dependencies = []              # No hard deps — framework needed at call time, not import time

[project.optional-dependencies]
dev = ["pytest>=7.0"]

[tool.hatch.build.targets.wheel]
packages = ["myframework_agentmesh"]

[tool.pytest.ini_options]
testpaths = ["tests"]

# myframework_agentmesh/trust.py
from __future__ import annotations

import time
from dataclasses import dataclass, field
from typing import Any


@dataclass
class AgentProfile:
    """Identity and trust state for a governed agent."""

    did: str                                # Decentralized identifier (did:agentmesh:...)
    name: str
    capabilities: list[str] = field(default_factory=list)
    trust_score: int = 500                  # 0-1000 scale
    status: str = "active"                  # active | suspended | revoked
    metadata: dict[str, Any] = field(default_factory=dict)

    def has_capability(self, capability: str) -> bool:
        return capability in self.capabilities

    def has_all_capabilities(self, required: list[str]) -> bool:
        return all(c in self.capabilities for c in required)

    def has_any_capability(self, required: list[str]) -> bool:
        return any(c in self.capabilities for c in required)

@dataclass
class ActionResult:
    """Outcome of a trust-gated action check."""

    allowed: bool
    agent_did: str
    action: str
    reason: str = ""
    trust_score: int = 0
    timestamp: float = field(default_factory=time.time)

    def to_dict(self) -> dict[str, Any]:
        return {
            "allowed": self.allowed,
            "agent_did": self.agent_did,
            "action": self.action,
            "reason": self.reason,
            "trust_score": self.trust_score,
            "timestamp": self.timestamp,
        }


class ActionGuard:
    """Trust-gated action enforcement.

    Checks agent trust score and capabilities before allowing an action.
    Supports per-action minimum trust thresholds for sensitive operations.
    """

    def __init__(
        self,
        min_trust_score: int = 500,
        sensitive_actions: dict[str, int] | None = None,
        blocked_actions: list[str] | None = None,
    ) -> None:
        self.min_trust_score = min_trust_score
        self.sensitive_actions: dict[str, int] = sensitive_actions or {}
        self.blocked_actions: list[str] = blocked_actions or []

    def check(
        self,
        agent: AgentProfile,
        action: str,
        required_capabilities: list[str] | None = None,
    ) -> ActionResult:
        """Evaluate whether an agent may perform an action."""
        # Hard block
        if action in self.blocked_actions:
            return ActionResult(
                allowed=False,
                agent_did=agent.did,
                action=action,
                reason=f"Action '{action}' is blocked by policy",
                trust_score=agent.trust_score,
            )

        # Status check
        if agent.status != "active":
            return ActionResult(
                allowed=False,
                agent_did=agent.did,
                action=action,
                reason=f"Agent status is '{agent.status}'",
                trust_score=agent.trust_score,
            )

        # Trust threshold (per-action or global)
        threshold = self.sensitive_actions.get(action, self.min_trust_score)
        if agent.trust_score < threshold:
            return ActionResult(
                allowed=False,
                agent_did=agent.did,
                action=action,
                reason=f"Trust score {agent.trust_score} below threshold {threshold}",
                trust_score=agent.trust_score,
            )

        # Capability check
        if required_capabilities and not agent.has_all_capabilities(required_capabilities):
            missing = [c for c in required_capabilities if not agent.has_capability(c)]
            return ActionResult(
                allowed=False,
                agent_did=agent.did,
                action=action,
                reason=f"Missing capabilities: {missing}",
                trust_score=agent.trust_score,
            )

        return ActionResult(
            allowed=True,
            agent_did=agent.did,
            action=action,
            trust_score=agent.trust_score,
        )

class TrustTracker:
    """Records agent outcomes and adjusts trust scores."""

    def __init__(self, reward: int = 10, penalty: int = 50) -> None:
        self.reward = reward
        self.penalty = penalty
        self._history: list[dict[str, Any]] = []

    def record_success(self, agent: AgentProfile, action: str) -> int:
        """Reward an agent for a successful action. Returns new score."""
        agent.trust_score = min(1000, agent.trust_score + self.reward)
        self._history.append({
            "did": agent.did, "action": action,
            "outcome": "success", "new_score": agent.trust_score,
            "timestamp": time.time(),
        })
        return agent.trust_score

    def record_failure(self, agent: AgentProfile, action: str) -> int:
        """Penalize an agent for a failed action. Returns new score."""
        agent.trust_score = max(0, agent.trust_score - self.penalty)
        self._history.append({
            "did": agent.did, "action": action,
            "outcome": "failure", "new_score": agent.trust_score,
            "timestamp": time.time(),
        })
        return agent.trust_score

    def get_history(self, did: str | None = None) -> list[dict[str, Any]]:
        if did:
            return [h for h in self._history if h["did"] == did]
        return list(self._history)

# myframework_agentmesh/__init__.py
"""AgentMesh trust layer for MyFramework."""

from myframework_agentmesh.trust import (
    ActionGuard,
    ActionResult,
    AgentProfile,
    TrustTracker,
)

__all__ = [
    "ActionGuard",
    "ActionResult",
    "AgentProfile",
    "TrustTracker",
]

# tests/test_trust.py
from myframework_agentmesh import AgentProfile, ActionGuard, TrustTracker, ActionResult


def test_allow_action_above_threshold():
    agent = AgentProfile(did="did:agentmesh:a1", name="Alpha", trust_score=700)
    guard = ActionGuard(min_trust_score=500)
    result = guard.check(agent, "search")
    assert result.allowed


def test_block_action_below_threshold():
    agent = AgentProfile(did="did:agentmesh:a1", name="Alpha", trust_score=300)
    guard = ActionGuard(min_trust_score=500)
    result = guard.check(agent, "search")
    assert not result.allowed
    assert "below threshold" in result.reason


def test_sensitive_action_requires_higher_trust():
    agent = AgentProfile(did="did:agentmesh:a1", name="Alpha", trust_score=600)
    guard = ActionGuard(
        min_trust_score=500,
        sensitive_actions={"delete_record": 800},
    )
    assert guard.check(agent, "search").allowed
    assert not guard.check(agent, "delete_record").allowed


def test_blocked_action_always_denied():
    agent = AgentProfile(did="did:agentmesh:a1", name="Alpha", trust_score=1000)
    guard = ActionGuard(blocked_actions=["drop_table"])
    result = guard.check(agent, "drop_table")
    assert not result.allowed
    assert "blocked by policy" in result.reason


def test_capability_check():
    agent = AgentProfile(
        did="did:agentmesh:a1", name="Alpha",
        capabilities=["read"], trust_score=700,
    )
    guard = ActionGuard(min_trust_score=500)
    assert guard.check(agent, "query", required_capabilities=["read"]).allowed
    assert not guard.check(agent, "query", required_capabilities=["write"]).allowed


def test_suspended_agent_blocked():
    agent = AgentProfile(did="did:agentmesh:a1", name="Alpha", trust_score=900, status="suspended")
    guard = ActionGuard(min_trust_score=500)
    result = guard.check(agent, "search")
    assert not result.allowed
    assert "suspended" in result.reason


def test_trust_tracker_adjusts_scores():
    agent = AgentProfile(did="did:agentmesh:a1", name="Alpha", trust_score=500)
    tracker = TrustTracker(reward=10, penalty=50)

    tracker.record_success(agent, "search")
    assert agent.trust_score == 510

    tracker.record_failure(agent, "search")
    assert agent.trust_score == 460

    history = tracker.get_history("did:agentmesh:a1")
    assert len(history) == 2
    assert history[0]["outcome"] == "success"
    assert history[1]["outcome"] == "failure"


def test_trust_score_clamped():
    agent = AgentProfile(did="did:agentmesh:a1", name="Alpha", trust_score=995)
    tracker = TrustTracker(reward=10, penalty=50)
    tracker.record_success(agent, "search")
    assert agent.trust_score == 1000  # Clamped at max

    agent.trust_score = 20
    tracker.record_failure(agent, "search")
    assert agent.trust_score == 0    # Clamped at min


def test_action_result_serialization():
    result = ActionResult(
        allowed=True, agent_did="did:agentmesh:a1",
        action="search", trust_score=700,
    )
    d = result.to_dict()
    assert d["allowed"] is True
    assert d["agent_did"] == "did:agentmesh:a1"
    assert "timestamp" in d

cd myframework-agentmesh
pip install -e ".[dev]"
pytest tests/ -x -q --tb=short

┌──────────────┐
│  wrap(agent) │
└──────┬───────┘
       │  creates ExecutionContext (deep-copies policy)
       ▼
┌──────────────────────────────────────────────────────┐
│                  Governed Proxy                       │
│                                                      │
│  run(prompt) ─────────────────────────────────────── │
│     │                                                │
│     ├─► pre_execute(ctx, input)                      │
│     │     enforces: tokens, timeout, blocked         │
│     │     patterns, human approval                   │
│     │     returns: (allowed: bool, reason: str|None) │
│     │                                                │
│     ├─► framework.run(prompt)                        │
│     │     the real LLM / agent call                  │
│     │                                                │
│     ├─► post_execute(ctx, output)                    │
│     │     runs: drift detection, checkpoints         │
│     │     emits: DRIFT_DETECTED event if threshold   │
│     │     exceeded (event-based, not blocking)       │
│     │     returns: (True, None) always               │
│     │                                                │
│     └─► emit(event_type, data)                       │
│           fires registered listeners                 │
│                                                      │
│  call_tool(name, args) ──────────────────────────── │
│     │                                                │
│     └─► PolicyInterceptor.intercept(request)         │
│           validates: allowed_tools, blocked_patterns │
│           validates: call count (if context passed)  │
│           returns: ToolCallResult(allowed, reason)   │
└──────────────────────────────────────────────────────┘

# myframework_kernel.py
from __future__ import annotations

from agent_os.integrations.base import (
    BaseIntegration,
    CompositeInterceptor,
    ExecutionContext,
    GovernanceEventType,
    GovernancePolicy,
    PolicyInterceptor,
    ToolCallRequest,
    ToolCallResult,
)
from agent_os.exceptions import PolicyViolationError


class MyFrameworkKernel(BaseIntegration):
    """Governance kernel for MyFramework."""

    def __init__(self, policy: GovernancePolicy | None = None) -> None:
        super().__init__(policy)

    def wrap(self, agent):
        """Wrap a MyFramework agent with governance. Returns a governed proxy."""
        agent_id = getattr(agent, "name", "unknown")
        ctx = self.create_context(agent_id=agent_id)

        self.emit(GovernanceEventType.POLICY_CHECK, {
            "agent_id": agent_id,
            "policy": self.policy.name,
            "action": "wrap",
        })

        return _GovernedAgent(agent, self, ctx)

    def unwrap(self, governed_agent):
        """Remove governance, return original agent."""
        return governed_agent._original


class _GovernedAgent:
    """Transparent governance proxy for a MyFramework agent."""

    def __init__(self, original, kernel: MyFrameworkKernel, ctx: ExecutionContext):
        self._original = original
        self._kernel = kernel
        self._ctx = ctx

    def run(self, prompt: str, **kwargs):
        """Execute with governance checks."""
        # Pre-execution: token limits, timeout, blocked patterns
        allowed, reason = self._kernel.pre_execute(self._ctx, prompt)
        if not allowed:
            self._kernel.emit(GovernanceEventType.POLICY_VIOLATION, {
                "agent_id": self._ctx.agent_id,
                "reason": reason,
                "phase": "pre_execute",
            })
            raise PolicyViolationError(reason)

        # Delegate to the real framework
        result = self._original.run(prompt, **kwargs)

        # Post-execution: drift detection (event-based, not blocking)
        self._kernel.post_execute(self._ctx, result)

        return result

    def call_tool(self, tool_name: str, arguments: dict):
        """Execute a tool call through the interception chain."""
        request = ToolCallRequest(
            tool_name=tool_name,
            arguments=arguments,
            agent_id=self._ctx.agent_id,
        )

        # Pass context so PolicyInterceptor can enforce max_tool_calls
        interceptor = PolicyInterceptor(self._kernel.policy, self._ctx)
        result: ToolCallResult = interceptor.intercept(request)

        if not result.allowed:
            self._kernel.emit(GovernanceEventType.TOOL_CALL_BLOCKED, {
                "agent_id": self._ctx.agent_id,
                "tool_name": tool_name,
                "reason": result.reason,
            })
            raise PolicyViolationError(f"Tool '{tool_name}' blocked: {result.reason}")

        final_args = result.modified_arguments or arguments
        return self._original.call_tool(tool_name, final_args)

    def get_context(self) -> ExecutionContext:
        return self._ctx

class RateLimitInterceptor:
    """Block tool calls when an agent exceeds its per-minute quota."""

    def __init__(self, max_per_minute: int = 30) -> None:
        self._max = max_per_minute
        self._counts: dict[str, list[float]] = {}

    def intercept(self, request: ToolCallRequest) -> ToolCallResult:
        import time
        now = time.time()
        calls = self._counts.setdefault(request.agent_id, [])
        calls[:] = [t for t in calls if now - t < 60]
        if len(calls) >= self._max:
            return ToolCallResult(allowed=False, reason="Rate limit exceeded")
        calls.append(now)
        return ToolCallResult(allowed=True)


# Wire into the governed proxy's call_tool method:
policy = GovernancePolicy(allowed_tools=["search", "summarize"], max_tool_calls=10)
composite = CompositeInterceptor([
    PolicyInterceptor(policy),  # pass ExecutionContext as 2nd arg to enforce max_tool_calls
    RateLimitInterceptor(max_per_minute=20),
])
result = composite.intercept(tool_request)  # all must allow

policy = GovernancePolicy(
    name="production",
    allowed_tools=["search", "summarize"],
    drift_threshold=0.15,
    log_all_calls=True,
)
kernel = MyFrameworkKernel(policy=policy)

# Alert on policy violations
kernel.on(GovernanceEventType.POLICY_VIOLATION, lambda data: (
    print(f"VIOLATION: {data['agent_id']} — {data['reason']}")
))

# Log blocked tool calls
kernel.on(GovernanceEventType.TOOL_CALL_BLOCKED, lambda data: (
    print(f"BLOCKED: {data['tool_name']} — {data['reason']}")
))

# Make drift detection blocking (optional)
def on_drift(data):
    raise PolicyViolationError(
        f"Drift {data['drift_score']:.2f} exceeds threshold"
    )

kernel.on(GovernanceEventType.DRIFT_DETECTED, on_drift)

governed = kernel.wrap(my_agent)

┌───────────────┐     ┌──────────────────┐     ┌──────────────┐
│ ActionGuard   │ ──► │ MyFrameworkKernel │ ──► │  Framework   │
│ (trust gate)  │     │ (policy gate)    │     │  (LLM call)  │
│               │     │                  │     │              │
│ who can act?  │     │ pre_execute      │     │              │
│ capabilities? │     │ tool intercept   │     │              │
│ trust score?  │     │ drift detection  │     │              │
└───────────────┘     └──────────────────┘     └──────────────┘

from __future__ import annotations

from myframework_agentmesh import AgentProfile, ActionGuard
from agent_os.integrations.base import (
    BaseIntegration,
    ExecutionContext,
    GovernanceEventType,
    GovernancePolicy,
    PolicyInterceptor,
    ToolCallRequest,
    ToolCallResult,
)
from agent_os.exceptions import PolicyViolationError


class TrustAwareKernel(BaseIntegration):
    """Kernel that checks trust before enforcing policy."""

    def __init__(
        self,
        policy: GovernancePolicy | None = None,
        guard: ActionGuard | None = None,
    ) -> None:
        super().__init__(policy)
        self.guard = guard or ActionGuard()

    def wrap(self, agent, profile: AgentProfile):
        """Wrap an agent with both trust verification and policy enforcement."""
        # agent_id must match ^[a-zA-Z0-9_-]+$ — DIDs contain colons,
        # so extract the final segment as the identifier.
        agent_id = profile.did.rsplit(":", 1)[-1]
        ctx = self.create_context(agent_id=agent_id)
        return _TrustGovernedAgent(agent, self, ctx, profile)

    def unwrap(self, governed_agent):
        return governed_agent._original


class _TrustGovernedAgent:
    def __init__(self, original, kernel: TrustAwareKernel,
                 ctx: ExecutionContext, profile: AgentProfile):
        self._original = original
        self._kernel = kernel
        self._ctx = ctx
        self._profile = profile

    def run(self, prompt: str, action: str = "execute", **kwargs):
        # Trust gate: is this agent allowed to perform this action?
        trust_result = self._kernel.guard.check(self._profile, action)
        if not trust_result.allowed:
            raise PolicyViolationError(
                f"Trust check failed for {self._profile.did}: {trust_result.reason}"
            )

        # Policy gate: does the input comply with governance rules?
        allowed, reason = self._kernel.pre_execute(self._ctx, prompt)
        if not allowed:
            raise PolicyViolationError(reason)

        result = self._original.run(prompt, **kwargs)
        self._kernel.post_execute(self._ctx, result)
        return result

guard = ActionGuard(
    min_trust_score=600,
    sensitive_actions={"deploy": 900, "delete": 800},
    blocked_actions=["drop_database"],
)

policy = GovernancePolicy(
    name="deploy-restricted",
    max_tool_calls=5,
    allowed_tools=["kubectl_apply", "helm_upgrade"],
    blocked_patterns=["--force", "delete namespace"],
    timeout_seconds=120,
    log_all_calls=True,
)

agent_profile = AgentProfile(
    did="did:agentmesh:deployer",
    name="Deployer",
    capabilities=["deploy", "rollback"],
    trust_score=750,
)

kernel = TrustAwareKernel(policy=policy, guard=guard)
governed = kernel.wrap(deploy_agent, profile=agent_profile)
governed.run("Deploy v2.1.0 to staging", action="deploy")