{ "nbformat": 4, "nbformat_minor": 5, "metadata": { "kernelspec": { "display_name": "Python 3", "language": "python", "name": "python3" }, "language_info": { "name": "python", "version": "3.13.0" }, "blog_metadata": { "topic": "My Playbook for Building Microsoft 365 Copilot Agents That Survive Contact With the Enterprise", "slug": "my-playbook-for-building-microsoft-365-copilot-agents-that-s", "generated_by": "LinkedIn Post Generator + Azure OpenAI", "generated_at": "2026-08-06T16:41:57.935Z" } }, "cells": [ { "cell_type": "markdown", "metadata": {}, "source": [ "# My Playbook for Building Microsoft 365 Copilot Agents That Survive Contact With the Enterprise\n", "\n", "This notebook turns the blog post into a hands-on validation playbook. It focuses on the operational controls that distinguish a production-ready Microsoft 365 Copilot agent from a polished demo: policy gates, grounding, identity boundaries, approvals, telemetry, and rollback posture.\n", "\n", "All examples are implemented in Python so you can test the patterns locally, even when the original post referenced Mermaid or PowerShell." ] }, { "cell_type": "code", "metadata": {}, "source": [ "%pip install pandas" ], "execution_count": null, "outputs": [] }, { "cell_type": "code", "metadata": {}, "source": [ "from dataclasses import dataclass, asdict\n", "from typing import Dict, List, Any, Optional\n", "import json\n", "import time\n", "import re\n", "import os\n", "import copy\n", "import pandas as pd" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Service contract first\n", "\n", "The post argues that every serious agent needs a service contract before prompt tuning. This cell creates a simple contract structure and validates whether key ownership and control fields are present." ] }, { "cell_type": "code", "metadata": {}, "source": [ "required_contract_fields = [\n", " \"business_outcome\",\n", " \"business_owner\",\n", " \"technical_owner\",\n", " \"support_route\",\n", " \"approved_user_population\",\n", " \"data_boundary\",\n", " \"action_inventory\",\n", " \"retirement_decision\",\n", "]\n", "\n", "service_contract = {\n", " \"business_outcome\": \"Answer HR benefits questions using approved content\",\n", " \"business_owner\": \"HR Operations Lead\",\n", " \"technical_owner\": \"M365 Platform Owner\",\n", " \"support_route\": \"ServiceNow: HR-AI-SUPPORT\",\n", " \"approved_user_population\": [\"HR\", \"People Managers\"],\n", " \"data_boundary\": \"EU\",\n", " \"action_inventory\": [\"retrieve\", \"summarize\", \"draft\"],\n", " \"retirement_decision\": \"Retire if policy content ownership cannot be maintained\",\n", "}\n", "\n", "missing = [f for f in required_contract_fields if f not in service_contract or service_contract[f] in (None, \"\", [])]\n", "print(\"Missing fields:\", missing if missing else \"None\")\n", "pd.DataFrame([service_contract])" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Architecture sketch as executable policy stages\n", "\n", "The original post used a Mermaid flowchart to show that policy checks must happen before grounding and generation. This Python example models the same stages as a simple pipeline." ] }, { "cell_type": "code", "metadata": {}, "source": [ "def policy_check(user: str, intent: str, sensitivity: str) -> Dict[str, Any]:\n", " allowed_sensitivity = {\"public\", \"internal\"}\n", " if sensitivity not in allowed_sensitivity:\n", " return {\"allowed\": False, \"reason\": \"sensitivity_block\"}\n", " return {\"allowed\": True, \"constraints\": {\"citations_required\": True, \"tool_mode\": \"allow_list_only\"}}\n", "\n", "\n", "def grounding_layer(intent: str) -> Dict[str, Any]:\n", " return {\n", " \"sources\": [\n", " \"sharepoint://hr/benefits.pdf\",\n", " \"sharepoint://hr/leave-policy.docx\"\n", " ],\n", " \"scoped_data\": True\n", " }\n", "\n", "\n", "def llm_orchestration(prompt: str, grounding: Dict[str, Any], constraints: Dict[str, Any]) -> Dict[str, Any]:\n", " return {\n", " \"response\": f\"Grounded answer for prompt: {prompt}\",\n", " \"citations\": grounding[\"sources\"] if constraints.get(\"citations_required\") else []\n", " }\n", "\n", "\n", "def run_pipeline(user: str, prompt: str, intent: str, sensitivity: str) -> Dict[str, Any]:\n", " decision = policy_check(user, intent, sensitivity)\n", " if not decision[\"allowed\"]:\n", " return {\"status\": \"denied\", \"message\": \"Safe refusal\", \"reason\": decision[\"reason\"]}\n", " grounding = grounding_layer(intent)\n", " result = llm_orchestration(prompt, grounding, decision[\"constraints\"])\n", " audit_event = {\"user\": user, \"intent\": intent, \"status\": \"ok\", \"sources\": grounding[\"sources\"]}\n", " return {\"status\": \"ok\", \"grounding\": grounding, \"result\": result, \"audit_event\": audit_event}\n", "\n", "pipeline_result = run_pipeline(\n", " user=\"u123\",\n", " prompt=\"Summarize current benefits changes\",\n", " intent=\"benefits_summary\",\n", " sensitivity=\"internal\"\n", ")\n", "print(json.dumps(pipeline_result, indent=2))" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Minimal request pipeline with policy gate, grounding, and audit event\n", "\n", "This is the direct Python example from the post. It blocks higher-sensitivity requests, attaches approved grounding, and emits a simple audit payload." ] }, { "cell_type": "code", "metadata": {}, "source": [ "from dataclasses import dataclass\n", "\n", "@dataclass\n", "class Request:\n", " user_id: str\n", " prompt: str\n", " sensitivity: str\n", "\n", "\n", "def handle_request(req: Request) -> dict:\n", " if req.sensitivity not in {\"public\", \"internal\"}:\n", " return {\"status\": \"denied\", \"message\": \"Request blocked by policy.\"}\n", " grounding = {\"sources\": [\"sharepoint://hr/benefits.pdf\"], \"citations\": True}\n", " response = f\"Answer for {req.user_id}: grounded on approved content.\"\n", " audit = {\"user\": req.user_id, \"action\": \"copilot.invoke\", \"sensitivity\": req.sensitivity}\n", " return {\"status\": \"ok\", \"grounding\": grounding, \"response\": response, \"audit\": audit}\n", "\n", "result = handle_request(Request(\"u123\", \"Summarize benefits changes\", \"internal\"))\n", "print(result)\n", "\n", "blocked = handle_request(Request(\"u123\", \"Summarize benefits changes\", \"confidential\"))\n", "print(blocked)" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Grounding is a data quality decision\n", "\n", "The post emphasizes that grounding quality depends on content ownership and review discipline, not just model behavior. This example scores candidate sources against owner, review cadence, source-of-truth status, and escalation path." ] }, { "cell_type": "code", "metadata": {}, "source": [ "sources = [\n", " {\n", " \"name\": \"benefits.pdf\",\n", " \"owner\": \"HR\",\n", " \"review_cadence\": \"quarterly\",\n", " \"source_of_truth\": True,\n", " \"escalation_path\": \"HR Operations\"\n", " },\n", " {\n", " \"name\": \"teams-thread-export.txt\",\n", " \"owner\": None,\n", " \"review_cadence\": None,\n", " \"source_of_truth\": False,\n", " \"escalation_path\": None\n", " },\n", " {\n", " \"name\": \"leave-policy-2022.pdf\",\n", " \"owner\": \"Legal\",\n", " \"review_cadence\": \"unknown\",\n", " \"source_of_truth\": False,\n", " \"escalation_path\": \"Legal Ops\"\n", " }\n", "]\n", "\n", "\n", "def evaluate_source(source: Dict[str, Any]) -> Dict[str, Any]:\n", " checks = {\n", " \"has_owner\": bool(source.get(\"owner\")),\n", " \"has_review_cadence\": bool(source.get(\"review_cadence\")),\n", " \"is_source_of_truth\": bool(source.get(\"source_of_truth\")),\n", " \"has_escalation_path\": bool(source.get(\"escalation_path\")),\n", " }\n", " score = sum(checks.values())\n", " return {**source, **checks, \"score\": score, \"approved_for_guidance\": score == 4}\n", "\n", "source_report = [evaluate_source(s) for s in sources]\n", "pd.DataFrame(source_report)" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Scenario classification by consequence\n", "\n", "The blog separates discovery, guidance, and action agents. This example classifies scenarios and maps them to expected controls." ] }, { "cell_type": "code", "metadata": {}, "source": [ "scenario_controls = {\n", " \"discovery\": {\"broad_search\": True, \"informational_only\": True, \"approval_required\": False},\n", " \"guidance\": {\"broad_search\": False, \"informational_only\": True, \"approval_required\": False},\n", " \"action\": {\"broad_search\": False, \"informational_only\": False, \"approval_required\": True},\n", "}\n", "\n", "scenarios = [\n", " {\"scenario\": \"Find QBR deck\", \"type\": \"discovery\"},\n", " {\"scenario\": \"Explain leave policy\", \"type\": \"guidance\"},\n", " {\"scenario\": \"Submit employee status change\", \"type\": \"action\"},\n", "]\n", "\n", "rows = []\n", "for s in scenarios:\n", " rows.append({**s, **scenario_controls[s[\"type\"]]})\n", "\n", "pd.DataFrame(rows)" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Identity boundary mapping\n", "\n", "A key point in the post is that 'it will just use the user's context' is not enough. This cell creates an explicit identity map for a request path." ] }, { "cell_type": "code", "metadata": {}, "source": [ "identity_map = {\n", " \"requesting_user_identity\": \"user:u123@contoso.com\",\n", " \"agent_runtime_identity\": \"managed-identity:copilot-agent-prod\",\n", " \"connected_service_identity\": \"service-principal:graph-reader-app\",\n", " \"downstream_system_identity\": \"delegated:servicenow-integration-user\",\n", " \"logging_audit_destination_identity\": \"workspace:log-analytics-prod\"\n", "}\n", "\n", "print(json.dumps(identity_map, indent=2))\n", "\n", "required_identity_keys = list(identity_map.keys())\n", "print(\"Identity map complete:\", all(identity_map.get(k) for k in required_identity_keys))" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Enforce tool allow-list so the agent only calls approved enterprise capabilities\n", "\n", "This is the direct allow-list pattern from the post. The point is to force explicit approval of capabilities rather than letting the agent call whatever it 'needs'." ] }, { "cell_type": "code", "metadata": {}, "source": [ "ALLOWED_TOOLS = {\n", " \"graph.search\",\n", " \"sharepoint.read\",\n", " \"servicenow.create_ticket\",\n", "}\n", "\n", "\n", "def invoke_tool(tool_name: str, payload: dict) -> dict:\n", " if tool_name not in ALLOWED_TOOLS:\n", " raise PermissionError(f\"Tool '{tool_name}' is not approved.\")\n", " return {\"tool\": tool_name, \"status\": \"executed\", \"payload\": payload}\n", "\n", "print(invoke_tool(\"graph.search\", {\"query\": \"Q3 OKRs\"}))\n", "\n", "try:\n", " print(invoke_tool(\"crm.delete_record\", {\"id\": 42}))\n", "except Exception as e:\n", " print(type(e).__name__, str(e))" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Redact obvious secrets before sending prompts or logs downstream\n", "\n", "This is the direct redaction example from the post. It demonstrates a simple data minimization layer for prompts, logs, or tool payloads." ] }, { "cell_type": "code", "metadata": {}, "source": [ "PATTERNS = [\n", " re.compile(r\"\\b\\d{3}-\\d{2}-\\d{4}\\b\"),\n", " re.compile(r\"\\b[A-Z0-9._%+-]+@[A-Z0-9.-]+\\.[A-Z]{2,}\\b\", re.I),\n", " re.compile(r\"\\b(?:\\d[ -]*?){13,16}\\b\"),\n", "]\n", "\n", "\n", "def redact(text: str) -> str:\n", " for pattern in PATTERNS:\n", " text = pattern.sub(\"[REDACTED]\", text)\n", " return text\n", "\n", "sample = \"Email jane@contoso.com and reference SSN 123-45-6789. Card 4111 1111 1111 1111.\"\n", "print(redact(sample))" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Approval is the default pattern for meaningful action\n", "\n", "The post recommends approvals for actions that create commitments, change records, or affect business processes. This example classifies actions by consequence and enforces approval for high-consequence operations." ] }, { "cell_type": "code", "metadata": {}, "source": [ "ACTION_CLASSES = {\n", " \"retrieve\": \"low\",\n", " \"summarize\": \"low\",\n", " \"draft\": \"low\",\n", " \"recommend\": \"medium\",\n", " \"prefill\": \"medium\",\n", " \"prepare\": \"medium\",\n", " \"submit\": \"high\",\n", " \"send\": \"high\",\n", " \"update\": \"high\",\n", " \"create\": \"high\",\n", " \"delete\": \"high\",\n", " \"approve\": \"high\",\n", "}\n", "\n", "\n", "def execute_action(action: str, payload: Dict[str, Any], approved: bool = False) -> Dict[str, Any]:\n", " consequence = ACTION_CLASSES.get(action, \"unknown\")\n", " if consequence == \"high\" and not approved:\n", " return {\n", " \"status\": \"pending_approval\",\n", " \"action\": action,\n", " \"payload\": payload,\n", " \"message\": \"High-consequence action requires approval.\"\n", " }\n", " return {\n", " \"status\": \"executed\",\n", " \"action\": action,\n", " \"payload\": payload,\n", " \"consequence\": consequence\n", " }\n", "\n", "print(execute_action(\"draft\", {\"document\": \"offer letter\"}))\n", "print(execute_action(\"update\", {\"employee_id\": \"E100\", \"status\": \"inactive\"}))\n", "print(execute_action(\"update\", {\"employee_id\": \"E100\", \"status\": \"inactive\"}, approved=True))" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Sequence flow as executable transaction steps\n", "\n", "The original Mermaid sequence diagram showed policy before data fetch and generation, with audit logging as part of the transaction. This Python version simulates that order." ] }, { "cell_type": "code", "metadata": {}, "source": [ "def check_user_intent_scope(user: str, intent: str, data_scope: str) -> Dict[str, Any]:\n", " if data_scope not in {\"public\", \"internal\"}:\n", " return {\"allow\": False, \"constraints\": None}\n", " return {\"allow\": True, \"constraints\": {\"max_sources\": 3, \"citations\": True}}\n", "\n", "\n", "def fetch_scoped_documents(intent: str, max_sources: int) -> List[str]:\n", " docs = [\n", " \"sharepoint://hr/benefits.pdf\",\n", " \"sharepoint://hr/leave-policy.docx\",\n", " \"graph://teams/hr-announcements\"\n", " ]\n", " return docs[:max_sources]\n", "\n", "\n", "def draft_answer(prompt: str, docs: List[str], constraints: Dict[str, Any]) -> Dict[str, Any]:\n", " return {\n", " \"draft\": f\"Draft answer for '{prompt}' using {len(docs)} sources.\",\n", " \"citations\": docs if constraints.get(\"citations\") else []\n", " }\n", "\n", "\n", "def write_audit(event: Dict[str, Any]) -> Dict[str, Any]:\n", " return {\"audit_written\": True, **event}\n", "\n", "\n", "def enterprise_answer_transaction(user: str, prompt: str, intent: str, data_scope: str) -> Dict[str, Any]:\n", " policy = check_user_intent_scope(user, intent, data_scope)\n", " if not policy[\"allow\"]:\n", " return {\"status\": \"denied\"}\n", " docs = fetch_scoped_documents(intent, policy[\"constraints\"][\"max_sources\"])\n", " answer = draft_answer(prompt, docs, policy[\"constraints\"])\n", " audit = write_audit({\"user\": user, \"intent\": intent, \"doc_count\": len(docs)})\n", " return {\"status\": \"ok\", \"answer\": answer, \"audit\": audit}\n", "\n", "print(json.dumps(enterprise_answer_transaction(\"u123\", \"Explain leave policy\", \"policy_guidance\", \"internal\"), indent=2))" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Exception handling branches\n", "\n", "The post asks teams to define behavior for ambiguous intent, dependency failure, denied approval, and stale target records. This example makes those branches explicit." ] }, { "cell_type": "code", "metadata": {}, "source": [ "def process_request(intent_confidence: float, dependency_up: bool, approval_granted: Optional[bool], record_version_matches: bool) -> Dict[str, Any]:\n", " if intent_confidence < 0.7:\n", " return {\"status\": \"needs_clarification\", \"message\": \"Ambiguous intent.\"}\n", " if not dependency_up:\n", " return {\"status\": \"degraded\", \"message\": \"Dependency unavailable. Retry later or route to human support.\"}\n", " if approval_granted is False:\n", " return {\"status\": \"cancelled\", \"message\": \"Approval denied.\"}\n", " if not record_version_matches:\n", " return {\"status\": \"stale\", \"message\": \"Target record changed. Rebuild draft before submit.\"}\n", " return {\"status\": \"ready\", \"message\": \"Proceed.\"}\n", "\n", "cases = [\n", " process_request(0.5, True, None, True),\n", " process_request(0.9, False, None, True),\n", " process_request(0.9, True, False, True),\n", " process_request(0.9, True, True, False),\n", " process_request(0.9, True, True, True),\n", "]\n", "\n", "pd.DataFrame(cases)" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Build surface selection based on control needs\n", "\n", "Instead of asking for a universal winner, the post recommends choosing the build surface that matches control, integration, and lifecycle requirements. This example encodes that decision logic." ] }, { "cell_type": "code", "metadata": {}, "source": [ "def choose_build_surface(bounded_use_case: bool, simple_operating_model: bool, needs_explicit_orchestration: bool, needs_enterprise_integration_contract: bool) -> str:\n", " if needs_enterprise_integration_contract:\n", " return \"Developer extensibility and APIs\"\n", " if needs_explicit_orchestration:\n", " return \"Copilot Studio\"\n", " if bounded_use_case and simple_operating_model:\n", " return \"Scenario-specific authoring / Agent Builder\"\n", " return \"Copilot Studio\"\n", "\n", "examples = [\n", " {\n", " \"scenario\": \"Writing coach\",\n", " \"choice\": choose_build_surface(True, True, False, False)\n", " },\n", " {\n", " \"scenario\": \"Cross-system onboarding orchestration\",\n", " \"choice\": choose_build_surface(False, False, True, False)\n", " },\n", " {\n", " \"scenario\": \"Governed engineering assistant with enterprise APIs\",\n", " \"choice\": choose_build_surface(False, False, True, True)\n", " },\n", "]\n", "\n", "pd.DataFrame(examples)" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Emit structured telemetry for prompt, latency, policy outcome, and citations\n", "\n", "This is the direct telemetry example from the post. It starts with a small, structured event that is easy to query and reason about." ] }, { "cell_type": "code", "metadata": {}, "source": [ "def log_event(name: str, **fields) -> None:\n", " event = {\"event\": name, **fields}\n", " print(json.dumps(event, separators=(\",\", \":\")))\n", "\n", "start = time.time()\n", "policy_outcome = \"allow\"\n", "citations = 2\n", "time.sleep(0.01)\n", "latency_ms = int((time.time() - start) * 1000)\n", "\n", "log_event(\"copilot_request\", user=\"u123\", policy=policy_outcome, latency_ms=latency_ms, citations=citations)" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Telemetry stream and quick operational checks\n", "\n", "The post recommends tracking request volume, policy outcomes, latency, grounding usage, tool success, approvals, feedback, and change history. This example creates a tiny event stream and summarizes it." ] }, { "cell_type": "code", "metadata": {}, "source": [ "events = [\n", " {\"event\": \"copilot_request\", \"user\": \"u1\", \"policy\": \"allow\", \"latency_ms\": 120, \"citations\": 2, \"tool_success\": True},\n", " {\"event\": \"copilot_request\", \"user\": \"u2\", \"policy\": \"deny\", \"latency_ms\": 15, \"citations\": 0, \"tool_success\": None},\n", " {\"event\": \"copilot_request\", \"user\": \"u3\", \"policy\": \"allow\", \"latency_ms\": 210, \"citations\": 1, \"tool_success\": False},\n", " {\"event\": \"copilot_request\", \"user\": \"u1\", \"policy\": \"allow\", \"latency_ms\": 95, \"citations\": 3, \"tool_success\": True},\n", "]\n", "\n", "df = pd.DataFrame(events)\n", "print(df)\n", "print(\"\\nPolicy counts:\")\n", "print(df[\"policy\"].value_counts())\n", "print(\"\\nAverage latency:\", round(df[\"latency_ms\"].mean(), 2), \"ms\")\n", "print(\"Average citations:\", round(df[\"citations\"].mean(), 2))" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Required environment variables for deployment validation\n", "\n", "The original post included a PowerShell example to validate required app settings. For a Python-based notebook, these are the variables you would typically require before deployment:\n", "\n", "- TENANT_ID\n", "- CLIENT_ID\n", "- KEY_VAULT_URI\n", "- GRAPH_SCOPE\n", "- APPINSIGHTS_CONNECTION_STRING" ] }, { "cell_type": "code", "metadata": {}, "source": [ "required = [\n", " \"TENANT_ID\",\n", " \"CLIENT_ID\",\n", " \"KEY_VAULT_URI\",\n", " \"GRAPH_SCOPE\",\n", " \"APPINSIGHTS_CONNECTION_STRING\",\n", "]\n", "\n", "missing = [name for name in required if not os.getenv(name)]\n", "if missing:\n", " print(\"Missing settings:\", \", \".join(missing))\n", "else:\n", " print(\"Environment validation passed.\")" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Required variables for token request payloads\n", "\n", "If you build service-to-service integrations, you need to know which values are required to request access tokens safely. In production, prefer managed identity or a secret store instead of embedding secrets.\n", "\n", "Typical values:\n", "\n", "- TENANT_ID\n", "- CLIENT_ID\n", "- GRAPH_SCOPE\n", "- CLIENT_SECRET or managed identity configuration" ] }, { "cell_type": "code", "metadata": {}, "source": [ "tenant_id = os.getenv(\"TENANT_ID\", \"contoso.onmicrosoft.com\")\n", "client_id = os.getenv(\"CLIENT_ID\", \"11111111-2222-3333-4444-555555555555\")\n", "scope = os.getenv(\"GRAPH_SCOPE\", \"https://graph.microsoft.com/.default\")\n", "\n", "body = {\n", " \"client_id\": client_id,\n", " \"scope\": scope,\n", " \"grant_type\": \"client_credentials\",\n", " \"client_secret\": \"use-managed-identity-or-key-vault-in-real-deployments\"\n", "}\n", "\n", "token_endpoint = f\"https://login.microsoftonline.com/{tenant_id}/oauth2/v2.0/token\"\n", "print(\"Token request payload preview:\")\n", "for k, v in body.items():\n", " print(f\"{k}={v}\")\n", "print(\"token_endpoint=\", token_endpoint)" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Prompt handling flow with minimization, policy, tool approval, grounding, generation, and audit\n", "\n", "The post included a Mermaid flowchart for a safer request path. This Python example executes those same stages in order." ] }, { "cell_type": "code", "metadata": {}, "source": [ "def classify_intent(prompt: str) -> str:\n", " prompt_l = prompt.lower()\n", " if any(word in prompt_l for word in [\"delete\", \"update\", \"submit\", \"send\"]):\n", " return \"action\"\n", " if any(word in prompt_l for word in [\"policy\", \"benefits\", \"guidance\"]):\n", " return \"guidance\"\n", " return \"discovery\"\n", "\n", "\n", "def is_sensitive(prompt: str) -> bool:\n", " return bool(re.search(r\"\\b\\d{3}-\\d{2}-\\d{4}\\b\", prompt)) or \"confidential\" in prompt.lower()\n", "\n", "\n", "def evaluate_policy(intent: str, tool_name: str) -> Dict[str, Any]:\n", " if tool_name not in ALLOWED_TOOLS:\n", " return {\"approved\": False, \"reason\": \"tool_blocked\"}\n", " if intent == \"action\" and tool_name != \"servicenow.create_ticket\":\n", " return {\"approved\": False, \"reason\": \"action_requires_specific_tool\"}\n", " return {\"approved\": True}\n", "\n", "\n", "def safe_request_flow(prompt: str, tool_name: str) -> Dict[str, Any]:\n", " intent = classify_intent(prompt)\n", " minimized_prompt = redact(prompt) if is_sensitive(prompt) else prompt\n", " policy = evaluate_policy(intent, tool_name)\n", " if not policy[\"approved\"]:\n", " return {\"status\": \"refused\", \"intent\": intent, \"reason\": policy[\"reason\"]}\n", " grounding = {\"sources\": [\"sharepoint://approved/source1\"], \"scoped\": True}\n", " answer = f\"Generated {intent} answer for: {minimized_prompt}\"\n", " audit = {\"event\": \"copilot_request\", \"intent\": intent, \"tool\": tool_name, \"grounded\": True}\n", " return {\"status\": \"ok\", \"intent\": intent, \"prompt\": minimized_prompt, \"grounding\": grounding, \"answer\": answer, \"audit\": audit}\n", "\n", "print(json.dumps(safe_request_flow(\"Explain benefits policy for employee 123-45-6789\", \"sharepoint.read\"), indent=2))\n", "print(json.dumps(safe_request_flow(\"Delete employee record\", \"graph.search\"), indent=2))" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Fail closed when deployment drifts from the approved enterprise configuration\n", "\n", "This converts the PowerShell drift-detection idea into Python. If the deployed configuration differs from the approved baseline, rollout should stop." ] }, { "cell_type": "code", "metadata": {}, "source": [ "approved = {\n", " \"AuthMode\": \"ManagedIdentity\",\n", " \"DataBoundary\": \"EU\",\n", " \"PublicNetworkAccess\": \"Disabled\",\n", " \"AuditLogging\": \"Enabled\",\n", "}\n", "\n", "current = {\n", " \"AuthMode\": \"ManagedIdentity\",\n", " \"DataBoundary\": \"EU\",\n", " \"PublicNetworkAccess\": \"Enabled\",\n", " \"AuditLogging\": \"Enabled\",\n", "}\n", "\n", "\n", "def detect_drift(approved_cfg: Dict[str, str], current_cfg: Dict[str, str]) -> List[str]:\n", " drift = []\n", " for key, expected in approved_cfg.items():\n", " actual = current_cfg.get(key)\n", " if expected != actual:\n", " drift.append(f\"{key}: expected={expected} actual={actual}\")\n", " return drift\n", "\n", "\n", "drift = detect_drift(approved, current)\n", "if drift:\n", " print(\"Deployment drift detected:\", \"; \".join(drift))\n", "else:\n", " print(\"Configuration matches approved baseline.\")" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Rollback posture\n", "\n", "The post recommends being able to reverse instructions, knowledge boundaries, tool availability, integration endpoints, audience scope, and availability itself. This example simulates rollback to a known-good baseline." ] }, { "cell_type": "code", "metadata": {}, "source": [ "baseline_config = {\n", " \"instructions_version\": \"v1.2\",\n", " \"knowledge_boundary\": [\"sharepoint://hr/benefits.pdf\"],\n", " \"allowed_tools\": [\"graph.search\", \"sharepoint.read\"],\n", " \"integration_endpoint\": \"https://api.contoso.internal/hr\",\n", " \"audience_scope\": [\"HR\", \"Managers\"],\n", " \"enabled\": True,\n", "}\n", "\n", "current_config = copy.deepcopy(baseline_config)\n", "current_config[\"instructions_version\"] = \"v1.3\"\n", "current_config[\"allowed_tools\"].append(\"crm.delete_record\")\n", "current_config[\"enabled\"] = True\n", "\n", "print(\"Current config before rollback:\")\n", "print(json.dumps(current_config, indent=2))\n", "\n", "current_config = copy.deepcopy(baseline_config)\n", "print(\"\\nConfig after rollback:\")\n", "print(json.dumps(current_config, indent=2))" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Support ownership and production gate\n", "\n", "The final gate in the post is operational accountability. This example checks whether each control area has a named owner before broad rollout." ] }, { "cell_type": "code", "metadata": {}, "source": [ "production_gate = {\n", " \"data_grounding_owner\": \"Content Services Lead\",\n", " \"identity_boundaries_owner\": \"IAM Architect\",\n", " \"approval_paths_owner\": \"Business Process Owner\",\n", " \"telemetry_owner\": \"Observability Lead\",\n", " \"rollback_owner\": \"Release Manager\",\n", " \"support_owner\": \"Service Desk Manager\",\n", "}\n", "\n", "missing_owners = [k for k, v in production_gate.items() if not v]\n", "print(\"Production gate passed:\" , not missing_owners)\n", "print(\"Missing owners:\", missing_owners if missing_owners else \"None\")\n", "pd.DataFrame([production_gate])" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Team readiness scorecard\n", "\n", "The post ends with a practical test: can your team disable a misbehaving agent, explain its last action, and name the owner in under 15 minutes? This scorecard turns that into a simple self-assessment." ] }, { "cell_type": "code", "metadata": {}, "source": [ "scorecard = {\n", " \"can_disable_agent_under_15_min\": 1,\n", " \"can_explain_last_action_under_15_min\": 1,\n", " \"can_name_owner_under_15_min\": 1,\n", " \"has_documented_policy_gate\": 1,\n", " \"has_rollback_runbook\": 1,\n", "}\n", "\n", "score = sum(scorecard.values())\n", "max_score = len(scorecard)\n", "rating_1_to_5 = round((score / max_score) * 5, 1)\n", "\n", "print(\"Score details:\")\n", "print(json.dumps(scorecard, indent=2))\n", "print(f\"\\nReadiness rating: {rating_1_to_5}/5\")" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Next Steps\n", "\n", "This notebook validated the blog's core claim: enterprise agent success is an operations problem, not a demo problem. The practical controls are explicit policy checks, scoped grounding, identity mapping, tool allow-lists, redaction, approvals for meaningful actions, structured telemetry, drift detection, rollback, and named support ownership.\n", "\n", "Next steps:\n", "\n", "1. Turn your current agent idea into a written service contract.\n", "2. Enumerate approved tools, data sources, and identities.\n", "3. Add a policy gate before grounding and generation.\n", "4. Require approval for high-consequence actions.\n", "5. Define telemetry, rollback, and support ownership before rollout.\n", "6. Run the 15-minute readiness test with your team." ] } ] }