{ "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": "Using Copilot Studio’s Agent Node to Orchestrate Business Workflows", "slug": "using-copilot-studio-s-agent-node-to-orchestrate-business-wo", "generated_by": "LinkedIn Post Generator + Azure OpenAI", "generated_at": "2026-07-02T12:38:40.353Z" } }, "cells": [ { "cell_type": "markdown", "metadata": {}, "source": [ "# Using Copilot Studio’s Agent Node to Orchestrate Business Workflows\n", "\n", "This notebook turns the blog post into a hands-on validation workflow in Python. It focuses on the core architectural idea: use an agent for context, routing, and exception handling, while keeping deterministic execution and approvals in controlled workflow layers.\n", "\n", "You will simulate a validation service, test routing outcomes, visualize orchestration paths, and run simple governance and readiness checks." ] }, { "cell_type": "code", "metadata": {}, "source": [ "%pip install fastapi pydantic requests pandas matplotlib" ], "execution_count": null, "outputs": [] }, { "cell_type": "code", "metadata": {}, "source": [ "from typing import Optional, List, Dict, Any\n", "from pydantic import BaseModel\n", "from fastapi import FastAPI\n", "import requests\n", "import pandas as pd\n", "import matplotlib.pyplot as plt\n", "from IPython.display import display\n", "import json" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Orchestration reference pattern\n", "\n", "The blog’s first diagram shows the intended control flow: a user request reaches an Agent Node, the agent calls a validation service, and the result determines whether to continue automatically, ask follow-up questions, or send the case to human review.\n", "\n", "The Python below encodes that same flow as structured data so you can inspect and validate the design." ] }, { "cell_type": "code", "metadata": {}, "source": [ "orchestration_flow = {\n", " 'User submits request in Copilot Studio': ['Agent Node'],\n", " 'Agent Node': ['Validation Webhook'],\n", " 'Validation Webhook': {\n", " 'Complete + Low Risk': 'Downstream Workflow',\n", " 'Missing Data': 'Ask Follow-up Question',\n", " 'High Risk': 'Human Approval Queue'\n", " },\n", " 'Downstream Workflow': ['ERP / CRM / Ticketing'],\n", " 'Human Approval Queue': ['ERP / CRM / Ticketing']\n", "}\n", "\n", "print(json.dumps(orchestration_flow, indent=2))" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## FastAPI-style validation service\n", "\n", "This example is the key implementation pattern from the post. It validates completeness and risk before any downstream business action is triggered, returning a structured decision surface the agent can use safely.\n", "\n", "For notebook portability, the code defines the FastAPI app and also calls the validation function directly." ] }, { "cell_type": "code", "metadata": {}, "source": [ "app = FastAPI()\n", "\n", "class WorkflowRequest(BaseModel):\n", " request_id: str\n", " requester_email: str\n", " amount: float\n", " department: Optional[str] = None\n", " justification: Optional[str] = None\n", "\n", "@app.post('/validate')\n", "def validate(req: WorkflowRequest):\n", " missing = [f for f in ['department', 'justification'] if not getattr(req, f)]\n", " risk_flags = []\n", " if req.amount > 10000:\n", " risk_flags.append('HIGH_VALUE')\n", " if not req.requester_email.endswith('@contoso.com'):\n", " risk_flags.append('EXTERNAL_IDENTITY')\n", " return {\n", " 'request_id': req.request_id,\n", " 'is_complete': len(missing) == 0,\n", " 'missing_fields': missing,\n", " 'risk_flags': risk_flags,\n", " 'route': 'approve' if not missing and not risk_flags else 'review'\n", " }\n", "\n", "sample_request = WorkflowRequest(\n", " request_id='REQ-1001',\n", " requester_email='alex@contoso.com',\n", " amount=2500,\n", " department='Finance',\n", " justification='Quarterly software true-up'\n", ")\n", "\n", "print(validate(sample_request))" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Required variables for live webhook testing\n", "\n", "If you want to call a running validation service over HTTP instead of using direct function calls, define:\n", "\n", "- `VALIDATION_URL`: base URL for the validation endpoint, for example `http://localhost:8000/validate`\n", "\n", "The next code cell uses a safe fallback so the notebook still works without a live server." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Python client pattern for Agent Node webhook calls\n", "\n", "The blog includes a client example showing how an agent could call the validation webhook and branch based on the response. In this notebook, the helper first tries HTTP if a URL is configured, then falls back to the local validation function for reproducible testing." ] }, { "cell_type": "code", "metadata": {}, "source": [ "import os\n", "\n", "VALIDATION_URL = os.getenv('VALIDATION_URL', '').strip()\n", "\n", "payload = {\n", " 'request_id': 'REQ-1042',\n", " 'requester_email': 'alex@contoso.com',\n", " 'amount': 12500,\n", " 'department': 'Finance',\n", " 'justification': 'Quarterly software true-up'\n", "}\n", "\n", "def call_validation(payload: Dict[str, Any]) -> Dict[str, Any]:\n", " if VALIDATION_URL:\n", " response = requests.post(VALIDATION_URL, json=payload, timeout=10)\n", " response.raise_for_status()\n", " return response.json()\n", " req = WorkflowRequest(**payload)\n", " return validate(req)\n", "\n", "result = call_validation(payload)\n", "print('Validation result:', result)\n", "\n", "if result['route'] == 'approve':\n", " print('Trigger downstream workflow')\n", "elif result['missing_fields']:\n", " print('Ask follow-up for:', ', '.join(result['missing_fields']))\n", "else:\n", " print('Send to human review with flags:', ', '.join(result['risk_flags']))" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Sequence simulation for low-risk, missing-data, and high-risk paths\n", "\n", "The original post used a sequence diagram to show how the agent, validation service, workflow engine, and human approver interact. The Python below simulates those same branches with executable logic so you can validate expected behavior." ] }, { "cell_type": "code", "metadata": {}, "source": [ "def simulate_sequence(payload: Dict[str, Any]) -> List[str]:\n", " events = []\n", " events.append('User -> Agent Node: Submit business request')\n", " result = call_validation(payload)\n", " events.append('Agent Node -> Validation Service: POST /validate')\n", " events.append(f\"Validation Service -> Agent Node: {result}\")\n", "\n", " if result['route'] == 'approve':\n", " events.append('Agent Node -> Power Automate Flow: Start downstream workflow')\n", " events.append('Power Automate Flow -> User: Confirmation')\n", " elif result['missing_fields']:\n", " events.append(f\"Agent Node -> User: Ask follow-up question for {', '.join(result['missing_fields'])}\")\n", " else:\n", " events.append(f\"Agent Node -> Human Approver: Create approval task with flags {', '.join(result['risk_flags'])}\")\n", " events.append('Human Approver -> Agent Node: Decision pending/returned')\n", " events.append('Agent Node -> Power Automate Flow: Continue or reject')\n", " return events\n", "\n", "scenarios = [\n", " {\n", " 'request_id': 'REQ-2001',\n", " 'requester_email': 'sam@contoso.com',\n", " 'amount': 500,\n", " 'department': 'IT',\n", " 'justification': 'Mouse replacement'\n", " },\n", " {\n", " 'request_id': 'REQ-2002',\n", " 'requester_email': 'sam@contoso.com',\n", " 'amount': 500,\n", " 'department': None,\n", " 'justification': 'Mouse replacement'\n", " },\n", " {\n", " 'request_id': 'REQ-2003',\n", " 'requester_email': 'vendor@example.com',\n", " 'amount': 25000,\n", " 'department': 'Procurement',\n", " 'justification': 'Contractor onboarding package'\n", " }\n", "]\n", "\n", "for s in scenarios:\n", " print('\\n--- Scenario', s['request_id'], '---')\n", " for event in simulate_sequence(s):\n", " print(event)" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Governance and environment readiness checks in Python\n", "\n", "The blog included PowerShell examples for readiness and governance. Since this notebook is Python-first, the next cells translate those checks into Python so you can validate environment variables, connector availability, and governance controls in the same workflow." ] }, { "cell_type": "code", "metadata": {}, "source": [ "import os\n", "\n", "def readiness_check(environment_name: str = 'Default') -> Dict[str, Any]:\n", " required_vars = ['PP_TENANT_ID', 'PP_CLIENT_ID']\n", " missing_vars = [var for var in required_vars if not os.getenv(var)]\n", " return {\n", " 'EnvironmentName': environment_name,\n", " 'HasTenantId': bool(os.getenv('PP_TENANT_ID')),\n", " 'HasClientId': bool(os.getenv('PP_CLIENT_ID')),\n", " 'MissingVars': ', '.join(missing_vars),\n", " 'Ready': len(missing_vars) == 0\n", " }\n", "\n", "readiness = readiness_check()\n", "print(json.dumps(readiness, indent=2))" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Required variables for readiness checks\n", "\n", "To make the readiness check pass, define these environment variables in your notebook or runtime:\n", "\n", "- `PP_TENANT_ID`\n", "- `PP_CLIENT_ID`\n", "\n", "They are placeholders here for validating deployment discipline rather than making authenticated platform calls." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Governance control check\n", "\n", "This example mirrors the blog’s governance check: verify required connectors and confirm that basic controls such as DLP and audit logging are enabled before piloting orchestration." ] }, { "cell_type": "code", "metadata": {}, "source": [ "def governance_check(\n", " required_connectors: List[str] = None,\n", " enabled_connectors: List[str] = None,\n", " admin_settings: Dict[str, bool] = None\n", ") -> Dict[str, Any]:\n", " required_connectors = required_connectors or ['shared_office365', 'shared_teams', 'shared_approvals']\n", " enabled_connectors = enabled_connectors or ['shared_office365', 'shared_teams']\n", " admin_settings = admin_settings or {\n", " 'DlpPolicyAssigned': True,\n", " 'AuditLoggingOn': True,\n", " 'MakersRestricted': False\n", " }\n", "\n", " missing_connectors = [c for c in required_connectors if c not in enabled_connectors]\n", " issues = []\n", " if missing_connectors:\n", " issues.append(f\"Missing connectors: {', '.join(missing_connectors)}\")\n", " if not admin_settings.get('DlpPolicyAssigned', False):\n", " issues.append('DLP policy not assigned')\n", " if not admin_settings.get('AuditLoggingOn', False):\n", " issues.append('Audit logging disabled')\n", "\n", " return {\n", " 'ConnectorsOk': len(missing_connectors) == 0,\n", " 'GovernanceOk': len(issues) == 0,\n", " 'Issues': ' | '.join(issues)\n", " }\n", "\n", "gov_result = governance_check()\n", "print(json.dumps(gov_result, indent=2))" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Deterministic downstream action stub\n", "\n", "A central point in the post is that the agent should decide, not blindly execute. This function enforces that rule by starting the downstream workflow only when validation explicitly returns `approve`." ] }, { "cell_type": "code", "metadata": {}, "source": [ "def run_workflow(validation_result: dict) -> dict:\n", " if validation_result.get('route') != 'approve':\n", " return {'status': 'skipped', 'reason': 'not approved for automation'}\n", " return {\n", " 'status': 'started',\n", " 'system': 'Power Automate',\n", " 'workflow_name': 'ProcurementRequestOrchestration',\n", " 'request_id': validation_result['request_id']\n", " }\n", "\n", "sample = {'request_id': 'REQ-1042', 'route': 'approve'}\n", "print(run_workflow(sample))\n", "print(run_workflow({'request_id': 'REQ-1043', 'route': 'review'}))" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Batch validation across realistic workflow scenarios\n", "\n", "To make the architecture testable, it helps to run multiple requests through the same decision surface. This cell creates a small dataset covering complete, incomplete, high-value, and external-identity cases." ] }, { "cell_type": "code", "metadata": {}, "source": [ "test_requests = [\n", " {'request_id': 'REQ-3001', 'requester_email': 'ana@contoso.com', 'amount': 1200, 'department': 'HR', 'justification': 'Training materials'},\n", " {'request_id': 'REQ-3002', 'requester_email': 'ben@contoso.com', 'amount': 18000, 'department': 'Finance', 'justification': 'Annual software renewal'},\n", " {'request_id': 'REQ-3003', 'requester_email': 'cara@contoso.com', 'amount': 300, 'department': None, 'justification': 'Office supplies'},\n", " {'request_id': 'REQ-3004', 'requester_email': 'drew@example.com', 'amount': 700, 'department': 'IT', 'justification': 'Peripheral purchase'},\n", " {'request_id': 'REQ-3005', 'requester_email': 'eli@contoso.com', 'amount': 9500, 'department': 'Procurement', 'justification': None},\n", " {'request_id': 'REQ-3006', 'requester_email': 'fay@contoso.com', 'amount': 250, 'department': 'Operations', 'justification': 'Label printer replacement'}\n", "]\n", "\n", "results = []\n", "for req in test_requests:\n", " outcome = call_validation(req)\n", " workflow = run_workflow(outcome)\n", " results.append({**req, **outcome, 'workflow_status': workflow['status']})\n", "\n", "df = pd.DataFrame(results)\n", "display(df)" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Analyze routing outcomes\n", "\n", "This summary helps validate whether your orchestration pattern is balanced correctly. In a real pilot, you would watch how many requests are auto-approved versus sent to review, and whether missing-data rates are too high." ] }, { "cell_type": "code", "metadata": {}, "source": [ "summary = df.assign(\n", " has_missing=df['missing_fields'].apply(lambda x: len(x) > 0),\n", " has_risk=df['risk_flags'].apply(lambda x: len(x) > 0)\n", ")\n", "\n", "route_counts = summary['route'].value_counts().rename_axis('route').reset_index(name='count')\n", "missing_counts = summary['has_missing'].value_counts().rename_axis('has_missing').reset_index(name='count')\n", "risk_counts = summary['has_risk'].value_counts().rename_axis('has_risk').reset_index(name='count')\n", "\n", "print('Route counts')\n", "display(route_counts)\n", "print('Missing field counts')\n", "display(missing_counts)\n", "print('Risk flag counts')\n", "display(risk_counts)" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Visualize orchestration decisions\n", "\n", "A simple chart makes the control pattern easier to inspect. This is useful when discussing pilot design with architects, process owners, and governance teams." ] }, { "cell_type": "code", "metadata": {}, "source": [ "plt.figure(figsize=(6, 4))\n", "plt.bar(route_counts['route'], route_counts['count'])\n", "plt.title('Validation Routing Outcomes')\n", "plt.xlabel('Route')\n", "plt.ylabel('Count')\n", "plt.tight_layout()\n", "plt.show()" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Deployment readiness decision flow in Python\n", "\n", "The final diagram in the blog described a readiness path: environment configuration, connector availability, governance controls, then pilot execution. The code below turns that into a reusable decision helper." ] }, { "cell_type": "code", "metadata": {}, "source": [ "def deployment_readiness_flow() -> Dict[str, Any]:\n", " readiness = readiness_check()\n", " governance = governance_check()\n", "\n", " if not readiness['Ready']:\n", " return {\n", " 'stage': 'Environment Configured?',\n", " 'status': 'No',\n", " 'action': 'Fix tenant/app settings',\n", " 'details': readiness\n", " }\n", " if not governance['ConnectorsOk']:\n", " return {\n", " 'stage': 'Connectors Available?',\n", " 'status': 'No',\n", " 'action': 'Enable required connectors',\n", " 'details': governance\n", " }\n", " if not governance['GovernanceOk']:\n", " return {\n", " 'stage': 'Governance Controls OK?',\n", " 'status': 'No',\n", " 'action': 'Apply DLP / audit / admin policies',\n", " 'details': governance\n", " }\n", " return {\n", " 'stage': 'Pilot Agent Node orchestration',\n", " 'status': 'Yes',\n", " 'action': 'Proceed with bounded pilot',\n", " 'details': {\n", " 'readiness': readiness,\n", " 'governance': governance\n", " }\n", " }\n", "\n", "print(json.dumps(deployment_readiness_flow(), indent=2))" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Summary\n", "\n", "This notebook validated the blog’s main design pattern: use Copilot Studio’s Agent Node for ambiguous intake, context gathering, routing, and exception handling, while reserving deterministic execution for workflow systems such as Power Automate.\n", "\n", "You also tested a structured validation contract, simulated approval and follow-up paths, and translated governance checks into Python so the orchestration model can be evaluated before rollout.\n", "\n", "## Next Steps\n", "\n", "1. Replace the local validation function with a real hosted endpoint.\n", "2. Add business-specific rules for approvals, SLAs, and exception thresholds.\n", "3. Connect the `approve` path to an actual downstream workflow engine.\n", "4. Expand governance checks to include environment, connector, and audit requirements from your tenant.\n", "5. Pilot one tightly bounded process such as procurement approval, HR intake triage, or IT access routing." ] } ] }