<\/p>\n\n\n\n
<\/p>\n\n\n\n
<\/p>\n\n\n\n
first, to. develop function apps, install the. commandline —<\/p>\n\n\n\n
npm install -g azure-functions-core-tools@4 –unsafe-perm true<\/em><\/strong> (windows machine)<\/p>\n\n\n\n I would not simply add a SQL query inside the current MCP functions<\/strong>. I would add a separate SQL \u2192 AI Search synchronization pipeline<\/strong> inside the same Function App.<\/p>\n\n\n\n That gives you:<\/p>\n\n\n\n Azure AI Search’s Python SDK supports Entra authentication with There are also a few bugs\/issues in your current code that I would fix at the same time.<\/p>\n\n\n\n You currently have:<\/p>\n\n\n\n This is important because a typo silently causes your application to use:<\/p>\n\n\n\n instead of the configured index.<\/p>\n\n\n\n Currently:<\/p>\n\n\n\n We need:<\/p>\n\n\n\n We don’t want:<\/p>\n\n\n\n every five minutes.<\/p>\n\n\n\n We want:<\/p>\n\n\n\n For simple SQL-to-Search synchronization, Azure AI Search also has a native SQL indexer with change tracking support.<\/p>\n\n\n\n But your Function App gives you more control over transformation, authorization metadata, joins, enrichment, telemetry and MCP-specific documents<\/strong>, so I think the Function-based pipeline makes sense for your architecture.<\/p>\n\n\n\n Your single Once tested, I would split these into modules.<\/p>\n\n\n\n Add these to For example:<\/p>\n\n\n\n The Search SDK supports Add:<\/p>\n\n\n\n Example:<\/p>\n\n\n\n The overlap in For a mature implementation, I would replace the lookback approach with SQL Do not<\/strong> put this in your Function App:<\/p>\n\n\n\n Use:<\/p>\n\n\n\n Your Function’s managed identity should have the minimum required SQL permissions.<\/p>\n\n\n\n For example:<\/p>\n\n\n\n If you later need the Function to write synchronization state back into SQL, create a separate least-privilege role.<\/p>\n\n\n\n Replace your existing:<\/p>\n\n\n\n with:<\/p>\n\n\n\n This fixes your typo and avoids unnecessarily constructing credentials repeatedly.<\/p>\n\n\n\n Add:<\/p>\n\n\n\n Then:<\/p>\n\n\n\n This allows the Function App to authenticate to Azure SQL through its managed identity.<\/p>\n\n\n\n Now add:<\/p>\n\n\n\n This is important because now you can retrieve from:<\/p>\n\n\n\n without duplicating the connection logic.<\/p>\n\n\n\n For your environment, I strongly recommend creating Search-specific SQL views<\/strong> rather than allowing the Function to understand every underlying relational table.<\/p>\n\n\n\n For example:<\/p>\n\n\n\n Then:<\/p>\n\n\n\n with:<\/p>\n\n\n\n This gives you a clean contract:<\/p>\n\n\n\n This is the part I think will make your implementation substantially better.<\/p>\n\n\n\n Add:<\/p>\n\n\n\n Then:<\/p>\n\n\n\n Now SQL:<\/p>\n\n\n\n can become:<\/p>\n\n\n\n That’s much better for AI Search and MCP.<\/p>\n\n\n\n Add:<\/p>\n\n\n\n Now we can build the reusable synchronization layer:<\/p>\n\n\n\n For example:<\/p>\n\n\n\n Now:<\/p>\n\n\n\n I intentionally don’t make one failure stop every other source.<\/p>\n\n\n\n For example:<\/p>\n\n\n\n is much better operationally than:<\/p>\n\n\n\n Your Function App currently doesn’t show a timer.<\/p>\n\n\n\n Add:<\/p>\n\n\n\n The Python v2 Functions model supports this decorator-based timer pattern, and Microsoft recommends For your development environment, this is extremely useful.<\/p>\n\n\n\n I deliberately recommend Your existing:<\/p>\n\n\n\n is something I would reconsider for a production techwyns application.<\/p>\n\n\n\n I would also add:<\/p>\n\n\n\n I would add a function specifically for testing the sources:<\/p>\n\n\n\n I would test in this order.<\/p>\n\n\n\n Call:<\/p>\n\n\n\n Expected:<\/p>\n\n\n\n Call:<\/p>\n\n\n\n Expected:<\/p>\n\n\n\n Execute:<\/p>\n\n\n\n Verify:<\/p>\n\n\n\n Verify:<\/p>\n\n\n\n has a valid:<\/p>\n\n\n\n Run:<\/p>\n\n\n\n Expected:<\/p>\n\n\n\n Query:<\/p>\n\n\n\n or your existing MCP tool.<\/p>\n\n\n\n Verify the SQL-derived record appears.<\/p>\n\n\n\n Change one SQL record:<\/p>\n\n\n\n Run synchronization.<\/p>\n\n\n\n Search:<\/p>\n\n\n\n Verify:<\/p>\n\n\n\n appears.<\/p>\n\n\n\n Temporarily use an invalid SQL view:<\/p>\n\n\n\n The Function should report:<\/p>\n\n\n\n without taking down the other synchronization jobs.<\/p>\n\n\n\n The implementation above handles:<\/p>\n\n\n\n but not:<\/p>\n\n\n\n For production, you need deletion handling.<\/p>\n\n\n\n Otherwise:<\/p>\n\n\n\n That’s dangerous.<\/p>\n\n\n\n I recommend adding a soft-delete field:<\/p>\n\n\n\n Then export:<\/p>\n\n\n\n or explicitly call:<\/p>\n\n\n\n The Search SDK supports deleting documents from an index.<\/p>\n\n\n\n For your production techwyns implementation, I would ultimately move away from:<\/p>\n\n\n\n and use SQL change tracking or another durable high-water-mark mechanism.<\/p>\n\n\n\n That gives you:<\/p>\n\n\n\n This avoids relying on a time window.<\/p>\n\n\n\n And if the synchronization function fails:<\/p>\n\n\n\n That’s the behavior you want.<\/p>\n\n\n\n Your current MCP search function:<\/p>\n\n\n\n is fine for the staff-letter use case.<\/p>\n\n\n\n I would not mix SQL synchronization into that function<\/strong>.<\/p>\n\n\n\n Instead:<\/p>\n\n\n\n The SQL sync is an ingestion pipeline<\/strong>, not an MCP tool.<\/p>\n\n\n\n The MCP tool should only read\/query the resulting index.<\/p>\n\n\n\n You could then expose:<\/p>\n\n\n\n Architecture:<\/p>\n\n\n\n I would now have:<\/p>\n\n\n\n Or, if your current For the techwyns implementation, I would make every SQL-derived Search document carry:<\/p>\n\n\n\n For example:<\/p>\n\n\n\n Then your MCP authorization layer constructs the Search filter.<\/p>\n\n\n\n The LLM never determines these permissions.<\/strong><\/p>\n\n\n\n You currently have:<\/p>\n\n\n\n I would be cautious about leaving the entire application anonymous in production.<\/p>\n\n\n\n Especially because you now want:<\/p>\n\n\n\n A better production topology is:<\/p>\n\n\n\n The Function’s MCP endpoint should not be an unrestricted anonymous endpoint.<\/p>\n\n\n\n This:<\/p>\n\n\n\n could potentially dump sensitive MCP context into Application Insights.<\/p>\n\n\n\n I would change it to:<\/p>\n\n\n\n Likewise:<\/p>\n\n\n\n should be redacted or removed in production.<\/p>\n\n\n\n For techwyns data, don’t routinely put MCP arguments, case content, prompts, tokens, or retrieved evidence into Application Insights<\/strong>.<\/p>\n\n\n\n Log:<\/p>\n\n\n\n rather than the sensitive payload.<\/p>\n\n\n\n Currently:<\/p>\n\n\n\n That’s acceptable for development but can expose internal information.<\/p>\n\n\n\n Production:<\/p>\n\n\n\n That keeps SQL connection errors, internal URLs, stack details, etc. away from the agent.<\/p>\n\n\n\n With these changes, your Function App becomes:<\/p>\n\n\n\n For your first implementation<\/strong>, use the code pattern above with SQL views + incremental That gives you a clean path from your current Function App to a production techwyns data ingestion\/MCP architecture without disrupting your existing staff-letter semantic search. Microsoft supports exactly the Entra-based Search client pattern and batch document operations we’re using here.<\/p>\n\n\n\n <\/p>\n\n\n\n <\/p>\n\n\n\n complete I have preserved your existing:<\/p>\n\n\n\n and added:<\/p>\n\n\n\n Azure AI Search supports Azure SQL\n \u250c\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2510\n \u2502 Tables \u2502\n \u2502 Views \u2502\n \u2502 Stored queries \u2502\n \u2514\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u252c\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2518\n \u2502\n Managed Identity\n \u2502\n \u25bc\n \u250c\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2510\n \u2502 Function App \u2502\n \u2502 \u2502\n \u2502 SQL extraction \u2502\n \u2502 transformation \u2502\n \u2502 validation \u2502\n \u2502 security metadata \u2502\n \u2502 batching \u2502\n \u2502 telemetry \u2502\n \u2514\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u252c\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2518\n \u2502\n \u25bc\n \u250c\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2510\n \u2502 Azure AI Search \u2502\n \u2502 \u2502\n \u2502 csl-metadata \u2502\n \u2502 staff-letters-new \u2502\n \u2502 SQL index \u2502\n \u2514\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u252c\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2518\n \u2502\n \u25bc\n MCP Tools\n \u2502\n \u25bc\n Foundry Agent Service<\/code><\/pre>\n\n\n\nDefaultAzureCredential<\/code> and supports batch merge_or_upload_documents<\/code>, which is exactly what we need here.<\/p>\n\n\n\n
\n\n\n\n1. Issues I found in your existing code<\/h1>\n\n\n\n
Issue 1 \u2014 typo in the index setting<\/h3>\n\n\n\n
index_name=os.environ.get(\"AZURE_SEARCH_INDEX_NAE\", \"csl-metadata\")<\/code><\/pre>\n\n\n\nNAE<\/code> should almost certainly be:<\/p>\n\n\n\nAZURE_SEARCH_INDEX_NAME<\/code><\/pre>\n\n\n\ncsl-metadata<\/code><\/pre>\n\n\n\n
\n\n\n\nIssue 2 \u2014 no SQL connectivity layer<\/h3>\n\n\n\n
Function\n \u2502\n \u251c\u2500\u2500 AI Search\n \u2514\u2500\u2500 Blob<\/code><\/pre>\n\n\n\nFunction\n \u2502\n \u251c\u2500\u2500 AI Search\n \u251c\u2500\u2500 Blob\n \u2514\u2500\u2500 Azure SQL\n \u2502\n \u251c\u2500\u2500 tables\n \u251c\u2500\u2500 views\n \u2514\u2500\u2500 stored queries<\/code><\/pre>\n\n\n\n
\n\n\n\nIssue 3 \u2014 no incremental synchronization<\/h3>\n\n\n\n
SQL 5 million rows\n \u2193\nFunction\n \u2193\nAI Search<\/code><\/pre>\n\n\n\nSQL\n \u2193\nOnly rows changed since last successful sync\n \u2193\nAI Search<\/code><\/pre>\n\n\n\n
\n\n\n\n2. I recommend this final Function App structure<\/h1>\n\n\n\n
function_app.py<\/code> can initially contain everything, but logically divide it into:<\/p>\n\n\n\nfunction_app.py\n\u2502\n\u251c\u2500\u2500 Configuration\n\u2502\n\u251c\u2500\u2500 Credentials\n\u2502\n\u251c\u2500\u2500 AI Search clients\n\u2502\n\u251c\u2500\u2500 Blob functions\n\u2502\n\u251c\u2500\u2500 Existing MCP search\n\u2502\n\u251c\u2500\u2500 Existing semantic search\n\u2502\n\u251c\u2500\u2500 SQL connection\n\u2502\n\u251c\u2500\u2500 SQL extraction\n\u2502\n\u251c\u2500\u2500 SQL \u2192 Search transformation\n\u2502\n\u251c\u2500\u2500 Search batch upload\n\u2502\n\u251c\u2500\u2500 SQL synchronization\n\u2502\n\u251c\u2500\u2500 Timer trigger\n\u2502\n\u2514\u2500\u2500 Health\/test endpoints<\/code><\/pre>\n\n\n\n
\n\n\n\n3. Required packages<\/h1>\n\n\n\n
requirements.txt<\/code>:<\/p>\n\n\n\nazure-functions\nazure-identity\nazure-search-documents\nazure-storage-blob\npypdf\npyodbc<\/code><\/pre>\n\n\n\nazure-functions==1.*\nazure-identity==1.*\nazure-search-documents==11.*\nazure-storage-blob==12.*\npypdf==6.*\npyodbc==5.*<\/code><\/pre>\n\n\n\nSearchClient<\/code> with DefaultAzureCredential<\/code>, and the same client supports adding\/updating\/deleting documents.<\/p>\n\n\n\n
\n\n\n\n4. Environment variables<\/h1>\n\n\n\n
AZURE_SEARCH_ENDPOINT\nAZURE_SEARCH_INDEX_NAME\nAZURE_SEARCH_SEMANTIC_INDEX_NAME\nAZURE_SEARCH_SEMANTIC_CONFIGURATION_NAME\n\nSQL_SERVER\nSQL_DATABASE\nSQL_DRIVER\n\nSQL_SYNC_SCHEDULE\nSQL_SYNC_BATCH_SIZE\nSQL_SYNC_LOOKBACK_MINUTES<\/code><\/pre>\n\n\n\nAZURE_SEARCH_ENDPOINT=https:\/\/x-search.search.windows.net\n\nAZURE_SEARCH_INDEX_NAME=csl-metadata\n\nAZURE_SEARCH_SEMANTIC_INDEX_NAME=-data-new\n\nAZURE_SEARCH_SEMANTIC_CONFIGURATION_NAME=staff-letters-new-semantic-configuration\n\nSQL_SERVER=-sql.database.windows.net\n\nSQL_DATABASE=techwyns\n\nSQL_DRIVER=ODBC Driver 18 for SQL Server\n\nSQL_SYNC_SCHEDULE=0 *\/5 * * * *\n\nSQL_SYNC_BATCH_SIZE=500\n\nSQL_SYNC_LOOKBACK_MINUTES=10<\/code><\/pre>\n\n\n\nSQL_SYNC_LOOKBACK_MINUTES<\/code> protects against small timing differences between the SQL update timestamp and the Function execution.<\/p>\n\n\n\nrowversion<\/code>\/change tracking.<\/p>\n\n\n\n
\n\n\n\n5. Azure SQL authentication<\/h1>\n\n\n\n
SQL username\nSQL password<\/code><\/pre>\n\n\n\nFunction App Managed Identity\n \u2502\n \u25bc\n Azure SQL<\/code><\/pre>\n\n\n\nCREATE USER [x-function-app]\nFROM EXTERNAL PROVIDER;\n\nALTER ROLE db_datareader\nADD MEMBER [x-function-app];<\/code><\/pre>\n\n\n\n
\n\n\n\n6. Corrected Search client<\/h1>\n\n\n\n
def get_search_client() -> SearchClient:<\/code><\/pre>\n\n\n\nfrom functools import lru_cache\n\n@lru_cache(maxsize=1)\ndef get_credential() -> DefaultAzureCredential:\n return DefaultAzureCredential()\n\n\n@lru_cache(maxsize=20)\ndef get_search_client(index_name: str | None = None) -> SearchClient:\n\n endpoint = os.environ[\"AZURE_SEARCH_ENDPOINT\"]\n\n index_name = (\n index_name\n or os.environ.get(\n \"AZURE_SEARCH_INDEX_NAME\",\n \"csl-metadata\",\n )\n )\n\n return SearchClient(\n endpoint=endpoint,\n index_name=index_name,\n credential=get_credential(),\n )<\/code><\/pre>\n\n\n\n
\n\n\n\n7. Add SQL connection support<\/h1>\n\n\n\n
import struct\nimport pyodbc<\/code><\/pre>\n\n\n\nSQL_ACCESS_TOKEN = 1256\n\n\ndef get_sql_connection() -> pyodbc.Connection:\n\n server = os.environ[\"SQL_SERVER\"]\n database = os.environ[\"SQL_DATABASE\"]\n\n driver = os.environ.get(\n \"SQL_DRIVER\",\n \"ODBC Driver 18 for SQL Server\",\n )\n\n credential = get_credential()\n\n token = credential.get_token(\n \"https:\/\/database.windows.net\/.default\"\n )\n\n token_bytes = token.token.encode(\n \"utf-16-le\"\n )\n\n token_struct = struct.pack(\n f\"<I{len(token_bytes)}s\",\n len(token_bytes),\n token_bytes,\n )\n\n connection_string = (\n f\"Driver={{{driver}}};\"\n f\"Server=tcp:{server},1433;\"\n f\"Database={database};\"\n \"Encrypt=yes;\"\n \"TrustServerCertificate=no;\"\n \"Connection Timeout=30;\"\n )\n\n return pyodbc.connect(\n connection_string,\n attrs_before={\n SQL_ACCESS_TOKEN: token_struct\n },\n )<\/code><\/pre>\n\n\n\n
\n\n\n\n8. Generic SQL query function<\/h1>\n\n\n\n
def execute_sql_query(\n sql: str,\n parameters: tuple[Any, ...] = (),\n) -> list[dict[str, Any]]:\n\n connection = None\n\n try:\n\n connection = get_sql_connection()\n\n cursor = connection.cursor()\n\n logging.info(\n \"Executing SQL query.\"\n )\n\n cursor.execute(\n sql,\n parameters,\n )\n\n columns = [\n column[0]\n for column in cursor.description\n ]\n\n rows = cursor.fetchall()\n\n return [\n dict(zip(columns, row))\n for row in rows\n ]\n\n except pyodbc.Error:\n\n logging.exception(\n \"Azure SQL query failed.\"\n )\n\n raise\n\n finally:\n\n if connection:\n connection.close()<\/code><\/pre>\n\n\n\nTables\nViews\nStored procedures\nCustom SELECT statements<\/code><\/pre>\n\n\n\n
\n\n\n\n9. SQL views are ideal for this<\/h1>\n\n\n\n
CREATE VIEW x.vw_SearchCases\nAS\nSELECT\n c.CaseId,\n c.Title,\n c.Description,\n c.Status,\n c.CaseType,\n c.Classification,\n c.OwnerDepartment,\n c.CreatedAt,\n c.UpdatedAt\nFROM x.Cases c;<\/code><\/pre>\n\n\n\nsql = \"\"\"\nSELECT *\nFROM x.vw_SearchCases\nWHERE UpdatedAt >= DATEADD(\n MINUTE,\n ?,\n SYSUTCDATETIME()\n)\n\"\"\"<\/code><\/pre>\n\n\n\nlookback = -10<\/code><\/pre>\n\n\n\nx operational database\n \u2502\n \u25bc\n Search-specific views\n \u2502\n \u25bc\n Function App\n \u2502\n \u25bc\n AI Search<\/code><\/pre>\n\n\n\n
\n\n\n\n10. Create a generic SQL \u2192 Search transformer<\/h1>\n\n\n\n
def normalize_value(value: Any) -> Any:\n\n if value is None:\n return None\n\n if hasattr(value, \"isoformat\"):\n return value.isoformat()\n\n if isinstance(value, bytes):\n return value.decode(\n \"utf-8\",\n errors=\"replace\",\n )\n\n return value<\/code><\/pre>\n\n\n\ndef row_to_search_document(\n row: dict[str, Any],\n *,\n id_field: str,\n entity_type: str,\n content_fields: list[str],\n index_fields: dict[str, str] | None = None,\n) -> dict[str, Any]:\n\n index_fields = index_fields or {}\n\n document_id = str(\n row[id_field]\n )\n\n document = {\n \"id\": document_id,\n \"entity_type\": entity_type,\n }\n\n for sql_field, search_field in index_fields.items():\n\n if sql_field in row:\n\n document[search_field] = (\n normalize_value(\n row[sql_field]\n )\n )\n\n content_parts = []\n\n for field in content_fields:\n\n value = row.get(field)\n\n if value is not None:\n\n content_parts.append(\n f\"{field}: {value}\"\n )\n\n document[\"content\"] = (\n \"\\n\".join(content_parts)\n )\n\n return document<\/code><\/pre>\n\n\n\n
\n\n\n\n11. Why this is important<\/h1>\n\n\n\n
CaseId\nTitle\nDescription\nStatus\nClassification\nOwnerDepartment<\/code><\/pre>\n\n\n\n{\n \"id\": \"CASE-10025\",\n \"entity_type\": \"case\",\n \"title\": \"Market Manipulation Investigation\",\n \"content\": \"Title: Market Manipulation Investigation\\nDescription: ...\",\n \"status\": \"Open\",\n \"classification\": \"Restricted\",\n \"department\": \"Enforcement\"\n}<\/code><\/pre>\n\n\n\n
\n\n\n\n12. Batch upload to AI Search<\/h1>\n\n\n\n
def upload_search_documents(\n index_name: str,\n documents: list[dict[str, Any]],\n batch_size: int = 500,\n) -> dict[str, Any]:\n\n if not documents:\n\n return {\n \"submitted\": 0,\n \"succeeded\": 0,\n \"failed\": 0,\n \"errors\": [],\n }\n\n client = get_search_client(\n index_name\n )\n\n total_success = 0\n total_failed = 0\n errors = []\n\n for start in range(\n 0,\n len(documents),\n batch_size,\n ):\n\n batch = documents[\n start:start + batch_size\n ]\n\n try:\n\n results = (\n client.merge_or_upload_documents(\n documents=batch\n )\n )\n\n for result in results:\n\n if result.succeeded:\n\n total_success += 1\n\n else:\n\n total_failed += 1\n\n errors.append(\n {\n \"key\": result.key,\n \"error\": result.error_message,\n }\n )\n\n except Exception as exc:\n\n logging.exception(\n \"AI Search batch failed.\"\n )\n\n total_failed += len(batch)\n\n errors.append(\n {\n \"batch_start\": start,\n \"error\": str(exc),\n }\n )\n\n return {\n \"submitted\": len(documents),\n \"succeeded\": total_success,\n \"failed\": total_failed,\n \"errors\": errors[:100],\n }<\/code><\/pre>\n\n\n\nmerge_or_upload_documents<\/code> is particularly appropriate because it updates an existing document when its key exists and creates it when it doesn’t.<\/p>\n\n\n\n
\n\n\n\n13. Add a generic SQL table\/view exporter<\/h1>\n\n\n\n
def sync_sql_source_to_search(\n *,\n sql_query: str,\n parameters: tuple[Any, ...],\n index_name: str,\n id_field: str,\n entity_type: str,\n content_fields: list[str],\n index_fields: dict[str, str],\n) -> dict[str, Any]:\n\n rows = execute_sql_query(\n sql_query,\n parameters,\n )\n\n logging.info(\n \"SQL source returned %d rows for %s.\",\n len(rows),\n entity_type,\n )\n\n documents = []\n\n for row in rows:\n\n try:\n\n document = row_to_search_document(\n row,\n id_field=id_field,\n entity_type=entity_type,\n content_fields=content_fields,\n index_fields=index_fields,\n )\n\n documents.append(\n document\n )\n\n except Exception:\n\n logging.exception(\n \"Failed transforming %s row.\",\n entity_type,\n )\n\n return upload_search_documents(\n index_name=index_name,\n documents=documents,\n batch_size=int(\n os.environ.get(\n \"SQL_SYNC_BATCH_SIZE\",\n \"500\",\n )\n ),\n )<\/code><\/pre>\n\n\n\n
\n\n\n\n14. Now you can synchronize multiple SQL views<\/h1>\n\n\n\n
def sync_cases() -> dict[str, Any]:\n\n lookback = int(\n os.environ.get(\n \"SQL_SYNC_LOOKBACK_MINUTES\",\n \"10\",\n )\n )\n\n return sync_sql_source_to_search(\n\n sql_query=\"\"\"\n SELECT *\n FROM Techwyns.vw_SearchCases\n WHERE UpdatedAt >= DATEADD(\n MINUTE,\n ?,\n SYSUTCDATETIME()\n )\n \"\"\",\n\n parameters=(-lookback,),\n\n index_name=os.environ[\n \"AZURE_SEARCH_INDEX_NAME\"\n ],\n\n id_field=\"CaseId\",\n\n entity_type=\"case\",\n\n content_fields=[\n \"Title\",\n \"Description\",\n \"Status\",\n \"CaseType\",\n ],\n\n index_fields={\n \"CaseId\": \"case_id\",\n \"Title\": \"title\",\n \"Status\": \"status\",\n \"CaseType\": \"case_type\",\n \"Classification\":\n \"classification\",\n \"OwnerDepartment\":\n \"department\",\n \"UpdatedAt\":\n \"last_modified\",\n },\n )<\/code><\/pre>\n\n\n\n
\n\n\n\n15. Add evidence<\/h1>\n\n\n\n
def sync_evidence() -> dict[str, Any]:\n\n lookback = int(\n os.environ.get(\n \"SQL_SYNC_LOOKBACK_MINUTES\",\n \"10\",\n )\n )\n\n return sync_sql_source_to_search(\n\n sql_query=\"\"\"\n SELECT *\n FROM techwyns.vw_SearchEvidence\n WHERE UpdatedAt >= DATEADD(\n MINUTE,\n ?,\n SYSUTCDATETIME()\n )\n \"\"\",\n\n parameters=(-lookback,),\n\n index_name=os.environ[\n \"AZURE_SEARCH_INDEX_NAME\"\n ],\n\n id_field=\"EvidenceId\",\n\n entity_type=\"evidence\",\n\n content_fields=[\n \"Title\",\n \"Description\",\n \"EvidenceType\",\n ],\n\n index_fields={\n \"EvidenceId\":\n \"evidence_id\",\n\n \"CaseId\":\n \"case_id\",\n\n \"Title\":\n \"title\",\n\n \"Description\":\n \"description\",\n\n \"EvidenceType\":\n \"evidence_type\",\n\n \"Classification\":\n \"classification\",\n\n \"StorageUri\":\n \"source_uri\",\n\n \"UpdatedAt\":\n \"last_modified\",\n },\n )<\/code><\/pre>\n\n\n\n
\n\n\n\n16. Add regulations<\/h1>\n\n\n\n
def sync_regulations() -> dict[str, Any]:\n\n lookback = int(\n os.environ.get(\n \"SQL_SYNC_LOOKBACK_MINUTES\",\n \"10\",\n )\n )\n\n return sync_sql_source_to_search(\n\n sql_query=\"\"\"\n SELECT *\n FROM tchwyns.vw_SearchRegulations\n WHERE UpdatedAt >= DATEADD(\n MINUTE,\n ?,\n SYSUTCDATETIME()\n )\n \"\"\",\n\n parameters=(-lookback,),\n\n index_name=os.environ[\n \"AZURE_SEARCH_INDEX_NAME\"\n ],\n\n id_field=\"RegulationId\",\n\n entity_type=\"regulation\",\n\n content_fields=[\n \"Title\",\n \"Citation\",\n \"TextContent\",\n ],\n\n index_fields={\n \"RegulationId\":\n \"regulation_id\",\n\n \"Title\":\n \"title\",\n\n \"Citation\":\n \"citation\",\n\n \"EffectiveDate\":\n \"effective_date\",\n\n \"Classification\":\n \"classification\",\n\n \"SourceUri\":\n \"source_uri\",\n\n \"UpdatedAt\":\n \"last_modified\",\n },\n )<\/code><\/pre>\n\n\n\n
\n\n\n\n17. Synchronize everything<\/h1>\n\n\n\n
def synchronize_sql_to_search() -> dict[str, Any]:\n\n logging.info(\n \"Starting SQL \u2192 AI Search synchronization.\"\n )\n\n results = {}\n\n try:\n results[\"cases\"] = sync_cases()\n except Exception as exc:\n logging.exception(\n \"Case synchronization failed.\"\n )\n results[\"cases\"] = {\n \"error\": str(exc)\n }\n\n try:\n results[\"evidence\"] = sync_evidence()\n except Exception as exc:\n logging.exception(\n \"Evidence synchronization failed.\"\n )\n results[\"evidence\"] = {\n \"error\": str(exc)\n }\n\n try:\n results[\"regulations\"] = sync_regulations()\n except Exception as exc:\n logging.exception(\n \"Regulation synchronization failed.\"\n )\n results[\"regulations\"] = {\n \"error\": str(exc)\n }\n\n return results<\/code><\/pre>\n\n\n\nCases SUCCESS\nEvidence SUCCESS\nRegulations FAILED\nMarket SUCCESS<\/code><\/pre>\n\n\n\nCases SUCCESS\nEvidence SUCCESS\nRegulations FAILED\n \u2193\n EVERYTHING\n \u2193\n STOPS<\/code><\/pre>\n\n\n\n
\n\n\n\n18. Add the timer trigger<\/h1>\n\n\n\n
@app.timer_trigger(\n schedule=\"%SQL_SYNC_SCHEDULE%\",\n arg_name=\"timer\",\n run_on_startup=False,\n use_monitor=True,\n)\ndef sql_to_ai_search_sync(\n timer: func.TimerRequest,\n) -> None:\n\n logging.info(\n \"SQL \u2192 AI Search timer triggered.\"\n )\n\n if timer.past_due:\n\n logging.warning(\n \"SQL \u2192 AI Search timer is past due.\"\n )\n\n try:\n\n result = (\n synchronize_sql_to_search()\n )\n\n logging.info(\n \"SQL \u2192 AI Search synchronization \"\n \"completed: %s\",\n json.dumps(\n result,\n default=str,\n ),\n )\n\n except Exception:\n\n logging.exception(\n \"Fatal SQL \u2192 AI Search \"\n \"synchronization failure.\"\n )\n\n raise<\/code><\/pre>\n\n\n\nrun_on_startup=False<\/code> for production.<\/p>\n\n\n\n
\n\n\n\n19. Add a manual test endpoint<\/h1>\n\n\n\n
@app.route(\n route=\"admin\/sync-sql-search\",\n methods=[\"POST\"],\n auth_level=func.AuthLevel.FUNCTION,\n)\ndef manual_sql_search_sync(\n req: func.HttpRequest,\n) -> func.HttpResponse:\n\n logging.info(\n \"Manual SQL \u2192 Search sync requested.\"\n )\n\n try:\n\n result = (\n synchronize_sql_to_search()\n )\n\n return func.HttpResponse(\n json.dumps(\n result,\n indent=2,\n default=str,\n ),\n status_code=200,\n mimetype=\"application\/json\",\n )\n\n except Exception as exc:\n\n logging.exception(\n \"Manual synchronization failed.\"\n )\n\n return func.HttpResponse(\n json.dumps(\n {\n \"error\": str(exc),\n \"error_type\":\n type(exc).__name__,\n }\n ),\n status_code=500,\n mimetype=\"application\/json\",\n )<\/code><\/pre>\n\n\n\nFUNCTION<\/code> authorization rather than anonymous for this endpoint.<\/p>\n\n\n\napp = func.FunctionApp(\n http_auth_level=func.AuthLevel.ANONYMOUS\n)<\/code><\/pre>\n\n\n\n
\n\n\n\n20. Add health check<\/h1>\n\n\n\n
@app.route(\n route=\"health\",\n methods=[\"GET\"],\n auth_level=func.AuthLevel.ANONYMOUS,\n)\ndef health(\n req: func.HttpRequest,\n) -> func.HttpResponse:\n\n return func.HttpResponse(\n json.dumps(\n {\n \"status\": \"healthy\",\n \"service\":\n \"techwyns-function-app\",\n }\n ),\n status_code=200,\n mimetype=\"application\/json\",\n )<\/code><\/pre>\n\n\n\n
\n\n\n\n21. Add SQL connectivity test<\/h1>\n\n\n\n
@app.route(\n route=\"admin\/test-sql\",\n methods=[\"GET\"],\n auth_level=func.AuthLevel.FUNCTION,\n)\ndef test_sql(\n req: func.HttpRequest,\n) -> func.HttpResponse:\n\n connection = None\n\n try:\n\n connection = get_sql_connection()\n\n cursor = connection.cursor()\n\n cursor.execute(\n \"SELECT 1 AS HealthCheck\"\n )\n\n row = cursor.fetchone()\n\n return func.HttpResponse(\n json.dumps(\n {\n \"sql\":\n \"connected\",\n \"health_check\":\n row.HealthCheck,\n }\n ),\n status_code=200,\n mimetype=\"application\/json\",\n )\n\n except Exception as exc:\n\n logging.exception(\n \"SQL connectivity test failed.\"\n )\n\n return func.HttpResponse(\n json.dumps(\n {\n \"sql\":\n \"failed\",\n \"error\":\n str(exc),\n }\n ),\n status_code=500,\n mimetype=\"application\/json\",\n )\n\n finally:\n\n if connection:\n connection.close()<\/code><\/pre>\n\n\n\n
\n\n\n\n22. Add AI Search connectivity test<\/h1>\n\n\n\n
@app.route(\n route=\"admin\/test-search\",\n methods=[\"GET\"],\n auth_level=func.AuthLevel.FUNCTION,\n)\ndef test_search(\n req: func.HttpRequest,\n) -> func.HttpResponse:\n\n try:\n\n index_name = os.environ[\n \"AZURE_SEARCH_INDEX_NAME\"\n ]\n\n client = get_search_client(\n index_name\n )\n\n count = (\n client.get_document_count()\n )\n\n return func.HttpResponse(\n json.dumps(\n {\n \"search\":\n \"connected\",\n \"index\":\n index_name,\n \"document_count\":\n count,\n }\n ),\n status_code=200,\n mimetype=\"application\/json\",\n )\n\n except Exception as exc:\n\n logging.exception(\n \"AI Search connectivity test failed.\"\n )\n\n return func.HttpResponse(\n json.dumps(\n {\n \"search\":\n \"failed\",\n \"error\":\n str(exc),\n }\n ),\n status_code=500,\n mimetype=\"application\/json\",\n )<\/code><\/pre>\n\n\n\n
\n\n\n\n23. Test individual SQL views<\/h1>\n\n\n\n
def test_sql_sources() -> dict[str, Any]:\n\n tests = {\n\n \"cases\": \"\"\"\n SELECT TOP 1 *\n FROM techwyns.vw_SearchCases\n \"\"\",\n\n \"evidence\": \"\"\"\n SELECT TOP 1 *\n FROM techwyns.vw_SearchEvidence\n \"\"\",\n\n \"regulations\": \"\"\"\n SELECT TOP 1 *\n FROM techwyns.vw_SearchRegulations\n \"\"\",\n }\n\n results = {}\n\n for name, query in tests.items():\n\n try:\n\n rows = execute_sql_query(\n query\n )\n\n results[name] = {\n \"status\":\n \"success\",\n \"rows\":\n len(rows),\n }\n\n except Exception as exc:\n\n logging.exception(\n \"SQL source test failed: %s\",\n name,\n )\n\n results[name] = {\n \"status\":\n \"failed\",\n \"error\":\n str(exc),\n }\n\n return results<\/code><\/pre>\n\n\n\n
\n\n\n\n24. Testing strategy<\/h1>\n\n\n\n
Test 1 \u2014 SQL authentication<\/h3>\n\n\n\n
GET \/api\/admin\/test-sql<\/code><\/pre>\n\n\n\n{\n \"sql\": \"connected\",\n \"health_check\": 1\n}<\/code><\/pre>\n\n\n\n
\n\n\n\nTest 2 \u2014 AI Search<\/h3>\n\n\n\n
GET \/api\/admin\/test-search<\/code><\/pre>\n\n\n\n{\n \"search\": \"connected\",\n \"index\": \"csl-metadata\",\n \"document_count\": 1234\n}<\/code><\/pre>\n\n\n\n
\n\n\n\nTest 3 \u2014 SQL view<\/h3>\n\n\n\n
SELECT TOP 10 *\nFROM techwyns.vw_SearchCases;<\/code><\/pre>\n\n\n\n\n
\n\n\n\nTest 4 \u2014 transformation<\/h3>\n\n\n\n
SQL row\n \u2193\nSearch document<\/code><\/pre>\n\n\n\nid\nentity_type\ncontent<\/code><\/pre>\n\n\n\n
\n\n\n\nTest 5 \u2014 Search upload<\/h3>\n\n\n\n
POST \/api\/admin\/sync-sql-search<\/code><\/pre>\n\n\n\n{\n \"cases\": {\n \"submitted\": 10,\n \"succeeded\": 10,\n \"failed\": 0\n },\n \"evidence\": {\n \"submitted\": 25,\n \"succeeded\": 25,\n \"failed\": 0\n }\n}<\/code><\/pre>\n\n\n\n
\n\n\n\nTest 6 \u2014 Search verification<\/h3>\n\n\n\n
GET\/Search<\/code><\/pre>\n\n\n\n
\n\n\n\nTest 7 \u2014 update test<\/h3>\n\n\n\n
UPDATE techwyns.Cases\nSET\n Title = 'Updated Test Case',\n UpdatedAt = SYSUTCDATETIME()\nWHERE CaseId = 'TEST-001';<\/code><\/pre>\n\n\n\nTEST-001<\/code><\/pre>\n\n\n\nUpdated Test Case<\/code><\/pre>\n\n\n\n
\n\n\n\nTest 8 \u2014 failure test<\/h3>\n\n\n\n
FROM techwyns.vw_DoesNotExist<\/code><\/pre>\n\n\n\nfailed<\/code><\/pre>\n\n\n\n
\n\n\n\n25. One important production issue: deletes<\/h1>\n\n\n\n
INSERT\nUPDATE<\/code><\/pre>\n\n\n\nDELETE<\/code><\/pre>\n\n\n\nSQL:\nCASE-123 exists\n\nAI Search:\nCASE-123 exists\n\nSQL:\nCASE-123 deleted\n\nAI Search:\nCASE-123 STILL EXISTS<\/code><\/pre>\n\n\n\nIsDeleted BIT NOT NULL DEFAULT 0<\/code><\/pre>\n\n\n\n{\n \"id\": \"CASE-123\",\n \"@search.action\": \"delete\"\n}<\/code><\/pre>\n\n\n\nclient.delete_documents(\n documents=[\n {\"id\": \"CASE-123\"}\n ]\n)<\/code><\/pre>\n\n\n\n
\n\n\n\n26. Better: SQL Change Tracking<\/h1>\n\n\n\n
UpdatedAt >= DATEADD(...)<\/code><\/pre>\n\n\n\nSQL Change Tracking\n \u2502\n \u25bc\nFunction\n \u2502\n \u251c\u2500\u2500 INSERT\n \u251c\u2500\u2500 UPDATE\n \u2514\u2500\u2500 DELETE\n \u2502\n \u25bc\nAzure AI Search<\/code><\/pre>\n\n\n\nRun #100\n \u2193\nSQL changes 1000-1100\n \u2193\nFunction crashes\n \u2193\nRun #101\n \u2193\nreprocesses changes\n \u2193\nSearch catches up<\/code><\/pre>\n\n\n\n
\n\n\n\n27. Important issue with your MCP design<\/h1>\n\n\n\n
run_staff_letters_search()<\/code><\/pre>\n\n\n\n Function App\n \u2502\n \u250c\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u253c\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2510\n \u2502 \u2502 \u2502\n \u25bc \u25bc \u25bc\n MCP Search SQL Sync Blob\/PDF\n \u2502 \u2502\n \u25bc \u25bc\n AI Search AI Search<\/code><\/pre>\n\n\n\n
\n\n\n\n28. Recommended techwyns MCP tools after this change<\/h1>\n\n\n\n
semantic_search_staff_letters\nsearch__cases\nsearch__evidence\nsearch__regulations\nsearch__market_data\nget_staff_letter\nget_case\nget_evidence<\/code><\/pre>\n\n\n\n techwyns MCP\n \u2502\n \u250c\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u253c\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2510\n \u2502 \u2502 \u2502\n \u25bc \u25bc \u25bc\n Staff Letter techwyns SQL techwyns operational\n Search Search tools\n \u2502 \u2502 \u2502\n \u2514\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u252c\u2500\u2500\u2500\u2500\u2518 \u2502\n \u25bc \u25bc\n Azure AI Search Azure SQL\/API<\/code><\/pre>\n\n\n\n
\n\n\n\n29. Your Search indexes<\/h1>\n\n\n\n
Azure AI Search\n\u2502\n\u251c\u2500\u2500 staff-letters-new\n\u2502 \u2514\u2500\u2500 existing semantic search\n\u2502\n\u251c\u2500\u2500 csl-metadata\n\u2502 \u2514\u2500\u2500 existing metadata\n\u2502\n\u2514\u2500\u2500 techwyns-enterprise\n \u251c\u2500\u2500 cases\n \u251c\u2500\u2500 evidence\n \u251c\u2500\u2500 regulations\n \u2514\u2500\u2500 market observations<\/code><\/pre>\n\n\n\ncsl-metadata<\/code> index already has the correct schema, you can reuse it rather than create another index.<\/p>\n\n\n\n
\n\n\n\n30. Security metadata is especially important<\/h1>\n\n\n\n
classification\ndepartment\ncase_id\nsource_system\nsource_uri\nallowed_groups\nallowed_roles\nlast_modified<\/code><\/pre>\n\n\n\n{\n \"id\": \"case-1001\",\n \"entity_type\": \"case\",\n \"title\": \"Market Investigation\",\n \"content\": \"...\",\n \"case_id\": \"1001\",\n \"classification\": \"techwyns-RESTRICTED\",\n \"department\": \"Enforcement\",\n \"allowed_groups\": [\n \"techwyns-Enforcement\",\n \"techwyns-Senior-Investigators\"\n ],\n \"source_system\": \"techwyns-SQL\",\n \"source_uri\": \"techwyns:\/\/case\/1001\",\n \"last_modified\": \"2026-09-10T21:15:00Z\"\n}<\/code><\/pre>\n\n\n\n
\n\n\n\n31. Your current anonymous Function App<\/h1>\n\n\n\n
app = func.FunctionApp(\n http_auth_level=func.AuthLevel.ANONYMOUS\n)<\/code><\/pre>\n\n\n\nSQL\nAI Search\nMCP\ntechwyns data\nFoundry Agents<\/code><\/pre>\n\n\n\nFoundry \/ SimpleChat \/ Teams\n \u2502\n \u25bc\n APIM\n \u2502\n Entra validation\n \u2502\n \u25bc\n Private Function App\n \u2502\n \u250c\u2500\u2500\u2500\u2500\u2500\u2534\u2500\u2500\u2500\u2500\u2500\u2510\n \u25bc \u25bc\n SQL AI Search<\/code><\/pre>\n\n\n\n
\n\n\n\n32. One other correction to your current code<\/h1>\n\n\n\n
logging.info(\n \"MCP context type=%s repr=%s\",\n type(context).__name__,\n context\n)<\/code><\/pre>\n\n\n\nlogging.info(\n \"MCP context type=%s\",\n type(context).__name__,\n)<\/code><\/pre>\n\n\n\nlogging.info(\n \"Resolved MCP arguments: %s\",\n arguments\n)<\/code><\/pre>\n\n\n\ncorrelation_id\ntool_name\nagent_id\nduration\nresult_count\nsuccess\nerror_type\nauthorization_decision<\/code><\/pre>\n\n\n\n
\n\n\n\n33. I would also improve your existing exception handling<\/h1>\n\n\n\n
except Exception as exc:\n logging.exception(\"Semantic search failed\")\n return json.dumps({\n \"error\": str(exc),\n \"error_type\": type(exc).__name__\n })<\/code><\/pre>\n\n\n\nexcept ValueError as exc:\n\n logging.warning(\n \"Invalid semantic search request: %s\",\n exc,\n )\n\n return json.dumps({\n \"error\": \"Invalid request\",\n \"error_type\": \"validation_error\",\n })\n\nexcept Exception:\n\n logging.exception(\n \"Semantic search failed.\"\n )\n\n return json.dumps({\n \"error\":\n \"Internal search service error\",\n \"error_type\":\n \"internal_error\",\n })<\/code><\/pre>\n\n\n\n
\n\n\n\n34. Final production flow<\/h1>\n\n\n\n
\u250c\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2510\n \u2502 Azure SQL \u2502\n \u2502 \u2502\n \u2502 Tables \u2502\n \u2502 Views \u2502\n \u2502 Change Tracking \u2502\n \u2514\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u252c\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2518\n \u2502\n \u2502 Managed Identity\n \u25bc\n \u250c\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2510\n \u2502 techwyns Function App \u2502\n \u2502 \u2502\n \u2502 SQL extraction \u2502\n \u2502 Validation \u2502\n \u2502 Transformation \u2502\n \u2502 Classification \u2502\n \u2502 Security metadata \u2502\n \u2502 Batch processing \u2502\n \u2502 Error handling \u2502\n \u2502 App Insights \u2502\n \u2514\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u252c\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2518\n \u2502\n \u2502 Entra\/RBAC\n \u25bc\n \u250c\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2510\n \u2502 Azure AI Search \u2502\n \u2502 \u2502\n \u2502 csl-metadata \u2502\n \u2502 staff-letters-new \u2502\n \u2502 techwyns-enterprise \u2502\n \u2514\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u252c\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2518\n \u2502\n \u25bc\n \u250c\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2510\n \u2502 techwyns MCP \u2502\n \u2502 \u2502\n \u2502 Search tools \u2502\n \u2502 Case tools \u2502\n \u2502 Evidence \u2502\n \u2502 Regulatory \u2502\n \u2514\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u252c\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2518\n \u2502\n \u25bc\n \u250c\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2510\n \u2502 Microsoft Foundry \u2502\n \u2502 Agent Service \u2502\n \u2502 \u2502\n \u2502 techwyns Orchestrator \u2502\n \u2502 Research Agent \u2502\n \u2502 Investigation Agent \u2502\n \u2502 Regulatory Agent \u2502\n \u2514\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2518<\/code><\/pre>\n\n\n\nMy strongest recommendation<\/h2>\n\n\n\n
UpdatedAt<\/code> synchronization + merge_or_upload_documents<\/code> + timer trigger + managed identity<\/strong>. Then, once it works end-to-end, upgrade the synchronization mechanism to SQL Change Tracking\/rowversion + durable sync state + delete detection<\/strong>.<\/p>\n\n\n\nfunction_app.py<\/code><\/strong> you can deploy as the replacement for the file you supplied.<\/p>\n\n\n\n\n
\n
merge_or_upload_documents<\/code><\/li>\n\n\n\nIsDeleted<\/code><\/li>\n\n\n\nAZURE_SEARCH_INDEX_NAE<\/code> typo<\/li>\n\n\n\nDefaultAzureCredential<\/code>\/Entra authentication and merge_or_upload_documents<\/code>; the identity needs Search Index Data Contributor<\/strong> for indexing and Search Index Data Reader<\/strong> for querying.<\/p>\n\n\n\n
\n\n\n\nComplete
function_app.py<\/code><\/h1>\n\n\n\n\n\"\"\"\ntechwyns Azure Function App\n=======================\n\nCapabilities\n------------\n1. Azure AI Search metadata search\n2. Semantic search over techwyns staff letters\n3. Staff-letter PDF retrieval from Azure Blob Storage\n4. MCP tools for Foundry Agent Service\n5. Azure SQL -> Azure AI Search synchronization\n6. SQL table\/view synchronization\n7. Incremental synchronization using UpdatedAt lookback\n8. Optional SQL IsDeleted -> AI Search delete handling\n9. Batch merge-or-upload into Azure AI Search\n10. SQL \/ AI Search health checks\n11. Manual synchronization endpoint\n12. Source-level synchronization testing\n13. Exception isolation and retry handling\n\nAuthentication\n--------------\nAzure SQL:\n Managed Identity \/ DefaultAzureCredential\n\nAzure AI Search:\n Managed Identity \/ DefaultAzureCredential\n\nBlob Storage:\n Managed Identity \/ DefaultAzureCredential\n or optional connection string\n\nIMPORTANT\n---------\nSQL synchronization sources are configured through:\n\n SQL_SYNC_SOURCES_JSON\n\nExample:\n\n[\n {\n \"name\": \"cases\",\n \"query\": \"SELECT CaseId, Title, Description, Status, CaseType, Classification, OwnerDepartment, UpdatedAt, IsDeleted FROM techwyns.vw_SearchCases WHERE UpdatedAt >= DATEADD(MINUTE, ?, SYSUTCDATETIME())\",\n \"id_field\": \"CaseId\",\n \"entity_type\": \"case\",\n \"index_name\": \"techwyns-enterprise\",\n \"content_fields\": [\n \"Title\",\n \"Description\",\n \"Status\",\n \"CaseType\"\n ],\n \"field_map\": {\n \"CaseId\": \"case_id\",\n \"Title\": \"title\",\n \"Status\": \"status\",\n \"CaseType\": \"case_type\",\n \"Classification\": \"classification\",\n \"OwnerDepartment\": \"department\",\n \"UpdatedAt\": \"last_modified\"\n },\n \"deleted_field\": \"IsDeleted\"\n }\n]\n\nThe SQL query is application-controlled configuration.\nThe MCP\/HTTP caller cannot submit arbitrary SQL.\n\"\"\"<\/code><\/pre>\n\n\n\n1.
requirements.txt<\/code><\/h2>\n\n\n\n