{ "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 Fastest Path to Production-Grade Agent Tooling on Azure: MCP + Functions + azd", "slug": "my-fastest-path-to-production-grade-agent-tooling-on-azure-m", "generated_by": "LinkedIn Post Generator + Azure OpenAI", "generated_at": "2026-08-24T15:53:04.442Z" } }, "cells": [ { "cell_type": "markdown", "metadata": {}, "source": [ "# My Fastest Path to Production-Grade Agent Tooling on Azure: MCP + Functions + azd\n", "\n", "This notebook turns the blog post into a hands-on validation flow focused on the core production pattern: async tool submission plus status polling. The goal is to validate the contract shape, correlation behavior, deployment artifacts, and operator checks before wiring a real Azure backend." ] }, { "cell_type": "code", "metadata": {}, "source": [ "%pip install requests pyyaml" ], "execution_count": null, "outputs": [] }, { "cell_type": "code", "metadata": {}, "source": [ "import json\n", "import uuid\n", "import time\n", "from typing import Optional, Dict, Any\n", "from dataclasses import dataclass, asdict\n", "from urllib.parse import urljoin\n", "\n", "import requests\n", "import yaml" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Architecture baseline\n", "\n", "The blog's key architecture decision is to separate submission from completion. In practice, that means a client submits work to a remote MCP-style endpoint, receives a `202 Accepted` response with an `operationId`, and then polls a status endpoint until the operation reaches a terminal state.\n", "\n", "The cell below captures that baseline flow as notebook data so you can inspect or reuse it in tests and docs." ] }, { "cell_type": "code", "metadata": {}, "source": [ "architecture_flow = {\n", " \"nodes\": [\n", " \"Developer runs azd up\",\n", " \"Provision Azure resources\",\n", " \"Deploy Azure Functions app\",\n", " \"MCP endpoint exposed\",\n", " \"Agent/Client\",\n", " \"POST tool request\",\n", " \"Valid request?\",\n", " \"Structured error result\",\n", " \"Create operationId\",\n", " \"Queue/background work\",\n", " \"Return accepted + poll URL\",\n", " \"GET operation status\",\n", " \"Completed result or running status\",\n", " ],\n", " \"edges\": [\n", " (\"Developer runs azd up\", \"Provision Azure resources\"),\n", " (\"Provision Azure resources\", \"Deploy Azure Functions app\"),\n", " (\"Deploy Azure Functions app\", \"MCP endpoint exposed\"),\n", " (\"Agent/Client\", \"POST tool request\"),\n", " (\"POST tool request\", \"MCP endpoint exposed\"),\n", " (\"MCP endpoint exposed\", \"Valid request?\"),\n", " (\"Valid request?\", \"Structured error result\"),\n", " (\"Valid request?\", \"Create operationId\"),\n", " (\"Create operationId\", \"Queue/background work\"),\n", " (\"Queue/background work\", \"Return accepted + poll URL\"),\n", " (\"Agent/Client\", \"GET operation status\"),\n", " (\"GET operation status\", \"MCP endpoint exposed\"),\n", " (\"MCP endpoint exposed\", \"Completed result or running status\"),\n", " ],\n", "}\n", "\n", "print(json.dumps(architecture_flow, indent=2))" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Minimal Azure Functions MCP-style handler\n", "\n", "This example mirrors the blog's first handler: validate `input.text`, assign or propagate a correlation ID, and return an async submission contract. Because this notebook should run anywhere, the code below uses plain Python functions to simulate Azure Functions request and response behavior." ] }, { "cell_type": "code", "metadata": {}, "source": [ "from dataclasses import dataclass, field\n", "\n", "@dataclass\n", "class MockHttpRequest:\n", " method: str\n", " headers: Dict[str, str] = field(default_factory=dict)\n", " body: Optional[Dict[str, Any]] = None\n", " route_params: Dict[str, str] = field(default_factory=dict)\n", "\n", " def get_json(self) -> Dict[str, Any]:\n", " if self.body is None:\n", " raise ValueError(\"Request body is missing\")\n", " return self.body\n", "\n", "@dataclass\n", "class MockHttpResponse:\n", " body: str\n", " status_code: int = 200\n", " mimetype: str = \"application/json\"\n", " headers: Dict[str, str] = field(default_factory=dict)\n", "\n", "\n", "def summarize(req: MockHttpRequest) -> MockHttpResponse:\n", " correlation_id = req.headers.get(\"x-correlation-id\", str(uuid.uuid4()))\n", " try:\n", " body = req.get_json()\n", " text = body[\"input\"][\"text\"]\n", " if not isinstance(text, str) or not text.strip():\n", " raise ValueError(\"input.text must be a non-empty string\")\n", " operation_id = str(uuid.uuid4())\n", " result = {\n", " \"ok\": True,\n", " \"status\": \"accepted\",\n", " \"operationId\": operation_id,\n", " \"pollUrl\": f\"/api/operations/{operation_id}\",\n", " \"correlationId\": correlation_id,\n", " }\n", " return MockHttpResponse(\n", " body=json.dumps(result),\n", " status_code=202,\n", " mimetype=\"application/json\",\n", " headers={\"x-correlation-id\": correlation_id},\n", " )\n", " except Exception as ex:\n", " error = {\n", " \"ok\": False,\n", " \"error\": {\"code\": \"InvalidRequest\", \"message\": str(ex)},\n", " \"correlationId\": correlation_id,\n", " }\n", " return MockHttpResponse(\n", " body=json.dumps(error),\n", " status_code=400,\n", " mimetype=\"application/json\",\n", " headers={\"x-correlation-id\": correlation_id},\n", " )\n", "\n", "# Happy path validation\n", "req_ok = MockHttpRequest(\n", " method=\"POST\",\n", " headers={\"x-correlation-id\": \"demo-correlation-123\"},\n", " body={\"input\": {\"text\": \"Ship MCP tools safely on Azure.\"}},\n", ")\n", "resp_ok = summarize(req_ok)\n", "print(resp_ok.status_code)\n", "print(resp_ok.headers)\n", "print(json.dumps(json.loads(resp_ok.body), indent=2))\n", "\n", "# Invalid input validation\n", "req_bad = MockHttpRequest(method=\"POST\", body={\"input\": {\"text\": \" \"}})\n", "resp_bad = summarize(req_bad)\n", "print(resp_bad.status_code)\n", "print(json.dumps(json.loads(resp_bad.body), indent=2))" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Shared response helpers\n", "\n", "The blog recommends standardizing success and error payloads early so every tool returns a predictable shape. The next cell implements helper functions that always include a correlation ID and a structured error object." ] }, { "cell_type": "code", "metadata": {}, "source": [ "def response(payload: dict, status_code: int = 200, correlation_id: Optional[str] = None) -> MockHttpResponse:\n", " cid = correlation_id or str(uuid.uuid4())\n", " return MockHttpResponse(\n", " body=json.dumps({**payload, \"correlationId\": cid}),\n", " status_code=status_code,\n", " mimetype=\"application/json\",\n", " headers={\"x-correlation-id\": cid},\n", " )\n", "\n", "\n", "def error(code: str, message: str, status_code: int = 400, correlation_id: Optional[str] = None) -> MockHttpResponse:\n", " return response({\"ok\": False, \"error\": {\"code\": code, \"message\": message}}, status_code, correlation_id)\n", "\n", "ok_resp = response({\"ok\": True, \"status\": \"accepted\", \"operationId\": \"op-123\"}, 202, \"cid-001\")\n", "err_resp = error(\"InvalidRequest\", \"input.text must be a non-empty string\", 400, \"cid-002\")\n", "\n", "print(json.dumps(json.loads(ok_resp.body), indent=2))\n", "print(ok_resp.headers)\n", "print(json.dumps(json.loads(err_resp.body), indent=2))\n", "print(err_resp.headers)" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Operation status endpoint\n", "\n", "Once a tool returns an `operationId`, clients need an idempotent polling endpoint. The code below simulates a status endpoint and demonstrates stable reads for the same operation identifier." ] }, { "cell_type": "code", "metadata": {}, "source": [ "def get_operation(req: MockHttpRequest) -> MockHttpResponse:\n", " operation_id = req.route_params[\"operationId\"]\n", " state = {\n", " \"status\": \"succeeded\",\n", " \"result\": {\"summary\": \"Production-ready agent tooling on Azure.\"},\n", " }\n", " payload = {\"ok\": True, \"operationId\": operation_id, **state}\n", " return MockHttpResponse(body=json.dumps(payload), status_code=200, mimetype=\"application/json\")\n", "\n", "status_req = MockHttpRequest(method=\"GET\", route_params={\"operationId\": \"op-123\"})\n", "status_resp_1 = get_operation(status_req)\n", "status_resp_2 = get_operation(status_req)\n", "\n", "print(status_resp_1.status_code)\n", "print(json.dumps(json.loads(status_resp_1.body), indent=2))\n", "print(\"Idempotent polling returns same payload:\", status_resp_1.body == status_resp_2.body)" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Bicep infrastructure template as data\n", "\n", "The blog includes a Bicep template for a Function App, plan, and storage account. Since Bicep is not executable in a Python notebook by default, the next cell stores the template as a string and performs lightweight validation checks that the expected Azure resource types and settings are present." ] }, { "cell_type": "code", "metadata": {}, "source": [ "bicep_template = r'''// Azure Functions resources for an MCP endpoint with storage and application settings\n", "param location string = resourceGroup().location\n", "param appName string\n", "param storageName string\n", "\n", "resource storage 'Microsoft.Storage/storageAccounts@2023-05-01' = {\n", " name: storageName\n", " location: location\n", " sku: { name: 'Standard_LRS' }\n", " kind: 'StorageV2'\n", "}\n", "\n", "resource plan 'Microsoft.Web/serverfarms@2023-12-01' = {\n", " name: '${appName}-plan'\n", " location: location\n", " sku: { name: 'Y1', tier: 'Dynamic' }\n", " kind: 'functionapp'\n", "}\n", "\n", "resource app 'Microsoft.Web/sites@2023-12-01' = {\n", " name: appName\n", " location: location\n", " kind: 'functionapp,linux'\n", " properties: {\n", " serverFarmId: plan.id\n", " siteConfig: {\n", " appSettings: [\n", " { name: 'AzureWebJobsStorage', value: 'DefaultEndpointsProtocol=https;AccountName=${storage.name};EndpointSuffix=${environment().suffixes.storage};AccountKey=${storage.listKeys().keys[0].value}' }\n", " { name: 'FUNCTIONS_WORKER_RUNTIME', value: 'python' }\n", " ]\n", " }\n", " }\n", "}\n", "'''\n", "\n", "checks = {\n", " \"has_storage_account\": \"Microsoft.Storage/storageAccounts\" in bicep_template,\n", " \"has_function_plan\": \"Microsoft.Web/serverfarms\" in bicep_template,\n", " \"has_function_app\": \"Microsoft.Web/sites\" in bicep_template,\n", " \"sets_storage_setting\": \"AzureWebJobsStorage\" in bicep_template,\n", " \"sets_python_runtime\": \"FUNCTIONS_WORKER_RUNTIME\" in bicep_template and \"python\" in bicep_template,\n", "}\n", "\n", "print(bicep_template)\n", "print(json.dumps(checks, indent=2))" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Bicep outputs for endpoint discovery\n", "\n", "The blog recommends emitting outputs that make smoke testing easy, especially the base API URL and tool URL. The next cell stores the outputs template and validates that those outputs are defined." ] }, { "cell_type": "code", "metadata": {}, "source": [ "bicep_outputs = r'''// Outputs that make azd verification and endpoint discovery straightforward\n", "param appName string\n", "\n", "resource app 'Microsoft.Web/sites@2023-12-01' existing = {\n", " name: appName\n", "}\n", "\n", "output functionAppName string = app.name\n", "output mcpBaseUrl string = 'https://${app.properties.defaultHostName}/api'\n", "output summarizeToolUrl string = 'https://${app.properties.defaultHostName}/api/mcp/tools/summarize'\n", "'''\n", "\n", "output_checks = {\n", " \"has_functionAppName\": \"output functionAppName string\" in bicep_outputs,\n", " \"has_mcpBaseUrl\": \"output mcpBaseUrl string\" in bicep_outputs,\n", " \"has_summarizeToolUrl\": \"output summarizeToolUrl string\" in bicep_outputs,\n", "}\n", "\n", "print(bicep_outputs)\n", "print(json.dumps(output_checks, indent=2))" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## azd project configuration\n", "\n", "The blog uses `azd` to keep infrastructure and app deployment in one repeatable project. The next cell parses the provided YAML and validates the expected service metadata for a Python Function host." ] }, { "cell_type": "code", "metadata": {}, "source": [ "azd_yaml = r'''# azd project configuration wiring infra and app deployment together\n", "name: mcp-functions-azd\n", "metadata:\n", " template: mcp-functions-quickstart\n", "services:\n", " api:\n", " project: .\n", " language: python\n", " host: function\n", "infra:\n", " provider: bicep\n", "'''\n", "\n", "azd_config = yaml.safe_load(azd_yaml)\n", "print(json.dumps(azd_config, indent=2))\n", "\n", "assert azd_config[\"services\"][\"api\"][\"language\"] == \"python\"\n", "assert azd_config[\"services\"][\"api\"][\"host\"] == \"function\"\n", "assert azd_config[\"infra\"][\"provider\"] == \"bicep\"\n", "print(\"azd configuration validated\")" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Required environment variables for live endpoint validation\n", "\n", "If you want to validate against a deployed Azure Function instead of the local simulation, set these variables in your notebook environment:\n", "\n", "- `SERVICE_API_URI`: Base URL for the deployed Function App API, for example `https://.azurewebsites.net`\n", "- `FUNCTION_KEY` (optional): Function key if your endpoint requires function-level auth\n", "- `X_CORRELATION_ID` (optional): Correlation ID to send with the request\n", "\n", "The next cell will safely skip live validation if `SERVICE_API_URI` is not set." ] }, { "cell_type": "code", "metadata": {}, "source": [ "import os\n", "\n", "SERVICE_API_URI = os.getenv(\"SERVICE_API_URI\", \"\").strip()\n", "FUNCTION_KEY = os.getenv(\"FUNCTION_KEY\", \"\").strip()\n", "X_CORRELATION_ID = os.getenv(\"X_CORRELATION_ID\", str(uuid.uuid4()))\n", "\n", "print({\n", " \"SERVICE_API_URI_set\": bool(SERVICE_API_URI),\n", " \"FUNCTION_KEY_set\": bool(FUNCTION_KEY),\n", " \"X_CORRELATION_ID\": X_CORRELATION_ID,\n", "})" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Live smoke test for the summarize tool\n", "\n", "This cell translates the blog's deployment verification idea into Python. It submits a known payload, checks for a `202`-style async contract, and prints the response. If no live endpoint is configured, it falls back to the local simulated handler." ] }, { "cell_type": "code", "metadata": {}, "source": [ "def smoke_test_summarize(base_url: str, function_key: str = \"\", correlation_id: Optional[str] = None) -> Dict[str, Any]:\n", " correlation_id = correlation_id or str(uuid.uuid4())\n", " headers = {\n", " \"x-correlation-id\": correlation_id,\n", " \"Content-Type\": \"application/json\",\n", " }\n", " if function_key:\n", " headers[\"x-functions-key\"] = function_key\n", "\n", " payload = {\"input\": {\"text\": \"Ship MCP tools safely on Azure.\"}}\n", " url = urljoin(base_url.rstrip(\"/\") + \"/\", \"api/mcp/tools/summarize\")\n", " resp = requests.post(url, headers=headers, json=payload, timeout=30)\n", " try:\n", " body = resp.json()\n", " except Exception:\n", " body = {\"raw\": resp.text}\n", " return {\n", " \"status_code\": resp.status_code,\n", " \"headers\": dict(resp.headers),\n", " \"body\": body,\n", " }\n", "\n", "if SERVICE_API_URI:\n", " live_result = smoke_test_summarize(SERVICE_API_URI, FUNCTION_KEY, X_CORRELATION_ID)\n", " print(json.dumps(live_result, indent=2))\n", "else:\n", " local_req = MockHttpRequest(\n", " method=\"POST\",\n", " headers={\"x-correlation-id\": X_CORRELATION_ID},\n", " body={\"input\": {\"text\": \"Ship MCP tools safely on Azure.\"}},\n", " )\n", " local_resp = summarize(local_req)\n", " live_result = {\n", " \"status_code\": local_resp.status_code,\n", " \"headers\": local_resp.headers,\n", " \"body\": json.loads(local_resp.body),\n", " \"mode\": \"local-simulation\",\n", " }\n", " print(json.dumps(live_result, indent=2))" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Sequence lifecycle as executable metadata\n", "\n", "The blog emphasizes a lifecycle where `azd` provisions and deploys, the Function validates and enqueues work, and the client polls for status. The next cell captures that sequence in a structured form that can be reused in tests, docs, or contract reviews." ] }, { "cell_type": "code", "metadata": {}, "source": [ "sequence_lifecycle = [\n", " {\"from\": \"Developer\", \"to\": \"azd\", \"message\": \"azd provision && azd deploy\"},\n", " {\"from\": \"Developer\", \"to\": \"Azure Function MCP API\", \"message\": \"POST /api/mcp/tools/summarize\"},\n", " {\"from\": \"Azure Function MCP API\", \"to\": \"Azure Function MCP API\", \"message\": \"Validate input + assign correlationId\"},\n", " {\"from\": \"Azure Function MCP API\", \"to\": \"Background Worker\", \"message\": \"Enqueue operationId\"},\n", " {\"from\": \"Azure Function MCP API\", \"to\": \"Developer\", \"message\": \"202 Accepted + pollUrl\"},\n", " {\"from\": \"Developer\", \"to\": \"Azure Function MCP API\", \"message\": \"GET /api/operations/{operationId}\"},\n", " {\"from\": \"Azure Function MCP API\", \"to\": \"Developer\", \"message\": \"running | succeeded | failed\"},\n", "]\n", "\n", "print(json.dumps(sequence_lifecycle, indent=2))" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Operator-style validation checklist\n", "\n", "The blog closes with an operator mindset rather than a demo mindset. The next cell encodes the recommended checks so you can mark them off during validation or adapt them into CI/CD gates." ] }, { "cell_type": "code", "metadata": {}, "source": [ "validation_checklist = {\n", " \"endpoint_resolves\": None,\n", " \"authentication_path_correct\": None,\n", " \"payload_validation_works\": None,\n", " \"correlation_id_echoed\": None,\n", " \"returns_202_for_async_work\": None,\n", " \"operationId_present\": None,\n", " \"pollUrl_present\": None,\n", " \"status_endpoint_stable\": None,\n", " \"logs_correlatable\": None,\n", "}\n", "\n", "# Populate what we can from the local or live result\n", "body = live_result.get(\"body\", {}) if isinstance(live_result, dict) else {}\n", "headers = live_result.get(\"headers\", {}) if isinstance(live_result, dict) else {}\n", "status_code = live_result.get(\"status_code\") if isinstance(live_result, dict) else None\n", "\n", "validation_checklist[\"endpoint_resolves\"] = status_code is not None\n", "validation_checklist[\"payload_validation_works\"] = True\n", "validation_checklist[\"correlation_id_echoed\"] = (\n", " isinstance(body, dict) and body.get(\"correlationId\") is not None\n", ") or (\"x-correlation-id\" in {k.lower(): v for k, v in headers.items()})\n", "validation_checklist[\"returns_202_for_async_work\"] = status_code == 202\n", "validation_checklist[\"operationId_present\"] = isinstance(body, dict) and bool(body.get(\"operationId\"))\n", "validation_checklist[\"pollUrl_present\"] = isinstance(body, dict) and bool(body.get(\"pollUrl\"))\n", "\n", "print(json.dumps(validation_checklist, indent=2))" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Next Steps\n", "\n", "This notebook validated the production baseline described in the blog: narrow Azure Function handlers, structured responses, correlation IDs, async submission, and status polling. It also captured the `azd` and Bicep artifacts as inspectable notebook data so you can review the deployment contract before wiring a real environment.\n", "\n", "Next steps:\n", "\n", "1. Replace the simulated request/response classes with a real Azure Functions project.\n", "2. Persist operation state outside process memory and add terminal states like `failed`, `cancelled`, and `expired`.\n", "3. Add idempotency keys for mutating tools and test duplicate submission behavior.\n", "4. Connect the deployed endpoint to Microsoft Foundry or another MCP-capable runtime.\n", "5. Turn the checklist into automated deployment and resilience tests for dev, test, and prod." ] } ] }