{
  "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": "How Fabric Local MCP Could Turn Microsoft Fabric Into an Agent Runtime",
      "slug": "how-fabric-local-mcp-could-turn-microsoft-fabric-into-an-age",
      "generated_by": "LinkedIn Post Generator + Azure OpenAI",
      "generated_at": "2026-09-16T17:49:09.823Z"
    }
  },
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "# How Fabric Local MCP Could Turn Microsoft Fabric Into an Agent Runtime\n",
        "\n",
        "This notebook turns the blog post into a hands-on validation workbook focused on architecture, policy, logging, and pilot-readiness patterns for Fabric Local MCP. The goal is not to prove that Fabric hosts the model, but to validate how governed analytics capabilities can become callable tools for agents with explicit controls, identity, and auditability."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "%pip install -q requests python-dotenv"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "import os\n",
        "import json\n",
        "import uuid\n",
        "from dataclasses import dataclass, asdict\n",
        "from datetime import datetime, timezone\n",
        "from typing import Any, Dict, Iterable, List\n",
        "\n",
        "import requests"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Runtime architecture sketch\n",
        "\n",
        "This cell renders the core architecture from the post as executable Python data so you can inspect the runtime boundary. The key idea is that the agent does not get unlimited direct access to Fabric; policy, identity, MCP, and logging sit in the middle."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "architecture = {\n",
        "    \"nodes\": [\n",
        "        \"User / App\",\n",
        "        \"Foundry Agent\",\n",
        "        \"Policy Layer\",\n",
        "        \"Local MCP Gateway\",\n",
        "        \"Microsoft Fabric APIs\",\n",
        "        \"Lakehouse / Warehouse / Semantic Model\",\n",
        "        \"Notebook / Job / Pipeline\",\n",
        "        \"Request + Response Logs\",\n",
        "        \"Managed Identity / Entra Token\",\n",
        "    ],\n",
        "    \"edges\": [\n",
        "        (\"User / App\", \"Foundry Agent\"),\n",
        "        (\"Foundry Agent\", \"Policy Layer\"),\n",
        "        (\"Policy Layer\", \"Local MCP Gateway\", \"allow tool\"),\n",
        "        (\"Local MCP Gateway\", \"Microsoft Fabric APIs\"),\n",
        "        (\"Microsoft Fabric APIs\", \"Lakehouse / Warehouse / Semantic Model\"),\n",
        "        (\"Microsoft Fabric APIs\", \"Notebook / Job / Pipeline\"),\n",
        "        (\"Local MCP Gateway\", \"Request + Response Logs\"),\n",
        "        (\"Policy Layer\", \"Managed Identity / Entra Token\"),\n",
        "    ],\n",
        "}\n",
        "\n",
        "print(\"Architecture nodes:\")\n",
        "for node in architecture[\"nodes\"]:\n",
        "    print(\"-\", node)\n",
        "\n",
        "print(\"\\nArchitecture edges:\")\n",
        "for edge in architecture[\"edges\"]:\n",
        "    print(edge)"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Required environment variables for local pilot validation\n",
        "\n",
        "Several examples use environment variables to simulate enterprise configuration. Set these before running the preflight and connectivity checks:\n",
        "\n",
        "- `FABRIC_WORKSPACE_ID`\n",
        "- `FABRIC_CAPACITY_ID`\n",
        "- `AZURE_TENANT_ID`\n",
        "- `AZURE_CLIENT_ID`\n",
        "- `MCP_BASE_URL`\n",
        "- `FABRIC_API_TOKEN` (optional for simulated authenticated calls)\n",
        "- `AZURE_ACCESS_TOKEN` (optional alternative token name)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Conceptual Foundry agent invocation with MCP tool wiring\n",
        "\n",
        "This example models how an agent request can include MCP tool configuration, allowed actions, and response handling expectations. It is intentionally conceptual so you can validate the request shape before integrating with a real runtime."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "from dataclasses import dataclass\n",
        "from typing import Any, Dict\n",
        "\n",
        "@dataclass\n",
        "class MCPToolConfig:\n",
        "    name: str\n",
        "    endpoint: str\n",
        "    allowed_actions: list[str]\n",
        "    log_requests: bool = True\n",
        "\n",
        "tool = MCPToolConfig(\n",
        "    name=\"fabric-local-mcp\",\n",
        "    endpoint=os.getenv(\"MCP_BASE_URL\", \"http://localhost:8080/mcp\"),\n",
        "    allowed_actions=[\"fabric.query_sql\", \"fabric.run_notebook\"]\n",
        ")\n",
        "\n",
        "agent_request: Dict[str, Any] = {\n",
        "    \"agent\": \"fabric-ops-agent\",\n",
        "    \"input\": \"Summarize yesterday's sales and refresh the semantic model.\",\n",
        "    \"tools\": [{\"type\": \"mcp\", \"config\": tool.__dict__}],\n",
        "    \"response_mode\": \"json\"\n",
        "}\n",
        "\n",
        "print(\"Attach policy checks before invocation:\", tool.allowed_actions)\n",
        "print(\"Send request to Foundry agent runtime:\")\n",
        "print(json.dumps(agent_request, indent=2))\n",
        "print(\"Log request/response IDs and redact secrets before persistence.\")"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Minimal policy gate for MCP tool permissions\n",
        "\n",
        "This example validates the post's main governance point: permissions should map to actions, not just broad role labels. You can change the role and requested actions to test how a simple allow-list behaves."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "from typing import Iterable\n",
        "\n",
        "ALLOWED_BY_ROLE = {\n",
        "    \"analyst\": {\"fabric.query_sql\"},\n",
        "    \"operator\": {\"fabric.query_sql\", \"fabric.run_notebook\", \"fabric.refresh_model\"},\n",
        "}\n",
        "\n",
        "def authorize(role: str, requested_actions: Iterable[str]) -> bool:\n",
        "    granted = ALLOWED_BY_ROLE.get(role, set())\n",
        "    return set(requested_actions).issubset(granted)\n",
        "\n",
        "role = \"operator\"\n",
        "requested = [\"fabric.query_sql\", \"fabric.run_notebook\"]\n",
        "\n",
        "if authorize(role, requested):\n",
        "    print(\"Authorized MCP tool actions:\", requested)\n",
        "else:\n",
        "    raise PermissionError(f\"Role '{role}' cannot use actions: {requested}\")\n",
        "\n",
        "# Negative test\n",
        "try:\n",
        "    bad_requested = [\"fabric.query_sql\", \"fabric.edit_semantic_model\"]\n",
        "    if authorize(\"analyst\", bad_requested):\n",
        "        print(\"Unexpected authorization\")\n",
        "    else:\n",
        "        raise PermissionError(f\"Role 'analyst' cannot use actions: {bad_requested}\")\n",
        "except PermissionError as e:\n",
        "    print(\"Expected denial:\", e)"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Request logging with correlation IDs and redaction\n",
        "\n",
        "This pattern separates operational logging from user-facing output. It adds a correlation ID, UTC timestamp, and basic secret redaction so request records can be persisted without leaking tokens."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "import json\n",
        "import uuid\n",
        "from datetime import datetime, timezone\n",
        "\n",
        "def redact(payload: dict) -> dict:\n",
        "    hidden = {\"token\", \"authorization\", \"client_secret\"}\n",
        "    return {k: (\"***\" if k.lower() in hidden else v) for k, v in payload.items()}\n",
        "\n",
        "request = {\n",
        "    \"action\": \"fabric.query_sql\",\n",
        "    \"workspace_id\": os.getenv(\"FABRIC_WORKSPACE_ID\", \"abc123\"),\n",
        "    \"sql\": \"SELECT TOP 10 * FROM Sales\",\n",
        "    \"token\": \"eyJhbGciOi...\"\n",
        "}\n",
        "\n",
        "log_record = {\n",
        "    \"timestamp\": datetime.now(timezone.utc).isoformat(),\n",
        "    \"correlation_id\": str(uuid.uuid4()),\n",
        "    \"direction\": \"request\",\n",
        "    \"payload\": redact(request),\n",
        "}\n",
        "\n",
        "print(json.dumps(log_record, indent=2))"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Response handling for tool output, user summary, and retry logic\n",
        "\n",
        "This example shows how to keep raw tool results separate from user-safe summaries. It also distinguishes retryable operational failures from escalation-worthy errors."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "from typing import Any, Dict\n",
        "\n",
        "def handle_mcp_response(response: Dict[str, Any]) -> str:\n",
        "    if response.get(\"status\") == \"ok\":\n",
        "        rows = response.get(\"data\", [])\n",
        "        return f\"Tool succeeded with {len(rows)} rows. Summarize for the user here.\"\n",
        "    error = response.get(\"error\", {})\n",
        "    if error.get(\"retryable\"):\n",
        "        return f\"Transient failure: {error.get('message')}. Queue retry or fallback.\"\n",
        "    return f\"Non-retryable failure: {error.get('message')}. Escalate with correlation ID.\"\n",
        "\n",
        "ok_response = {\"status\": \"ok\", \"data\": [{\"sales\": 42}, {\"sales\": 99}]}\n",
        "fail_response = {\"status\": \"error\", \"error\": {\"message\": \"Notebook busy\", \"retryable\": True}}\n",
        "\n",
        "print(handle_mcp_response(ok_response))\n",
        "print(handle_mcp_response(fail_response))"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Sequence validation of the governed tool call path\n",
        "\n",
        "The original post used a sequence diagram. Here we express the same flow as ordered Python steps so you can validate the control points in a notebook."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "sequence_steps = [\n",
        "    \"User asks for data summary + refresh\",\n",
        "    \"Foundry Agent validates requested tool and action scope with Policy Check\",\n",
        "    \"Policy allows fabric.query_sql and optionally refresh_model\",\n",
        "    \"Agent invokes Local MCP with correlation ID\",\n",
        "    \"Local MCP calls Fabric API for SQL / Notebook / Refresh\",\n",
        "    \"Fabric returns results / status\",\n",
        "    \"Local MCP returns structured tool response\",\n",
        "    \"Agent returns final answer with safe summary\",\n",
        "]\n",
        "\n",
        "for i, step in enumerate(sequence_steps, start=1):\n",
        "    print(f\"{i}. {step}\")"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Preflight validation for a local MCP pilot\n",
        "\n",
        "This Python version replaces the PowerShell checklist from the post. It checks required environment variables, token presence, and MCP health so a team can decide whether a pilot is actually runnable."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "REQUIRED_VARS = [\n",
        "    \"FABRIC_WORKSPACE_ID\",\n",
        "    \"FABRIC_CAPACITY_ID\",\n",
        "    \"AZURE_TENANT_ID\",\n",
        "    \"AZURE_CLIENT_ID\",\n",
        "    \"MCP_BASE_URL\",\n",
        "]\n",
        "\n",
        "def check_required_env(required_vars: List[str]) -> Dict[str, Any]:\n",
        "    missing = [name for name in required_vars if not os.getenv(name)]\n",
        "    return {\n",
        "        \"ok\": len(missing) == 0,\n",
        "        \"missing\": missing,\n",
        "    }\n",
        "\n",
        "def check_token_presence() -> Dict[str, Any]:\n",
        "    token = os.getenv(\"FABRIC_API_TOKEN\") or os.getenv(\"AZURE_ACCESS_TOKEN\")\n",
        "    return {\n",
        "        \"ok\": bool(token),\n",
        "        \"token_source\": \"FABRIC_API_TOKEN\" if os.getenv(\"FABRIC_API_TOKEN\") else (\"AZURE_ACCESS_TOKEN\" if os.getenv(\"AZURE_ACCESS_TOKEN\") else None),\n",
        "    }\n",
        "\n",
        "def check_mcp_health(base_url: str, timeout: int = 5) -> Dict[str, Any]:\n",
        "    health_url = base_url.rstrip(\"/\") + \"/health\"\n",
        "    try:\n",
        "        response = requests.get(health_url, timeout=timeout)\n",
        "        return {\n",
        "            \"ok\": response.status_code == 200,\n",
        "            \"status_code\": response.status_code,\n",
        "            \"url\": health_url,\n",
        "        }\n",
        "    except Exception as e:\n",
        "        return {\n",
        "            \"ok\": False,\n",
        "            \"status_code\": None,\n",
        "            \"url\": health_url,\n",
        "            \"error\": str(e),\n",
        "        }\n",
        "\n",
        "env_result = check_required_env(REQUIRED_VARS)\n",
        "token_result = check_token_presence()\n",
        "base_url = os.getenv(\"MCP_BASE_URL\", \"http://localhost:8080\")\n",
        "health_result = check_mcp_health(base_url)\n",
        "\n",
        "preflight = {\n",
        "    \"env\": env_result,\n",
        "    \"token\": token_result,\n",
        "    \"mcp_health\": health_result,\n",
        "    \"ready_for_limited_pilot\": env_result[\"ok\"] and token_result[\"ok\"] and health_result[\"ok\"],\n",
        "}\n",
        "\n",
        "print(json.dumps(preflight, indent=2))"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Pilot decision flow\n",
        "\n",
        "This cell mirrors the preflight flowchart from the post. It turns the boring but necessary checks into a deterministic decision path."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "def pilot_decision(preflight_result: Dict[str, Any]) -> str:\n",
        "    if not preflight_result[\"env\"][\"ok\"]:\n",
        "        return \"Stop pilot: missing required environment variables.\"\n",
        "    if not preflight_result[\"token\"][\"ok\"]:\n",
        "        return \"Stop pilot: no access token available.\"\n",
        "    if not preflight_result[\"mcp_health\"][\"ok\"]:\n",
        "        return \"Stop pilot: MCP /health is not reachable.\"\n",
        "    return \"Run limited pilot and collect logs, failures, and permission gaps.\"\n",
        "\n",
        "print(pilot_decision(preflight))"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Governance classification of MCP actions\n",
        "\n",
        "The blog distinguishes metadata inspection, DAX execution, and semantic-model editing as different risk classes. This cell creates a simple governance registry you can adapt for your own control model."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "governance_registry = {\n",
        "    \"metadata.inspect\": {\n",
        "        \"risk_level\": \"low-to-moderate\",\n",
        "        \"approval_required\": False,\n",
        "        \"notes\": \"Discovery only, but may reveal sensitive structures and business definitions.\",\n",
        "    },\n",
        "    \"dax.execute\": {\n",
        "        \"risk_level\": \"moderate\",\n",
        "        \"approval_required\": False,\n",
        "        \"notes\": \"Governed analytical execution using semantic logic; requires scope, identity, and logging controls.\",\n",
        "    },\n",
        "    \"semantic_model.edit\": {\n",
        "        \"risk_level\": \"high\",\n",
        "        \"approval_required\": True,\n",
        "        \"notes\": \"Change management territory; requires rollback, peer review, deployment path, and accountable ownership.\",\n",
        "    },\n",
        "}\n",
        "\n",
        "print(json.dumps(governance_registry, indent=2))"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Named ownership model before production\n",
        "\n",
        "A recurring theme in the post is that the connector is not the runtime; the operating model is. This cell makes ownership explicit so teams can validate whether production readiness has real accountability."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "required_owners = {\n",
        "    \"agent_lifecycle_owner\": None,\n",
        "    \"fabric_asset_owner\": None,\n",
        "    \"semantic_model_change_owner\": None,\n",
        "    \"identity_and_app_registration_owner\": None,\n",
        "}\n",
        "\n",
        "def production_cutover_ready(owners: Dict[str, Any]) -> bool:\n",
        "    return all(bool(v) for v in owners.values())\n",
        "\n",
        "print(\"Current ownership map:\")\n",
        "print(json.dumps(required_owners, indent=2))\n",
        "print(\"Ready for production cutover?\", production_cutover_ready(required_owners))\n",
        "\n",
        "# Example completed ownership assignment\n",
        "example_owners = {\n",
        "    \"agent_lifecycle_owner\": \"Platform Engineering\",\n",
        "    \"fabric_asset_owner\": \"Data Platform Team\",\n",
        "    \"semantic_model_change_owner\": \"BI Governance Lead\",\n",
        "    \"identity_and_app_registration_owner\": \"IAM Team\",\n",
        "}\n",
        "print(\"\\nExample completed ownership map:\")\n",
        "print(json.dumps(example_owners, indent=2))\n",
        "print(\"Ready for production cutover?\", production_cutover_ready(example_owners))"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Practical pilot scenario validation\n",
        "\n",
        "This cell encodes the recommended first pilot from the post: inspect schema, run approved analytics, return a safe summary, and require human approval for follow-on actions. It helps validate architecture rather than just answer quality."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "pilot_scenario = {\n",
        "    \"scenario\": \"Governed semantic analytics pilot\",\n",
        "    \"steps\": [\n",
        "        \"Agent inspects a semantic model schema\",\n",
        "        \"Agent runs approved DAX or SQL against a known workspace\",\n",
        "        \"Agent returns a user-safe summary\",\n",
        "        \"Human approves any follow-on action\",\n",
        "    ],\n",
        "    \"validation_questions\": [\n",
        "        \"Is the tool scope correct?\",\n",
        "        \"Is the identity path stable?\",\n",
        "        \"Are logs usable?\",\n",
        "        \"Are failures understandable?\",\n",
        "        \"Does the answer respect the semantic layer rather than bypassing it?\",\n",
        "    ],\n",
        "}\n",
        "\n",
        "print(json.dumps(pilot_scenario, indent=2))"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Organization self-rating prompt\n",
        "\n",
        "The post ends with a practical governance question. Use this cell to capture a simple readiness score and rationale for your organization."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "def rate_readiness(score: int, rationale: str) -> Dict[str, Any]:\n",
        "    if score < 1 or score > 5:\n",
        "        raise ValueError(\"Score must be between 1 and 5.\")\n",
        "    return {\n",
        "        \"score\": score,\n",
        "        \"question\": \"If an agent edits a semantic model through MCP tomorrow, do you already know who approves it, who logs it, and who rolls it back?\",\n",
        "        \"rationale\": rationale,\n",
        "    }\n",
        "\n",
        "example_rating = rate_readiness(\n",
        "    3,\n",
        "    \"Approval and logging owners are mostly known, but rollback ownership and standardized change paths are still incomplete.\"\n",
        ")\n",
        "print(json.dumps(example_rating, indent=2))"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Summary\n",
        "\n",
        "Fabric Local MCP is most valuable when it turns governed analytics capabilities into callable tools for agents without collapsing policy, identity, and audit boundaries. The notebook examples validated the core patterns from the post: action-based authorization, request logging with redaction, response separation, preflight checks, ownership assignment, and a narrow pilot design.\n",
        "\n",
        "## Next Steps\n",
        "\n",
        "1. Populate the required environment variables and rerun the preflight checks.\n",
        "2. Replace the simulated MCP endpoint with a real local or hosted test endpoint.\n",
        "3. Extend the policy model to distinguish read, execute, refresh, and edit actions.\n",
        "4. Persist correlation-ID-based logs to a governed store.\n",
        "5. Define named owners for lifecycle, Fabric assets, semantic changes, and identity before any production pilot."
      ]
    }
  ]
}