An Implementation of the Microsoft Agent Governance Toolkit for Safe AI Agent Tool Use with Policies, Approvals, Audit Logs, and Risk Controls

scenarios = [
{
“name”: “Safe database read”,
“tool”: research_db,
“kwargs”: {
“table”: “customers”,
“operation”: “select”,
“type”: “select”,
“sensitivity”: “medium”
}
},
{
“name”: “Blocked destructive database action”,
“tool”: research_db,
“kwargs”: {
“table”: “customers”,
“operation”: “drop”,
“type”: “drop_table”,
“sensitivity”: “critical”
}
},
{
“name”: “External email requiring approval”,
“tool”: research_email,
“kwargs”: {
“to”: “[email protected]”,
“recipient_domain”: “example.com”,
“subject”: “Quarterly update”,
“body”: “Sharing a non-confidential quarterly update.”,
“type”: “send_email”,
“sensitivity”: “medium”
}
},
{
“name”: “External email denied due to approval rejection”,
“tool”: research_email,
“kwargs”: {
“to”: “[email protected]”,
“recipient_domain”: “example.com”,
“subject”: “Confidential strategy”,
“body”: “This contains confidential strategy.”,
“type”: “send_email”,
“sensitivity”: “critical”
}
},
{
“name”: “Safe sandbox shell command”,
“tool”: ops_shell,
“kwargs”: {
“command”: “echo Agent governance is active”,
“type”: “shell_exec”,
“sensitivity”: “low”
}
},
{
“name”: “Dangerous shell command blocked”,
“tool”: ops_shell,
“kwargs”: {
“command”: “rm -rf /content/something”,
“type”: “shell_exec”,
“sensitivity”: “critical”
}
},
{
“name”: “Low-trust agent blocked from sensitive data”,
“tool”: shadow_db,
“kwargs”: {
“table”: “executive_compensation”,
“operation”: “select”,
“type”: “select”,
“sensitivity”: “critical”
}
},
{
“name”: “Financial transfer requiring approval”,
“tool”: finance_transfer,
“kwargs”: {
“amount”: 2500,
“destination”: “vendor-123”,
“type”: “transfer_money”,
“sensitivity”: “high”
}
},
{
“name”: “Large financial transfer rejected”,
“tool”: finance_transfer,
“kwargs”: {
“amount”: 15000,
“destination”: “vendor-999”,
“type”: “transfer_money”,
“sensitivity”: “critical”
}
},
]
results = []
for scenario in scenarios:
try:
output = scenario[“tool”](**scenario[“kwargs”])
results.append({
“scenario”: scenario[“name”],
“status”: “executed”,
“output”: output
})
except Exception as e:
results.append({
“scenario”: scenario[“name”],
“status”: “blocked_or_pending”,
“error”: str(e)
})
audit_df = audit_log.to_dataframe()
display_cols = [
“timestamp”,
“agent_name”,
“tool_name”,
“decision”,
“matched_rule”,
“severity”,
“reason”,
“record_hash”
]
display(audit_df[display_cols])
test_cases = [
{
“name”: “drop_table must be denied”,
“identity”: research_agent,
“tool_name”: “query_database”,
“action”: {“type”: “drop_table”, “sensitivity”: “critical”, “autonomous”: True},
“expected”: “deny”
},
{
“name”: “safe select should be allowed”,
“identity”: research_agent,
“tool_name”: “query_database”,
“action”: {“type”: “select”, “sensitivity”: “low”, “autonomous”: True},
“expected”: “allow”
},
{
“name”: “external email should require approval”,
“identity”: research_agent,
“tool_name”: “send_email”,
“action”: {
“type”: “send_email”,
“recipient_domain”: “example.com”,
“sensitivity”: “medium”,
“autonomous”: True
},
“expected”: “require_approval”
},
{
“name”: “low trust sensitive access denied”,
“identity”: unknown_agent,
“tool_name”: “query_database”,
“action”: {“type”: “select”, “sensitivity”: “critical”, “autonomous”: True},
“expected”: “deny”
},
{
“name”: “shell command should enter sandbox”,
“identity”: ops_agent,
“tool_name”: “shell_exec”,
“action”: {
“type”: “shell_exec”,
“command”: “echo hello”,
“sensitivity”: “low”,
“autonomous”: True
},
“expected”: “sandbox”
},
]
test_results = []
for test in test_cases:
decision = engine.evaluate(
identity=test[“identity”],
tool_name=test[“tool_name”],
action=test[“action”]
)
passed = decision.decision == test[“expected”]
test_results.append({
“test”: test[“name”],
“expected”: test[“expected”],
“actual”: decision.decision,
“passed”: passed,
“matched_rule”: decision.matched_rule
})
test_df = pd.DataFrame(test_results)
display(test_df)
engine.activate_kill_switch()
try:
research_db(
table=”customers”,
operation=”select”,
type=”select”,
sensitivity=”low”
)
except Exception as e:
pass
engine.deactivate_kill_switch()
audit_df = audit_log.to_dataframe()
summary = (
audit_df
.groupby([“decision”, “severity”], dropna=False)
.size()
.reset_index(name=”count”)
.sort_values(“count”, ascending=False)
)
display(summary)
agent_summary = (
audit_df
.groupby([“agent_name”, “decision”])
.size()
.reset_index(name=”count”)
.sort_values([“agent_name”, “count”], ascending=[True, False])
)
display(agent_summary)
decision_counts = audit_df[“decision”].value_counts()
plt.figure(figsize=(8, 5))
decision_counts.plot(kind=”bar”)
plt.title(“Governance Decisions Across Agent Actions”)
plt.xlabel(“Decision”)
plt.ylabel(“Count”)
plt.xticks(rotation=30)
plt.tight_layout()
plt.show()
severity_counts = audit_df[“severity”].fillna(“none”).value_counts()
plt.figure(figsize=(8, 5))
severity_counts.plot(kind=”bar”)
plt.title(“Governance Events by Severity”)
plt.xlabel(“Severity”)
plt.ylabel(“Count”)
plt.xticks(rotation=30)
plt.tight_layout()
plt.show()
G = nx.DiGraph()
for _, row in audit_df.iterrows():
agent_node = f”Agent: {row[‘agent_name’]}”
tool_node = f”Tool: {row[‘tool_name’]}”
decision_node = f”Decision: {row[‘decision’]}”
rule_node = f”Rule: {row[‘matched_rule’]}” if pd.notna(row[“matched_rule”]) else “Rule: default”
G.add_node(agent_node, node_type=”agent”)
G.add_node(tool_node, node_type=”tool”)
G.add_node(decision_node, node_type=”decision”)
G.add_node(rule_node, node_type=”rule”)
G.add_edge(agent_node, tool_node, relation=”calls”)
G.add_edge(tool_node, decision_node, relation=”produces”)
G.add_edge(decision_node, rule_node, relation=”matched”)
plt.figure(figsize=(14, 9))
pos = nx.spring_layout(G, seed=42, k=0.8)
nx.draw_networkx_nodes(G, pos, node_size=1800)
nx.draw_networkx_edges(G, pos, arrows=True, arrowstyle=”->”, arrowsize=15)
nx.draw_networkx_labels(G, pos, font_size=8)
plt.title(“Agent Governance Graph: Agents, Tools, Decisions, and Policy Rules”)
plt.axis(“off”)
plt.tight_layout()
plt.show()
EXPORT_DIR = “/content/agt_tutorial_outputs”
os.makedirs(EXPORT_DIR, exist_ok=True)
audit_json_path = os.path.join(EXPORT_DIR, “tamper_evident_audit_log.json”)
audit_csv_path = os.path.join(EXPORT_DIR, “governance_audit_log.csv”)
policy_copy_path = os.path.join(EXPORT_DIR, “advanced_agent_policy.yaml”)
test_results_path = os.path.join(EXPORT_DIR, “policy_test_results.csv”)
with open(audit_json_path, “w”) as f:
json.dump([asdict(r) for r in audit_log.records], f, indent=2, default=str)
audit_df.to_csv(audit_csv_path, index=False)
test_df.to_csv(test_results_path, index=False)
shutil.copy(POLICY_PATH, policy_copy_path)