{"id":1016,"date":"2026-06-01T04:07:01","date_gmt":"2026-05-31T20:07:01","guid":{"rendered":"https:\/\/connectword.dpdns.org\/?p=1016"},"modified":"2026-06-01T04:07:01","modified_gmt":"2026-05-31T20:07:01","slug":"an-implementation-of-the-microsoft-agent-governance-toolkit-for-safe-ai-agent-tool-use-with-policies-approvals-audit-logs-and-risk-controls","status":"publish","type":"post","link":"https:\/\/connectword.dpdns.org\/?p=1016","title":{"rendered":"An Implementation of the Microsoft Agent Governance Toolkit for Safe AI Agent Tool Use with Policies, Approvals, Audit Logs, and Risk Controls"},"content":{"rendered":"

In this tutorial, we build a governed AI-agent workflow using Microsoft\u2019s Agent Governance Toolkit<\/strong><\/a> as the reference point. We create a Colab-ready implementation where agents do not directly execute tools; instead, every action first passes through a governance layer that checks the agent\u2019s identity, trust score, risk tier, requested tool, action type, sensitivity level, and policy rules. We define a YAML-based policy that controls destructive database operations, external email sending, shell execution, access to sensitive data, and financial transfers. We then wrap each tool with governance logic so that actions can be allowed, denied, sandboxed, or routed through an approval step before execution. We also generate tamper-evident audit records, run policy tests, activate a kill switch, summarize governance decisions, and visualize the relationships between agents, tools, rules, and outcomes as a graph.<\/p>\n

\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
import os\nimport sys\nimport json\nimport time\nimport uuid\nimport hmac\nimport yaml\nimport hashlib\nimport random\nimport shutil\nimport subprocess\nfrom dataclasses import dataclass, asdict\nfrom datetime import datetime, timezone\nfrom typing import Any, Dict, List, Callable, Optional\ndef pip_install(*packages):\n   subprocess.run(\n       [sys.executable, \"-m\", \"pip\", \"install\", \"-q\", *packages],\n       check=False\n   )\npip_install(\"pyyaml\", \"pandas\", \"networkx\", \"matplotlib\", \"rich\")\npip_install(\"agent-governance-toolkit[full]\")\nfrom rich.console import Console\nfrom rich.table import Table\nfrom rich.panel import Panel\nfrom rich import box\nimport pandas as pd\nimport networkx as nx\nimport matplotlib.pyplot as plt\nconsole = Console()\nREPO_URL = \"https:\/\/github.com\/microsoft\/agent-governance-toolkit\"\nREPO_DIR = \"\/content\/agent-governance-toolkit\"\nif not os.path.exists(REPO_DIR):\n   subprocess.run([\"git\", \"clone\", \"--depth\", \"1\", REPO_URL, REPO_DIR], check=False)\nofficial_govern = None\nofficial_import_error = None\ntry:\n   from agentmesh.governance import govern as official_govern\nexcept Exception as e:\n   official_import_error = repr(e)<\/code><\/pre>\n<\/div>\n<\/div>\n
We set up the Colab environment by installing the required libraries and importing everything needed for policy handling, auditing, visualization, and data analysis. We also clone the Microsoft Agent Governance Toolkit repository to keep the notebook connected to the original project. We then try to import the official governance function, while keeping the tutorial runnable even if the preview package changes.<\/p>\n
\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
POLICY_PATH = \"\/content\/advanced_agent_policy.yaml\"\npolicy_yaml = \"\"\"\napiVersion: governance.toolkit\/v1\nname: advanced-colab-governance-policy\ndefault_action: allow\nmetadata:\n owner: ai-platform-team\n environment: tutorial\n description: >\n   Demonstrates deterministic governance controls for AI agent tool calls.\nrules:\n - name: block-destructive-database-actions\n   description: \"Agents must not perform destructive database operations.\"\n   condition: \"action.type in ['drop_table', 'delete_table', 'truncate_table']\"\n   action: deny\n   severity: critical\n   owasp_risk: \"Tool misuse \/ Excessive agency\"\n - name: require-human-approval-for-email\n   description: \"External email requires approval before execution.\"\n   condition: \"action.type == 'send_email' and action.recipient_domain != 'internal.local'\"\n   action: require_approval\n   approvers: [\"security-team\", \"business-owner\"]\n   severity: high\n   owasp_risk: \"Goal hijacking \/ Unauthorized action\"\n - name: sandbox-shell-execution\n   description: \"Shell commands must run in a sandbox with blocked dangerous commands.\"\n   condition: \"action.type == 'shell_exec'\"\n   action: sandbox\n   sandbox:\n     blocked_terms: [\"rm -rf\", \"curl http\", \"wget http\", \"chmod 777\", \"sudo\"]\n     max_runtime_seconds: 2\n   severity: high\n   owasp_risk: \"Tool misuse \/ Unsafe execution\"\n - name: deny-low-trust-agent-sensitive-data\n   description: \"Low-trust agents cannot access sensitive data.\"\n   condition: \"identity.trust_score < 0.65 and action.sensitivity in ['high', 'critical']\"\n   action: deny\n   severity: high\n   owasp_risk: \"Identity abuse \/ Data exfiltration\"\n - name: require-approval-for-financial-transaction\n   description: \"Financial transactions above 1000 require approval.\"\n   condition: \"action.type == 'transfer_money' and action.amount > 1000\"\n   action: require_approval\n   approvers: [\"finance-controller\"]\n   severity: critical\n   owasp_risk: \"Excessive agency \/ Business process compromise\"\n - name: rate-limit-high-risk-agent\n   description: \"High-risk agents are blocked from repeated autonomous actions.\"\n   condition: \"identity.risk_tier == 'high' and action.autonomous == True\"\n   action: deny\n   severity: medium\n   owasp_risk: \"Rogue agent \/ Cascading failure\"\n\"\"\"\nwith open(POLICY_PATH, \"w\") as f:\n   f.write(policy_yaml)\nwith open(POLICY_PATH, \"r\") as f:\n   policy = yaml.safe_load(f)<\/code><\/pre>\n<\/div>\n<\/div>\n
We create a YAML governance policy that defines how agent actions should be handled before execution. We add rules to block destructive database actions, require approval for external emails and financial transfers, sandbox shell commands, and restrict low-trust agents from sensitive data. We then save and reload this policy so the rest of the tutorial can use it as the main governance configuration.<\/p>\n
\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
@dataclass\nclass AgentIdentity:\n   agent_id: str\n   name: str\n   role: str\n   owner: str\n   trust_score: float\n   risk_tier: str\n   scopes: List[str]\n@dataclass\nclass GovernanceDecision:\n   decision_id: str\n   timestamp: str\n   policy_name: str\n   agent_id: str\n   agent_name: str\n   tool_name: str\n   action: Dict[str, Any]\n   decision: str\n   matched_rule: Optional[str]\n   severity: Optional[str]\n   reason: str\n   approved_by: Optional[str]\n   previous_hash: str\n   record_hash: str\nclass GovernanceDenied(Exception):\n   pass\nclass ApprovalRequired(Exception):\n   pass\nclass SandboxViolation(Exception):\n   pass\nclass DotDict(dict):\n   def __getattr__(self, item):\n       value = self.get(item)\n       if isinstance(value, dict):\n           return DotDict(value)\n       return value\ndef safe_eval_condition(condition: str, action: Dict[str, Any], identity: AgentIdentity) -> bool:\n   safe_globals = {\n       \"__builtins__\": {},\n       \"True\": True,\n       \"False\": False,\n       \"None\": None,\n   }\n   safe_locals = {\n       \"action\": DotDict(action),\n       \"identity\": DotDict(asdict(identity)),\n   }\n   try:\n       return bool(eval(condition, safe_globals, safe_locals))\n   except Exception as e:\n       return False<\/code><\/pre>\n<\/div>\n<\/div>\n
We define the core data structures for representing agent identities, governance decisions, and governance-related exceptions. We also create a small dot-access dictionary helper so that policy conditions can read values such as action.type and identity.trust_score. We then build a safe condition evaluator that checks whether each policy rule matches the current agent action.<\/p>\n
\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
class TamperEvidentAuditLog:\n   def __init__(self, secret: bytes = b\"tutorial-secret-key\"):\n       self.records: List[GovernanceDecision] = []\n       self.secret = secret\n       self.last_hash = \"GENESIS\"\n   def _hash_record(self, payload: Dict[str, Any], previous_hash: str) -> str:\n       canonical = json.dumps(\n           {\"payload\": payload, \"previous_hash\": previous_hash},\n           sort_keys=True,\n           default=str\n       ).encode()\n       return hmac.new(self.secret, canonical, hashlib.sha256).hexdigest()\n   def append(\n       self,\n       policy_name: str,\n       identity: AgentIdentity,\n       tool_name: str,\n       action: Dict[str, Any],\n       decision: str,\n       matched_rule: Optional[str],\n       severity: Optional[str],\n       reason: str,\n       approved_by: Optional[str] = None\n   ) -> GovernanceDecision:\n       base_payload = {\n           \"decision_id\": str(uuid.uuid4()),\n           \"timestamp\": datetime.now(timezone.utc).isoformat(),\n           \"policy_name\": policy_name,\n           \"agent_id\": identity.agent_id,\n           \"agent_name\": identity.name,\n           \"tool_name\": tool_name,\n           \"action\": action,\n           \"decision\": decision,\n           \"matched_rule\": matched_rule,\n           \"severity\": severity,\n           \"reason\": reason,\n           \"approved_by\": approved_by,\n       }\n       record_hash = self._hash_record(base_payload, self.last_hash)\n       record = GovernanceDecision(\n           **base_payload,\n           previous_hash=self.last_hash,\n           record_hash=record_hash\n       )\n       self.records.append(record)\n       self.last_hash = record_hash\n       return record\n   def verify(self) -> bool:\n       previous = \"GENESIS\"\n       for r in self.records:\n           payload = asdict(r)\n           record_hash = payload.pop(\"record_hash\")\n           previous_hash = payload.pop(\"previous_hash\")\n           if previous_hash != previous:\n               return False\n           expected = self._hash_record(payload, previous_hash)\n           if expected != record_hash:\n               return False\n           previous = record_hash\n       return True\n   def to_dataframe(self) -> pd.DataFrame:\n       return pd.DataFrame([asdict(r) for r in self.records])\naudit_log = TamperEvidentAuditLog()<\/code><\/pre>\n<\/div>\n<\/div>\n
We implement a tamper-evident audit log that records every governance decision made by the system. We use chained hashes, so each new record depends on the previous record, making changes easier to detect. We also add methods to verify the audit chain and convert the records into a dataframe for later analysis.<\/p>\n
\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
class TutorialGovernanceEngine:\n   def __init__(self, policy: Dict[str, Any], audit_log: TamperEvidentAuditLog):\n       self.policy = policy\n       self.audit_log = audit_log\n       self.kill_switch_enabled = False\n       self.error_budget = 5\n       self.recent_denials = 0\n   def activate_kill_switch(self):\n       self.kill_switch_enabled = True\n   def deactivate_kill_switch(self):\n       self.kill_switch_enabled = False\n   def evaluate(\n       self,\n       identity: AgentIdentity,\n       tool_name: str,\n       action: Dict[str, Any]\n   ) -> GovernanceDecision:\n       if self.kill_switch_enabled:\n           return self.audit_log.append(\n               policy_name=self.policy[\"name\"],\n               identity=identity,\n               tool_name=tool_name,\n               action=action,\n               decision=\"deny\",\n               matched_rule=\"global-kill-switch\",\n               severity=\"critical\",\n               reason=\"Global governance kill switch is active.\"\n           )\n       for rule in self.policy.get(\"rules\", []):\n           condition = rule.get(\"condition\", \"\")\n           if safe_eval_condition(condition, action, identity):\n               rule_action = rule.get(\"action\", \"deny\")\n               matched_rule = rule.get(\"name\")\n               severity = rule.get(\"severity\")\n               description = rule.get(\"description\", \"Policy rule matched.\")\n               if rule_action == \"deny\":\n                   self.recent_denials += 1\n                   return self.audit_log.append(\n                       policy_name=self.policy[\"name\"],\n                       identity=identity,\n                       tool_name=tool_name,\n                       action=action,\n                       decision=\"deny\",\n                       matched_rule=matched_rule,\n                       severity=severity,\n                       reason=description\n                   )\n               if rule_action == \"require_approval\":\n                   return self.audit_log.append(\n                       policy_name=self.policy[\"name\"],\n                       identity=identity,\n                       tool_name=tool_name,\n                       action=action,\n                       decision=\"require_approval\",\n                       matched_rule=matched_rule,\n                       severity=severity,\n                       reason=description\n                   )\n               if rule_action == \"sandbox\":\n                   blocked_terms = rule.get(\"sandbox\", {}).get(\"blocked_terms\", [])\n                   command = str(action.get(\"command\", \"\"))\n                   for term in blocked_terms:\n                       if term in command:\n                           self.recent_denials += 1\n                           return self.audit_log.append(\n                               policy_name=self.policy[\"name\"],\n                               identity=identity,\n                               tool_name=tool_name,\n                               action=action,\n                               decision=\"deny\",\n                               matched_rule=matched_rule,\n                               severity=severity,\n                               reason=f\"Sandbox blocked command term: {term}\"\n                           )\n                   return self.audit_log.append(\n                       policy_name=self.policy[\"name\"],\n                       identity=identity,\n                       tool_name=tool_name,\n                       action=action,\n                       decision=\"sandbox\",\n                       matched_rule=matched_rule,\n                       severity=severity,\n                       reason=description\n                   )\n       return self.audit_log.append(\n           policy_name=self.policy[\"name\"],\n           identity=identity,\n           tool_name=tool_name,\n           action=action,\n           decision=self.policy.get(\"default_action\", \"allow\"),\n           matched_rule=None,\n           severity=None,\n           reason=\"No policy rule matched. Default action applied.\"\n       )\nengine = TutorialGovernanceEngine(policy, audit_log)<\/code><\/pre>\n<\/div>\n<\/div>\n
We build the main governance engine that compares each agent action against the YAML policy rules. We handle different outcomes such as deny, approval required, sandbox mode, and default allow. We also include a kill switch that immediately blocks all actions when needed.<\/p>\n
\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
def query_database(table: str, operation: str = \"select\") -> Dict[str, Any]:\n   return {\n       \"status\": \"success\",\n       \"operation\": operation,\n       \"table\": table,\n       \"rows_returned\": random.randint(10, 100)\n   }\ndef send_email(to: str, subject: str, body: str) -> Dict[str, Any]:\n   return {\n       \"status\": \"sent\",\n       \"to\": to,\n       \"subject\": subject,\n       \"body_preview\": body[:80]\n   }\ndef shell_exec(command: str) -> Dict[str, Any]:\n   allowed_commands = [\"echo\", \"date\", \"pwd\", \"ls\"]\n   first = command.strip().split()[0] if command.strip() else \"\"\n   if first not in allowed_commands:\n       return {\n           \"status\": \"blocked_by_tutorial_shell\",\n           \"command\": command,\n           \"reason\": \"Only harmless demo shell commands are executed.\"\n       }\n   result = subprocess.run(\n       command,\n       shell=True,\n       capture_output=True,\n       text=True,\n       timeout=2\n   )\n   return {\n       \"status\": \"executed\",\n       \"command\": command,\n       \"stdout\": result.stdout.strip(),\n       \"stderr\": result.stderr.strip()\n   }\ndef transfer_money(amount: float, destination: str) -> Dict[str, Any]:\n   return {\n       \"status\": \"transferred\",\n       \"amount\": amount,\n       \"destination\": destination\n   }\nclass GovernedTool:\n   def __init__(\n       self,\n       name: str,\n       fn: Callable,\n       engine: TutorialGovernanceEngine,\n       identity: AgentIdentity,\n       approval_simulator: Optional[Callable[[GovernanceDecision], bool]] = None\n   ):\n       self.name = name\n       self.fn = fn\n       self.engine = engine\n       self.identity = identity\n       self.approval_simulator = approval_simulator or (lambda decision: False)\n   def __call__(self, **kwargs):\n       action = dict(kwargs)\n       action.setdefault(\"autonomous\", True)\n       decision = self.engine.evaluate(\n           identity=self.identity,\n           tool_name=self.name,\n           action=action\n       )\n       if decision.decision == \"deny\":\n           raise GovernanceDenied(\n               f\"Action denied by rule '{decision.matched_rule}': {decision.reason}\"\n           )\n       if decision.decision == \"require_approval\":\n           approved = self.approval_simulator(decision)\n           if not approved:\n               raise ApprovalRequired(\n                   f\"Approval required by rule '{decision.matched_rule}': {decision.reason}\"\n               )\n           self.engine.audit_log.append(\n               policy_name=self.engine.policy[\"name\"],\n               identity=self.identity,\n               tool_name=self.name,\n               action=action,\n               decision=\"approved\",\n               matched_rule=decision.matched_rule,\n               severity=decision.severity,\n               reason=\"Human approval simulated for tutorial.\",\n               approved_by=\"tutorial-approver\"\n           )\n       return self.fn(**kwargs)<\/code><\/pre>\n<\/div>\n<\/div>\n
We define sample tools that represent real agent capabilities, including database access, email sending, shell execution, and money transfer. We then create a governed tool wrapper that ensures every tool call passes through the governance engine first. We ensure denied actions stop immediately, that approval-based actions require a simulated approval, and that only approved or allowed actions reach the actual tool.<\/p>\n
\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
research_agent = AgentIdentity(\n   agent_id=\"agent-research-001\",\n   name=\"ResearchAgent\",\n   role=\"market_research\",\n   owner=\"strategy-team\",\n   trust_score=0.91,\n   risk_tier=\"low\",\n   scopes=[\"read_database\", \"web_search\", \"internal_email\"]\n)\nops_agent = AgentIdentity(\n   agent_id=\"agent-ops-002\",\n   name=\"OpsAgent\",\n   role=\"automation\",\n   owner=\"platform-team\",\n   trust_score=0.72,\n   risk_tier=\"medium\",\n   scopes=[\"shell_exec\", \"read_database\"]\n)\nunknown_agent = AgentIdentity(\n   agent_id=\"agent-shadow-999\",\n   name=\"ShadowAgent\",\n   role=\"unknown\",\n   owner=\"unknown\",\n   trust_score=0.42,\n   risk_tier=\"high\",\n   scopes=[\"unknown\"]\n)\nfinance_agent = AgentIdentity(\n   agent_id=\"agent-finance-003\",\n   name=\"FinanceAgent\",\n   role=\"finance_ops\",\n   owner=\"finance-team\",\n   trust_score=0.88,\n   risk_tier=\"low\",\n   scopes=[\"transfer_money\", \"read_database\"]\n)\ndef tutorial_approval_simulator(decision: GovernanceDecision) -> bool:\n   action = decision.action\n   if decision.matched_rule == \"require-approval-for-financial-transaction\":\n       return action.get(\"amount\", 0) <= 5000\n   if decision.matched_rule == \"require-human-approval-for-email\":\n       return \"confidential\" not in str(action).lower()\n   return False\nresearch_db = GovernedTool(\n   name=\"query_database\",\n   fn=query_database,\n   engine=engine,\n   identity=research_agent,\n   approval_simulator=tutorial_approval_simulator\n)\nops_shell = GovernedTool(\n   name=\"shell_exec\",\n   fn=shell_exec,\n   engine=engine,\n   identity=ops_agent,\n   approval_simulator=tutorial_approval_simulator\n)\nshadow_db = GovernedTool(\n   name=\"query_database\",\n   fn=query_database,\n   engine=engine,\n   identity=unknown_agent,\n   approval_simulator=tutorial_approval_simulator\n)\nresearch_email = GovernedTool(\n   name=\"send_email\",\n   fn=send_email,\n   engine=engine,\n   identity=research_agent,\n   approval_simulator=tutorial_approval_simulator\n)\nfinance_transfer = GovernedTool(\n   name=\"transfer_money\",\n   fn=transfer_money,\n   engine=engine,\n   identity=finance_agent,\n   approval_simulator=tutorial_approval_simulator\n)\n<\/code><\/pre>\n<\/div>\n<\/div>\n
We create multiple agents with different roles, trust scores, risk levels, and scopes to simulate a realistic multi-agent environment. We also define an approval simulator that accepts or rejects actions based on simple business logic. We then wrap each tool with the correct agent identity so the governance layer can make identity-aware decisions.<\/p>\n
\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
scenarios = [\n   {\n       \"name\": \"Safe database read\",\n       \"tool\": research_db,\n       \"kwargs\": {\n           \"table\": \"customers\",\n           \"operation\": \"select\",\n           \"type\": \"select\",\n           \"sensitivity\": \"medium\"\n       }\n   },\n   {\n       \"name\": \"Blocked destructive database action\",\n       \"tool\": research_db,\n       \"kwargs\": {\n           \"table\": \"customers\",\n           \"operation\": \"drop\",\n           \"type\": \"drop_table\",\n           \"sensitivity\": \"critical\"\n       }\n   },\n   {\n       \"name\": \"External email requiring approval\",\n       \"tool\": research_email,\n       \"kwargs\": {\n           \"to\": \"partner@example.com\",\n           \"recipient_domain\": \"example.com\",\n           \"subject\": \"Quarterly update\",\n           \"body\": \"Sharing a non-confidential quarterly update.\",\n           \"type\": \"send_email\",\n           \"sensitivity\": \"medium\"\n       }\n   },\n   {\n       \"name\": \"External email denied due to approval rejection\",\n       \"tool\": research_email,\n       \"kwargs\": {\n           \"to\": \"external@example.com\",\n           \"recipient_domain\": \"example.com\",\n           \"subject\": \"Confidential strategy\",\n           \"body\": \"This contains confidential strategy.\",\n           \"type\": \"send_email\",\n           \"sensitivity\": \"critical\"\n       }\n   },\n   {\n       \"name\": \"Safe sandbox shell command\",\n       \"tool\": ops_shell,\n       \"kwargs\": {\n           \"command\": \"echo Agent governance is active\",\n           \"type\": \"shell_exec\",\n           \"sensitivity\": \"low\"\n       }\n   },\n   {\n       \"name\": \"Dangerous shell command blocked\",\n       \"tool\": ops_shell,\n       \"kwargs\": {\n           \"command\": \"rm -rf \/content\/something\",\n           \"type\": \"shell_exec\",\n           \"sensitivity\": \"critical\"\n       }\n   },\n   {\n       \"name\": \"Low-trust agent blocked from sensitive data\",\n       \"tool\": shadow_db,\n       \"kwargs\": {\n           \"table\": \"executive_compensation\",\n           \"operation\": \"select\",\n           \"type\": \"select\",\n           \"sensitivity\": \"critical\"\n       }\n   },\n   {\n       \"name\": \"Financial transfer requiring approval\",\n       \"tool\": finance_transfer,\n       \"kwargs\": {\n           \"amount\": 2500,\n           \"destination\": \"vendor-123\",\n           \"type\": \"transfer_money\",\n           \"sensitivity\": \"high\"\n       }\n   },\n   {\n       \"name\": \"Large financial transfer rejected\",\n       \"tool\": finance_transfer,\n       \"kwargs\": {\n           \"amount\": 15000,\n           \"destination\": \"vendor-999\",\n           \"type\": \"transfer_money\",\n           \"sensitivity\": \"critical\"\n       }\n   },\n]\nresults = []\nfor scenario in scenarios:\n   try:\n       output = scenario[\"tool\"](**scenario[\"kwargs\"])\n       results.append({\n           \"scenario\": scenario[\"name\"],\n           \"status\": \"executed\",\n           \"output\": output\n       })\n   except Exception as e:\n       results.append({\n           \"scenario\": scenario[\"name\"],\n           \"status\": \"blocked_or_pending\",\n           \"error\": str(e)\n       })\naudit_df = audit_log.to_dataframe()\ndisplay_cols = [\n   \"timestamp\",\n   \"agent_name\",\n   \"tool_name\",\n   \"decision\",\n   \"matched_rule\",\n   \"severity\",\n   \"reason\",\n   \"record_hash\"\n]\ndisplay(audit_df[display_cols])\ntest_cases = [\n   {\n       \"name\": \"drop_table must be denied\",\n       \"identity\": research_agent,\n       \"tool_name\": \"query_database\",\n       \"action\": {\"type\": \"drop_table\", \"sensitivity\": \"critical\", \"autonomous\": True},\n       \"expected\": \"deny\"\n   },\n   {\n       \"name\": \"safe select should be allowed\",\n       \"identity\": research_agent,\n       \"tool_name\": \"query_database\",\n       \"action\": {\"type\": \"select\", \"sensitivity\": \"low\", \"autonomous\": True},\n       \"expected\": \"allow\"\n   },\n   {\n       \"name\": \"external email should require approval\",\n       \"identity\": research_agent,\n       \"tool_name\": \"send_email\",\n       \"action\": {\n           \"type\": \"send_email\",\n           \"recipient_domain\": \"example.com\",\n           \"sensitivity\": \"medium\",\n           \"autonomous\": True\n       },\n       \"expected\": \"require_approval\"\n   },\n   {\n       \"name\": \"low trust sensitive access denied\",\n       \"identity\": unknown_agent,\n       \"tool_name\": \"query_database\",\n       \"action\": {\"type\": \"select\", \"sensitivity\": \"critical\", \"autonomous\": True},\n       \"expected\": \"deny\"\n   },\n   {\n       \"name\": \"shell command should enter sandbox\",\n       \"identity\": ops_agent,\n       \"tool_name\": \"shell_exec\",\n       \"action\": {\n           \"type\": \"shell_exec\",\n           \"command\": \"echo hello\",\n           \"sensitivity\": \"low\",\n           \"autonomous\": True\n       },\n       \"expected\": \"sandbox\"\n   },\n]\ntest_results = []\nfor test in test_cases:\n   decision = engine.evaluate(\n       identity=test[\"identity\"],\n       tool_name=test[\"tool_name\"],\n       action=test[\"action\"]\n   )\n   passed = decision.decision == test[\"expected\"]\n   test_results.append({\n       \"test\": test[\"name\"],\n       \"expected\": test[\"expected\"],\n       \"actual\": decision.decision,\n       \"passed\": passed,\n       \"matched_rule\": decision.matched_rule\n   })\ntest_df = pd.DataFrame(test_results)\ndisplay(test_df)\nengine.activate_kill_switch()\ntry:\n   research_db(\n       table=\"customers\",\n       operation=\"select\",\n       type=\"select\",\n       sensitivity=\"low\"\n   )\nexcept Exception as e:\n   pass\nengine.deactivate_kill_switch()\naudit_df = audit_log.to_dataframe()\nsummary = (\n   audit_df\n   .groupby([\"decision\", \"severity\"], dropna=False)\n   .size()\n   .reset_index(name=\"count\")\n   .sort_values(\"count\", ascending=False)\n)\ndisplay(summary)\nagent_summary = (\n   audit_df\n   .groupby([\"agent_name\", \"decision\"])\n   .size()\n   .reset_index(name=\"count\")\n   .sort_values([\"agent_name\", \"count\"], ascending=[True, False])\n)\ndisplay(agent_summary)\ndecision_counts = audit_df[\"decision\"].value_counts()\nplt.figure(figsize=(8, 5))\ndecision_counts.plot(kind=\"bar\")\nplt.title(\"Governance Decisions Across Agent Actions\")\nplt.xlabel(\"Decision\")\nplt.ylabel(\"Count\")\nplt.xticks(rotation=30)\nplt.tight_layout()\nplt.show()\nseverity_counts = audit_df[\"severity\"].fillna(\"none\").value_counts()\nplt.figure(figsize=(8, 5))\nseverity_counts.plot(kind=\"bar\")\nplt.title(\"Governance Events by Severity\")\nplt.xlabel(\"Severity\")\nplt.ylabel(\"Count\")\nplt.xticks(rotation=30)\nplt.tight_layout()\nplt.show()\nG = nx.DiGraph()\nfor _, row in audit_df.iterrows():\n   agent_node = f\"Agent: {row['agent_name']}\"\n   tool_node = f\"Tool: {row['tool_name']}\"\n   decision_node = f\"Decision: {row['decision']}\"\n   rule_node = f\"Rule: {row['matched_rule']}\" if pd.notna(row[\"matched_rule\"]) else \"Rule: default\"\n   G.add_node(agent_node, node_type=\"agent\")\n   G.add_node(tool_node, node_type=\"tool\")\n   G.add_node(decision_node, node_type=\"decision\")\n   G.add_node(rule_node, node_type=\"rule\")\n   G.add_edge(agent_node, tool_node, relation=\"calls\")\n   G.add_edge(tool_node, decision_node, relation=\"produces\")\n   G.add_edge(decision_node, rule_node, relation=\"matched\")\nplt.figure(figsize=(14, 9))\npos = nx.spring_layout(G, seed=42, k=0.8)\nnx.draw_networkx_nodes(G, pos, node_size=1800)\nnx.draw_networkx_edges(G, pos, arrows=True, arrowstyle=\"->\", arrowsize=15)\nnx.draw_networkx_labels(G, pos, font_size=8)\nplt.title(\"Agent Governance Graph: Agents, Tools, Decisions, and Policy Rules\")\nplt.axis(\"off\")\nplt.tight_layout()\nplt.show()\nEXPORT_DIR = \"\/content\/agt_tutorial_outputs\"\nos.makedirs(EXPORT_DIR, exist_ok=True)\naudit_json_path = os.path.join(EXPORT_DIR, \"tamper_evident_audit_log.json\")\naudit_csv_path = os.path.join(EXPORT_DIR, \"governance_audit_log.csv\")\npolicy_copy_path = os.path.join(EXPORT_DIR, \"advanced_agent_policy.yaml\")\ntest_results_path = os.path.join(EXPORT_DIR, \"policy_test_results.csv\")\nwith open(audit_json_path, \"w\") as f:\n   json.dump([asdict(r) for r in audit_log.records], f, indent=2, default=str)\naudit_df.to_csv(audit_csv_path, index=False)\ntest_df.to_csv(test_results_path, index=False)\nshutil.copy(POLICY_PATH, policy_copy_path)<\/code><\/pre>\n<\/div>\n<\/div>\n
We run a set of test scenarios that show how the governed system handles safe actions, risky actions, approval flows, and blocked operations. We display the audit log, run policy tests, activate and deactivate the kill switch, and summarize governance decisions with tables and charts. We also create a governance graph and export the audit logs, policy file, and test results as reusable artifacts.<\/p>\n
In conclusion, we have a fully governed-agent workflow that covers both policy enforcement and observability. We simulated multiple agents with varying trust levels. We showed how the same system responds differently depending on the agent\u2019s identity, the action\u2019s sensitivity, and the rules defined in the policy file. Safe actions, such as simple database reads, are executed. In contrast, risky actions, such as destructive database changes, unsafe shell commands, low-trust sensitive access, and large financial transfers, are blocked or sent for approval. We also recorded every decision in an audit log, verified the audit chain, ran policy tests, exported governance artifacts, and created visual summaries that make the system\u2019s behavior easier to review.<\/p>\n
\n
\n<\/p>
\n<\/p>
Check out\u00a0the\u00a0Full Codes here<\/a>.\u00a0<\/strong>Also,\u00a0feel free to follow us on\u00a0Twitter<\/mark><\/a><\/strong>\u00a0and don\u2019t forget to join our\u00a0150k+ ML SubReddit<\/a><\/strong>\u00a0and Subscribe to\u00a0our Newsletter<\/a><\/strong>. Wait! are you on telegram?\u00a0now you can join us on telegram as well.<\/a><\/strong><\/p>\n
Need to partner with us for promoting your GitHub Repo OR Hugging Face Page OR Product Release OR Webinar etc.?\u00a0Connect with us<\/mark><\/a><\/strong><\/p>\n
The post An Implementation of the Microsoft Agent Governance Toolkit for Safe AI Agent Tool Use with Policies, Approvals, Audit Logs, and Risk Controls<\/a> appeared first on MarkTechPost<\/a>.<\/p>","protected":false},"excerpt":{"rendered":"
In this tutorial, we build a g…<\/p>\n","protected":false},"author":1,"featured_media":29,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[1],"tags":[],"class_list":["post-1016","post","type-post","status-publish","format-standard","has-post-thumbnail","hentry","category-uncategorized"],"_links":{"self":[{"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/posts\/1016","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=%2Fwp%2Fv2%2Fcomments&post=1016"}],"version-history":[{"count":0,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/posts\/1016\/revisions"}],"wp:featuredmedia":[{"embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/media\/29"}],"wp:attachment":[{"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=%2Fwp%2Fv2%2Fmedia&parent=1016"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=%2Fwp%2Fv2%2Fcategories&post=1016"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=%2Fwp%2Fv2%2Ftags&post=1016"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}