MCP-based Azure Function App for SQL Data indexing

visual architecture package so you can see the design, flow, context, container, and class diagrams for your real-time + scheduled SQL → Azure AI Search knowledge base with UI and MCP/Foundry agent integration.

1. System Design Overview

Goal:
A self-updating knowledge base that:

Pulls from Azure SQL (scheduled + real-time CDC/Event Grid).
Indexes into Azure AI Search.
Serves results to UI and MCP agent.
Tracks last sync in Blob Storage.
Supports manual refresh.

2. High-Level Flow Diagram

Copy code+-------------------+       +-------------------+       +-------------------+
| Azure SQL DB      |  -->  | CDC / Change Track |  -->  | Event Grid Topic  |
+-------------------+       +-------------------+       +-------------------+
       |                              |                           |
       | (Scheduled Query)            | (Real-time Event)        |
       v                              v                           v
+-------------------+       +-------------------+       +-------------------+
| Timer Trigger Fn  |       | Event Grid Fn     |       | Blob Storage      |
| (Incremental Sync)|       | (Immediate Sync)  |       | (Last Sync Time)  |
+-------------------+       +-------------------+       +-------------------+
       \_________________________   ___________________________/
                                     |
                                     v
                           +-------------------+
                           | Azure AI Search   |
                           +-------------------+
                                     |
                   +----------------+----------------+
                   |                                 |
          +-------------------+             +-------------------+
          | UI Frontend       |             | MCP/Foundry Agent |
          +-------------------+             +-------------------+

3. Context Diagram (C4 Level 1)

System Context:

Copy code[User] ---> [Knowledge Base System] ---> [Azure AI Search]
   ^                ^                          ^
   |                |                          |
   |                |                          |
[UI Frontend]   [MCP Agent]               [Azure SQL DB]

Key External Actors:

User: Queries knowledge base via UI or MCP agent.
Azure SQL DB: Source of truth for documents.
Azure AI Search: Search index for fast retrieval.

4. Container Diagram (C4 Level 2)

Copy code+--------------------------------------------------------------+
| Knowledge Base System                                        |
|                                                              |
|  +-------------------+     +-------------------+             |
|  | Timer Trigger Fn  |     | Event Grid Fn     |             |
|  | (Incremental Sync)|     | (Real-time Sync)  |             |
|  +-------------------+     +-------------------+             |
|           |                         |                        |
|           v                         v                        |
|     +-------------------+     +-------------------+          |
|     | Index Manager     |<----| Change Processor  |          |
|     +-------------------+     +-------------------+          |
|           |                         |                        |
|           v                         v                        |
|     +-------------------+     +-------------------+          |
|     | Azure AI Search   |     | Blob Storage      |          |
|     +-------------------+     +-------------------+          |
|                                                              |
+--------------------------------------------------------------+

5. Class Diagram (Core Components)

PlaintextCopy code+-------------------+
| IndexManager      |
+-------------------+
| - searchEndpoint  |
| - searchApiKey    |
| - indexName       |
+-------------------+
| + pushDocs(docs)  |
| + deleteDocs(ids) |
+-------------------+

+-------------------+
| SqlConnector      |
+-------------------+
| - connString      |
+-------------------+
| + getAllIds()     |
| + getChanges(ts)  |
| + getCDCChanges() |
+-------------------+

+-------------------+
| BlobStateManager  |
+-------------------+
| - blobConnStr     |
| - containerName   |
| - blobName        |
+-------------------+
| + getLastSync()   |
| + setLastSync(ts) |
+-------------------+

+-------------------+
| EventProcessor    |
+-------------------+
| + handleEvent(e)  |
| + mapToAction()   |
+-------------------+

6. Sequence Flow (Real-Time Event)

Copy codeUser updates SQL row
    |
    v
Azure SQL CDC logs change
    |
    v
Azure Data Factory/Synapse streams change to Event Grid
    |
    v
Event Grid triggers Azure Function

proceed and complete the design package by adding the sequence flows, deployment view, and final integration notes so you have a full blueprint you can hand to your dev + ops teams.

7. Sequence Flow — Scheduled Sync

PlaintextCopy code[Timer Trigger Function] → getLastSync() from BlobStateManager
    → SqlConnector.getAllIds()
    → IndexManager.getAllIndexIds()
    → Compare sets → deleteDocs(staleIds)
    → SqlConnector.getChanges(lastSync)
    → IndexManager.pushDocs(newDocs)
    → BlobStateManager.setLastSync(now)

Key Points:

Runs on CRON schedule (e.g., hourly).
Deletes stale docs first, then indexes new/updated ones.
Updates last sync timestamp in Blob Storage.

8. Sequence Flow — Real-Time Event

PlaintextCopy code[User/Process updates SQL row]
    ↓
Azure SQL CDC logs change
    ↓
Azure Data Factory/Synapse Link streams change
    ↓
Event Grid publishes event
    ↓
Event Grid Trigger Function receives event
    ↓
EventProcessor.handleEvent()
    → If delete → IndexManager.deleteDocs([id])
    → If insert/update → IndexManager.pushDocs([doc])

Key Points:

Latency is seconds, not minutes/hours.
Only processes changed rows.
Works alongside scheduled sync as a safety net.

9. Deployment View (C4 Level 3)

Copy code+---------------------------+       +---------------------------+
| Azure Function App        |       | Azure SQL Database        |
|  - Timer Trigger Fn       |<----->| CDC Enabled               |
|  - Event Grid Trigger Fn  |       | Change Tracking Enabled   |
|  - HTTP API Fn (UI/MCP)   |       +---------------------------+
|                           |
|  +---------------------+  |       +---------------------------+
|  | Blob Storage        |  |<----->| Azure Event Grid          |
|  | (Last Sync State)   |  |       +---------------------------+
|  +---------------------+  |
|                           |       +---------------------------+
|  +---------------------+  |<----->| Azure AI Search           |
|  | IndexManager        |  |       +---------------------------+
|  +---------------------+  |
+---------------------------+

10. Component Responsibilities

Component	Responsibility
SqlConnector	Connect to SQL, fetch IDs, fetch changes, fetch CDC changes.
IndexManager	Push, merge, or delete documents in Azure AI Search.
BlobStateManager	Store/retrieve last sync timestamp.
EventProcessor	Interpret Event Grid payloads and map to index actions.
UI Frontend	Display last sync time, doc count, allow manual refresh.
MCP Agent	Query AI Search, check freshness, optionally trigger refresh.

11. Final Integration Notes

Security:
- Use Managed Identity for SQL, Blob, and AI Search.
- Store API keys in Azure Key Vault.
Resilience:
- Scheduled sync acts as a catch-up if real-time events are missed.
- Blob state ensures incremental indexing is consistent.
Scalability:
- Event Grid scales automatically with change volume.
- AI Search supports batch indexing for efficiency.
Transparency:
- UI + MCP agent both use the same GetIndexStatus endpoint.
- Users can see freshness and trigger manual refresh.

✅ Now you have:

Context Diagram (C4 Level 1)
Container Diagram (C4 Level 2)
Class Diagram
Sequence Flows (Scheduled + Real-Time)
Deployment View
Component Responsibilities

–

proceed and turn the architecture we’ve been discussing into graphical UML-style diagrams you can drop directly into your documentation, Confluence, or design reviews.

Here are the PlantUML and Mermaid definitions — you can paste these into any UML tool (PlantUML, Mermaid Live Editor, Lucidchart, Draw.io, etc.) to instantly generate the images.

1. Context Diagram (C4 Level 1) — Mermaid

graph TD
    User[User] --> UI[UI Frontend]
    User --> MCP[MCP/Foundry Agent]
    UI --> KB[Knowledge Base System]
    MCP --> KB
    KB --> SQL[Azure SQL Database]
    KB --> AIS[Azure AI Search]

2. Container Diagram (C4 Level 2) — Mermaid

Mermaid
 graph TD
    subgraph KB[Knowledge Base System]
        TimerFn[Timer Trigger Function] --> IndexMgr[Index Manager]
        EventFn[Event Grid Trigger Function] --> ChangeProc[Change Processor]
        IndexMgr --> AIS[Azure AI Search]
        ChangeProc --> AIS
        TimerFn --> Blob[Blob Storage - Last Sync]
        EventFn --> Blob
    end

    SQL[Azure SQL Database] --> TimerFn
    SQL --> EventFn
    AIS --> UI[UI Frontend]
    AIS --> MCP[MCP Agent]

3. Class Diagram — PlantUML

Plantuml 

@startuml
class IndexManager {
    - searchEndpoint : string
    - searchApiKey : string
    - indexName : string
    + pushDocs(docs)
    + deleteDocs(ids)
}

class SqlConnector {
    - connString : string
    + getAllIds()
    + getChanges(ts)
    + getCDCChanges()
}

class BlobStateManager {
    - blobConnStr : string
    - containerName : string
    - blobName : string
    + getLastSync()
    + setLastSync(ts)
}

class EventProcessor {
    + handleEvent(e)
    + mapToAction()
}

IndexManager --> SqlConnector
IndexManager --> BlobStateManager
EventProcessor --> IndexManager
@enduml

4. Sequence Diagram — Scheduled Sync — PlantUML

Plantuml

@startuml
actor Timer
participant TimerTriggerFn
participant BlobStateManager
participant SqlConnector
participant IndexManager
participant AzureAISearch

Timer -> TimerTriggerFn: Trigger
TimerTriggerFn -> BlobStateManager: getLastSync()
BlobStateManager --> TimerTriggerFn: lastSync
TimerTriggerFn -> SqlConnector: getAllIds()
SqlConnector --> TimerTriggerFn: sqlIds
TimerTriggerFn -> IndexManager: getAllIndexIds()
IndexManager --> TimerTriggerFn: aiIds
TimerTriggerFn -> IndexManager: deleteDocs(staleIds)
TimerTriggerFn -> SqlConnector: getChanges(lastSync)
SqlConnector --> TimerTriggerFn: newDocs
TimerTriggerFn -> IndexManager: pushDocs(newDocs)
TimerTriggerFn -> BlobStateManager: setLastSync(now)
@enduml

5. Sequence Diagram — Real-Time Event — PlantUML

Plantuml

@startuml
actor SQLUser
participant AzureSQL
participant EventGrid
participant EventGridFn
participant EventProcessor
participant IndexManager
participant AzureAISearch

SQLUser -> AzureSQL: Insert/Update/Delete row
AzureSQL -> EventGrid: CDC/Change Tracking Event
EventGrid -> EventGridFn: Trigger
EventGridFn -> EventProcessor: handleEvent(event)
EventProcessor -> IndexManager: pushDocs() / deleteDocs()
IndexManager -> AzureAISearch: Update Index
@enduml

6. Deployment Diagram — PlantUML

Plantuml

@startuml
node "Azure Cloud" {
    node "Function App" {
        component "Timer Trigger Fn"
        component "Event Grid Trigger Fn"
        component "HTTP API Fn"
    }
    node "Azure SQL Database" {
        artifact "KnowledgeTable"
        artifact "CDC Change Tables"
    }
    node "Azure Blob Storage" {
        artifact "last_sync.txt"
    }
    node "Azure AI Search" {
        artifact "knowledge-index"
    }
    node "Azure Event Grid" {
        artifact "Change Events"
    }
    node "UI Frontend"
    node "MCP Agent"
}
@enduml

How to Generate the Images

Mermaid:
- Go to Mermaid Live Editor, paste

===MAIN IMPLEMENTATION STARTS HERE ====

extend your existing Azure Functions code so that it:

Connects to a SQL database
Ingests data into an Azure AI Search index
Updates MCP so the data is searchable and retrievable
Keeps your semantic search logic intact

I’ll merge your provided snippet into a full Azure Functions app using func.FunctionApp with anonymous HTTP auth for testing.

Updated Code — `function_app.py`

proceed and finish the updated Azure Functions code so it’s complete, deployable, and ready to:

Pull data from SQL
Ingest into Azure AI Search
Expose endpoints for ingestion and semantic search
Be callable from MCP or any HTTP client

We stopped at the ingest_sql_to_search function, so let’s complete that and add the semantic search endpoint.

Completed `function_app.py`

PythonCopy codeimport os
import json
import pyodbc
import azure.functions as func
from typing import Any
from datetime import datetime
from azure.identity import DefaultAzureCredential
from azure.search.documents import SearchClient
from azure.search.documents.models import IndexDocumentsBatch, IndexAction

# ── Azure Function App ───────────────────────────────────────────────
app = func.FunctionApp(http_auth_level=func.AuthLevel.ANONYMOUS)

# ── SQL Connection ───────────────────────────────────────────────────
def get_sql_connection():
    conn_str = os.environ["SQL_CONNECTION_STRING"]
    return pyodbc.connect(conn_str)

def fetch_sql_data():
    """Fetch all rows from SQL table for indexing."""
    with get_sql_connection() as conn:
        cursor = conn.cursor()
        cursor.execute("SELECT id, title, content, lastModified FROM Documents")
        rows = cursor.fetchall()
        docs = []
        for row in rows:
            docs.append({
                "id": str(row.id),
                "title": row.title,
                "content": row.content,
                "lastModified": row.lastModified.isoformat() if row.lastModified else None
            })
        return docs

# ── Azure Search Clients ─────────────────────────────────────────────
def get_search_client() -> SearchClient:
    return SearchClient(
        endpoint=os.environ["AZURE_SEARCH_ENDPOINT"],
        index_name=os.environ.get("AZURE_SEARCH_INDEX_NAME", "csl-metadata"),
        credential=DefaultAzureCredential(),
    )

SEMANTIC_INDEX_NAME = os.environ.get("AZURE_SEARCH_SEMANTIC_INDEX_NAME", "staff-letters-new")
SEMANTIC_CONFIGURATION_NAME = os.environ.get(
    "AZURE_SEARCH_SEMANTIC_CONFIGURATION_NAME", "staff-letters-new-semantic-configuration"
)

def get_semantic_search_client() -> SearchClient:
    return SearchClient(
        endpoint=os.environ["AZURE_SEARCH_ENDPOINT"],
        index_name=SEMANTIC_INDEX_NAME,
        credential=DefaultAzureCredential(),
    )

# ── Ingest Data into AI Search ────────────────────────────────────────
def ingest_data_to_search(docs: list[dict]):
    if not docs:
        return {"status": "no documents to index"}

    client = get_search_client()
    batch = IndexDocumentsBatch(actions=[
        IndexAction.merge_or_upload(doc) for doc in docs
    ])
    result = client.index_documents(batch=batch)
    return {"status": "indexed", "count": len(docs), "result": str(result)}

# ── Semantic Search ──────────────────────────────────────────────────
def run_semantic_staff_letters_search(args: dict[str, Any]) -> dict[str, Any]:
    query = args.get("query")
    if not query or not str(query).strip():
        raise ValueError("query is required")

    top: int = int(args.get("top", 50))
    skip: int = int(args.get("skip", 0))
    semantic_configuration: str = args.get("semantic_configuration_name", SEMANTIC_CONFIGURATION_NAME)

    if top < 1:
        raise ValueError("top must be greater than 0")
    if skip < 0:
        raise ValueError("skip must be 0 or greater")

    results = get_semantic_search_client().search(
        search_text=query,
        query_type="semantic",
        semantic_configuration_name=semantic_configuration,
        query_caption="extractive",
        query_answer="extractive",
        top=top,
        skip=skip,
        include_total_count=True,
    )

    docs = []
    for result in results:
        doc = dict(result)
        doc.pop("text_vector", None)
        captions = doc.pop("@search.captions", None)
        if captions:
            doc["captions"] = [c["text"] for c in captions]
        docs.append(doc)

    return {"count": results.get_count(), "results": docs}

# ── HTTP Endpoints ───────────────────────────────────────────────────
@app.route(route="ingest-sql-to-search", methods=["POST"])
def ingest_sql_to_search(req: func.HttpRequest) -> func.HttpResponse:
    """Fetch data from SQL and ingest into Azure Search."""
    try:
        docs = fetch_sql_data()
        result = ingest_data_to_search(docs)
        return func.HttpResponse(
            json.dumps(result),

proceed and finish the last part of the code so your Azure Function app is fully functional with both ingestion and semantic search endpoints.

We stopped inside the ingest_sql_to_search function, so now we’ll complete it and add the semantic search HTTP route.

Final Completed `function_app.py`

PythonCopy codeimport os
import json
import pyodbc
import azure.functions as func
from typing import Any
from datetime import datetime
from azure.identity import DefaultAzureCredential
from azure.search.documents import SearchClient
from azure.search.documents.models import IndexDocumentsBatch, IndexAction

# ── Azure Function App ───────────────────────────────────────────────
app = func.FunctionApp(http_auth_level=func.AuthLevel.ANONYMOUS)

# ── SQL Connection ───────────────────────────────────────────────────
def get_sql_connection():
    conn_str = os.environ["SQL_CONNECTION_STRING"]
    return pyodbc.connect(conn_str)

def fetch_sql_data():
    """Fetch all rows from SQL table for indexing."""
    with get_sql_connection() as conn:
        cursor = conn.cursor()
        cursor.execute("SELECT id, title, content, lastModified FROM Documents")
        rows = cursor.fetchall()
        docs = []
        for row in rows:
            docs.append({
                "id": str(row.id),
                "title": row.title,
                "content": row.content,
                "lastModified": row.lastModified.isoformat() if row.lastModified else None
            })
        return docs

# ── Azure Search Clients ─────────────────────────────────────────────
def get_search_client() -> SearchClient:
    return SearchClient(
        endpoint=os.environ["AZURE_SEARCH_ENDPOINT"],
        index_name=os.environ.get("AZURE_SEARCH_INDEX_NAME", "csl-metadata"),
        credential=DefaultAzureCredential(),
    )

SEMANTIC_INDEX_NAME = os.environ.get("AZURE_SEARCH_SEMANTIC_INDEX_NAME", "staff-letters-new")
SEMANTIC_CONFIGURATION_NAME = os.environ.get(
    "AZURE_SEARCH_SEMANTIC_CONFIGURATION_NAME", "staff-letters-new-semantic-configuration"
)

def get_semantic_search_client() -> SearchClient:
    return SearchClient(
        endpoint=os.environ["AZURE_SEARCH_ENDPOINT"],
        index_name=SEMANTIC_INDEX_NAME,
        credential=DefaultAzureCredential(),
    )

# ── Ingest Data into AI Search ────────────────────────────────────────
def ingest_data_to_search(docs: list[dict]):
    if not docs:
        return {"status": "no documents to index"}

    client = get_search_client()
    batch = IndexDocumentsBatch(actions=[
        IndexAction.merge_or_upload(doc) for doc in docs
    ])
    result = client.index_documents(batch=batch)
    return {"status": "indexed", "count": len(docs), "result": str(result)}

# ── Semantic Search ──────────────────────────────────────────────────
def run_semantic_staff_letters_search(args: dict[str, Any]) -> dict[str, Any]:
    query = args.get("query")
    if not query or not str(query).strip():
        raise ValueError("query is required")

    top: int = int(args.get("top", 50))
    skip: int = int(args.get("skip", 0))
    semantic_configuration: str = args.get("semantic_configuration_name", SEMANTIC_CONFIGURATION_NAME)

    if top < 1:
        raise ValueError("top must be greater than 0")
    if skip < 0:
        raise ValueError("skip must be 0 or greater")

    results = get_semantic_search_client().search(
        search_text=query,
        query_type="semantic",
        semantic_configuration_name=semantic_configuration,
        query_caption="extractive",
        query_answer="extractive",
        top=top,
        skip=skip,
        include_total_count=True,
    )

    docs = []
    for result in results:
        doc = dict(result)
        doc.pop("text_vector", None)
        captions = doc.pop("@search.captions", None)
        if captions:
            doc["captions"] = [c["text"] for c in captions]
        docs.append(doc)

    return {"count": results.get_count(), "results": docs}

# ── HTTP Endpoints ───────────────────────────────────────────────────
@app.route(route="ingest-sql-to-search", methods=["POST"])
def ingest_sql_to_search(req: func.HttpRequest) -> func.HttpResponse:
    """Fetch data from SQL and ingest into Azure Search."""
    try:
        docs = fetch_sql_data()
        result = ingest_data_to_search(docs)
        return func.HttpResponse(
            json.dumps(result),
            mimetype="application/json",
            status_code=200
        )
    except Exception as e:
        return func.HttpResponse(
            json

proceed and finish the last part of the Azure Function so it’s fully deployable and includes the semantic search endpoint.

We stopped inside the ingest_sql_to_search exception handler, so now we’ll complete that and add the semantic-search route.

Final Completed `function_app.py`

PythonCopy codeimport os
import json
import pyodbc
import azure.functions as func
from typing import Any
from datetime import datetime
from azure.identity import DefaultAzureCredential
from azure.search.documents import SearchClient
from azure.search.documents.models import IndexDocumentsBatch, IndexAction

# ── Azure Function App ───────────────────────────────────────────────
app = func.FunctionApp(http_auth_level=func.AuthLevel.ANONYMOUS)

# ── SQL Connection ───────────────────────────────────────────────────
def get_sql_connection():
    conn_str = os.environ["SQL_CONNECTION_STRING"]
    return pyodbc.connect(conn_str)

def fetch_sql_data():
    """Fetch all rows from SQL table for indexing."""
    with get_sql_connection() as conn:
        cursor = conn.cursor()
        cursor.execute("SELECT id, title, content, lastModified FROM Documents")
        rows = cursor.fetchall()
        docs = []
        for row in rows:
            docs.append({
                "id": str(row.id),
                "title": row.title,
                "content": row.content,
                "lastModified": row.lastModified.isoformat() if row.lastModified else None
            })
        return docs

# ── Azure Search Clients ─────────────────────────────────────────────
def get_search_client() -> SearchClient:
    return SearchClient(
        endpoint=os.environ["AZURE_SEARCH_ENDPOINT"],
        index_name=os.environ.get("AZURE_SEARCH_INDEX_NAME", "csl-metadata"),
        credential=DefaultAzureCredential(),
    )

SEMANTIC_INDEX_NAME = os.environ.get("AZURE_SEARCH_SEMANTIC_INDEX_NAME", "staff-letters-new")
SEMANTIC_CONFIGURATION_NAME = os.environ.get(
    "AZURE_SEARCH_SEMANTIC_CONFIGURATION_NAME", "staff-letters-new-semantic-configuration"
)

def get_semantic_search_client() -> SearchClient:
    return SearchClient(
        endpoint=os.environ["AZURE_SEARCH_ENDPOINT"],
        index_name=SEMANTIC_INDEX_NAME,
        credential=DefaultAzureCredential(),
    )

# ── Ingest Data into AI Search ────────────────────────────────────────
def ingest_data_to_search(docs: list[dict]):
    if not docs:
        return {"status": "no documents to index"}

    client = get_search_client()
    batch = IndexDocumentsBatch(actions=[
        IndexAction.merge_or_upload(doc) for doc in docs
    ])
    result = client.index_documents(batch=batch)
    return {"status": "indexed", "count": len(docs), "result": str(result)}

# ── Semantic Search ──────────────────────────────────────────────────
def run_semantic_staff_letters_search(args: dict[str, Any]) -> dict[str, Any]:
    query = args.get("query")
    if not query or not str(query).strip():
        raise ValueError("query is required")

    top: int = int(args.get("top", 50))
    skip: int = int(args.get("skip", 0))
    semantic_configuration: str = args.get("semantic_configuration_name", SEMANTIC_CONFIGURATION_NAME)

    if top < 1:
        raise ValueError("top must be greater than 0")
    if skip < 0:
        raise ValueError("skip must be 0 or greater")

    results = get_semantic_search_client().search(
        search_text=query,
        query_type="semantic",
        semantic_configuration_name=semantic_configuration,
        query_caption="extractive",
        query_answer="extractive",
        top=top,
        skip=skip,
        include_total_count=True,
    )

    docs = []
    for result in results:
        doc = dict(result)
        doc.pop("text_vector", None)
        captions = doc.pop("@search.captions", None)
        if captions:
            doc["captions"] = [c["text"] for c in captions]
        docs.append(doc)

    return {"count": results.get_count(), "results": docs}

# ── HTTP Endpoints ───────────────────────────────────────────────────
@app.route(route="ingest-sql-to-search", methods=["POST"])
def ingest_sql_to_search(req: func.HttpRequest) -> func.HttpResponse:
    """Fetch data from SQL and ingest into Azure Search."""
    try:
        docs = fetch_sql_data()
        result = ingest_data_to_search(docs)
        return func.HttpResponse(
            json.dumps(result),
            mimetype="application/json",
            status_code=200
        )
    except Exception as e:
        return func.HttpResponse(
            json

SECTION 2: complete, step-by-step guide plus full code for an MCP-based Azure Function App that:

Connects to Azure AI Search (formerly Cognitive Search)
Uses knowledge sources (SQL Database in this case)
Indexes the data into a knowledge base
Makes it accessible to a UI frontend via an API endpoint

I’ll break this into phases so you can follow along from setup to deployment.

PHASE 1 — Architecture Overview

Flow:

Azure SQL Database → Data source
Azure Function App (MCP Tool) → Extracts data from SQL, transforms it, and pushes it to Azure AI Search
Azure AI Search Index → Stores searchable knowledge base
UI Frontend → Calls the Function App to query the AI Search index

PHASE 2 — Prerequisites

Azure Subscription
Azure SQL Database with your data
Azure AI Search service
Azure Function App (Python or Node.js — here we’ll use Python)
Azure Storage Account (for Function App)
MCP Tool integration in Foundry Agent (optional for automation)

PHASE 3 — Azure AI Search Setup

Create Azure AI Search service in the portal.
Create an index (e.g., knowledge-index) with fields:
- id (key)
- title (searchable)
- content (searchable, retrievable)
- lastModified (filterable, sortable)
Get API Key and Endpoint URL from Azure AI Search portal.

PHASE 4 — Azure SQL Database Setup

Ensure your SQL DB has a table with at least:
- Id (primary key)
- Title
- Content
- LastModified
Allow Azure services to access the DB.
Create a read-only SQL user for the Function App.

PHASE 5 — Azure Function App MCP Tool

We’ll create a Python Azure Function that:

Connects to SQL DB
Reads new/updated rows
Pushes them to Azure AI Search index

`function_app/init.py`

Python

import logging
import json
import pyodbc
import azure.functions as func
import requests
from datetime import datetime

# Azure AI Search config
SEARCH_ENDPOINT = "https://<YOUR-SEARCH-NAME>.search.windows.net"
SEARCH_API_KEY = "<YOUR-SEARCH-ADMIN-KEY>"
SEARCH_INDEX = "knowledge-index"

# SQL config
SQL_SERVER = "<YOUR-SQL-SERVER>.database.windows.net"
SQL_DATABASE = "<YOUR-DATABASE>"
SQL_USERNAME = "<YOUR-USERNAME>"
SQL_PASSWORD = "<YOUR-PASSWORD>"
SQL_DRIVER = "{ODBC Driver 18 for SQL Server}"

def get_sql_data(last_sync=None):
    conn_str = f"DRIVER={SQL_DRIVER};SERVER={SQL_SERVER};DATABASE={SQL_DATABASE};UID={SQL_USERNAME};PWD={SQL_PASSWORD}"
    conn = pyodbc.connect(conn_str)
    cursor = conn.cursor()

    if last_sync:
        cursor.execute("SELECT Id, Title, Content, LastModified FROM KnowledgeTable WHERE LastModified > ?", last_sync)
    else:
        cursor.execute("SELECT Id, Title, Content, LastModified FROM KnowledgeTable")

    rows = cursor.fetchall()
    conn.close()

    docs = []
    for row in rows:
        docs.append({
            "id": str(row.Id),
            "title": row.Title,
            "content": row.Content,
            "lastModified": row.LastModified.isoformat()
        })
    return docs

def push_to_ai_search(docs):
    url = f"{SEARCH_ENDPOINT}/indexes/{SEARCH_INDEX}/docs/index?api-version=2023-07-01-Preview"
    headers = {
        "Content-Type": "application/json",
        "api-key": SEARCH_API_KEY
    }
    payload = {
        "value": [{"@search.action": "mergeOrUpload", **doc} for doc in docs]
    }
    response = requests.post(url, headers=headers, json=payload)
    response.raise_for_status()
    return response.json()

def main(req: func.HttpRequest) -> func.HttpResponse:
    logging.info("MCP Function triggered for SQL → AI Search indexing")

    last_sync = req.params.get("lastSync")
    docs = get_sql_data(last_sync)

    if not docs:
        return func.HttpResponse(json.dumps({"message": "No new documents"}), mimetype="application

—

et’s proceed and finish the MCP-based Azure Function so it can fully index SQL data into Azure AI Search and be ready for your UI frontend to consume.

We left off inside the main() function — now we’ll complete it, add error handling, and then I’ll give you the UI integration steps.

PHASE 5 (continued) — Complete Azure Function

`function_app/init.py` (Final Version)

Python

import logging
import json
import pyodbc
import azure.functions as func
import requests
from datetime import datetime

# Azure AI Search config
SEARCH_ENDPOINT = "https://<YOUR-SEARCH-NAME>.search.windows.net"
SEARCH_API_KEY = "<YOUR-SEARCH-ADMIN-KEY>"
SEARCH_INDEX = "knowledge-index"

# SQL config
SQL_SERVER = "<YOUR-SQL-SERVER>.database.windows.net"
SQL_DATABASE = "<YOUR-DATABASE>"
SQL_USERNAME = "<YOUR-USERNAME>"
SQL_PASSWORD = "<YOUR-PASSWORD>"
SQL_DRIVER = "{ODBC Driver 18 for SQL Server}"

def get_sql_data(last_sync=None):
    conn_str = f"DRIVER={SQL_DRIVER};SERVER={SQL_SERVER};DATABASE={SQL_DATABASE};UID={SQL_USERNAME};PWD={SQL_PASSWORD}"
    conn = pyodbc.connect(conn_str)
    cursor = conn.cursor()

    if last_sync:
        cursor.execute("""
            SELECT Id, Title, Content, LastModified 
            FROM KnowledgeTable 
            WHERE LastModified > ?
        """, last_sync)
    else:
        cursor.execute("""
            SELECT Id, Title, Content, LastModified 
            FROM KnowledgeTable
        """)

    rows = cursor.fetchall()
    conn.close()

    docs = []
    for row in rows:
        docs.append({
            "id": str(row.Id),
            "title": row.Title,
            "content": row.Content,
            "lastModified": row.LastModified.isoformat()
        })
    return docs

def push_to_ai_search(docs):
    url = f"{SEARCH_ENDPOINT}/indexes/{SEARCH_INDEX}/docs/index?api-version=2023-07-01-Preview"
    headers = {
        "Content-Type": "application/json",
        "api-key": SEARCH_API_KEY
    }
    payload = {
        "value": [{"@search.action": "mergeOrUpload", **doc} for doc in docs]
    }
    response = requests.post(url, headers=headers, json=payload)
    response.raise_for_status()
    return response.json()

def main(req: func.HttpRequest) -> func.HttpResponse:
    logging.info("MCP Function triggered for SQL → AI Search indexing")

    try:
        last_sync = req.params.get("lastSync")
        if last_sync:
            try:
                last_sync = datetime.fromisoformat(last_sync)
            except ValueError:
                return func.HttpResponse(
                    json.dumps({"error": "Invalid lastSync format. Use ISO 8601."}),
                    mimetype="application/json",
                    status_code=400
                )

        docs = get_sql_data(last_sync)

        if not docs:
            return func.HttpResponse(
                json.dumps({"message": "No new documents found"}),
                mimetype="application/json",
                status_code=200
            )

        result = push_to_ai_search(docs)

        return func.HttpResponse(
            json.dumps({
                "message": f"Indexed {len(docs)} documents successfully",
                "aiSearchResponse": result
            }),
            mimetype="application/json",
            status_code=200
        )

    except Exception as e:
        logging.error(f"Error during indexing: {str(e)}", exc_info=True)
        return func.HttpResponse(
            json.dumps({"error": str(e)}),
            mimetype="application/json",
            status_code=500
        )

PHASE 6 — Deployment Steps

Create Function App in Azure Portal (Python runtime).
Add Application Settings:
- SEARCH_ENDPOINT
- SEARCH_API_KEY
- SEARCH_INDEX
- SQL_SERVER
- SQL_DATABASE
- SQL_USERNAME
- SQL_PASSWORD
Deploy using VS Code Azure Functions extension or func azure functionapp publish <APP_NAME>.
alternative deployment methods for your Azure Functions project (in case func azure functionapp publish fails), plus a recommended folder structure so your codebase stays clean and production-ready.
I’ll break this into two parts:
1. Alternative Ways to Deploy Azure Functions
If func azure functionapp publish doesn’t work (common issues: auth errors, CLI version mismatch, network restrictions), here are three reliable alternatives:
A. Deploy via Azure Portal (Zip Deploy)
Build your function locally:
Bash

Copy code
func azure functionapp publish <APP_NAME> --build local --nozip
This creates a .zip in your bin/Release or bin/Debug folder (for C#) or .python_packages (for Python).
Go to Azure Portal → Your Function App → Deployment Center → Zip Deploy.
Upload the .zip file manually.
✅ Pros: No CLI dependency, works behind firewalls.
⚠️ Cons: Manual step, not great for CI/CD.
B. Deploy via Azure CLI + Zip Deploy
Bash

Copy code
# Create zip zip -r functionapp.zip * .[^.]* -x "*.git*" "*.vscode*" "__pycache__/*" # Deploy az functionapp deployment source config-zip \ --resource-group <RESOURCE_GROUP> \ --name <APP_NAME> \ --src functionapp.zip
✅ Pros: Scriptable, works in CI/CD.
⚠️ Cons: Must manage zip creation.
C. Deploy via GitHub Actions (CI/CD)
Push your code to GitHub.
In Azure Portal → Function App → Deployment Center → GitHub Actions → Configure.
Azure auto-generates a .github/workflows/azure-functions.yml file.
✅ Pros: Fully automated, integrates with PRs.
⚠️ Cons: Requires GitHub repo and permissions.
D. Deploy via VS Code Azure Functions Extension
Install Azure Functions extension in VS Code.
Sign in to Azure in VS Code.
Right-click your function project → Deploy to Function App.
✅ Pros: One-click deploy, good for dev/test.
⚠️ Cons: Not ideal for production automation.
2. Recommended Folder Structure
Here’s a Python Azure Functions project structure that works well for your scheduled + event-driven + HTTP API setup:

Copy code
project-root/ │ ├── host.json # Global function app settings ├── local.settings.json # Local dev settings (not committed) ├── requirements.txt # Python dependencies ├── .funcignore # Files to ignore during publish ├── .gitignore │ ├── shared/ # Shared code modules │ ├── __init__.py │ ├── index_manager.py # Push/delete docs in AI Search │ ├── sql_connector.py # SQL queries + CDC fetch │ ├── blob_state.py # Last sync state management │ └── event_processor.py # Event Grid payload handling │ ├── timer_sync/ # Scheduled sync function │ ├── __init__.py │ └── function.json │ ├── eventgrid_sync/ # Real-time Event Grid function │ ├── __init__.py │ └── function.json │ ├── get_index_status/ # HTTP API for UI/MCP │ ├── __init__.py │ └── function.json │ └── manual_refresh/ # HTTP API to trigger reindex ├── __init__.py └── function.json
3. Deployment Tips if CLI Publish Fails
Check Azure Functions Core Tools version:
Bash

Copy code
func --version
Make sure it matches your runtime (e.g., v4 for Python 3.9+).
Force build before publish:
Bash

Copy code
func azure functionapp publish <APP_NAME> --build local
Use az webapp deployment as fallback:
Bash

code
az webapp deployment source config-local-git \ --name <APP_NAME> \ --resource-group <RESOURCE_GROUP>
Then push via Git:
Bash

git push azure master

–

proceed and finish this with a full deployment blueprint so you have a step-by-step fallback plan for getting your Azure Functions + AI Search + SQL CDC/Event Grid solution live, even if func azure functionapp publish fails.

I’ll give you:

Deployment Strategy Matrix (choose based on your environment)
Step-by-Step for Each Method
CI/CD Pipeline Example (GitHub Actions)
Post-Deployment Verification Checklist

1. Deployment Strategy Matrix

Method	Best For	Pros	Cons
Zip Deploy via Azure CLI	Quick manual/scripted deploy	Works without Core Tools, easy to automate	Must package manually
GitHub Actions CI/CD	Production automation	Fully automated, integrates with PRs	Requires GitHub repo
VS Code Deploy	Dev/test	One-click, no CLI needed	Not ideal for production
Local Git Deploy	Teams with Git workflows	Simple push-to-deploy	Slower than Zip Deploy
Azure DevOps Pipeline	Enterprise CI/CD	Full control, approvals	More setup time

2. Step-by-Step for Each Method

A. Zip Deploy via Azure CLI

BashCopy code# 1. Create zip (exclude dev files)
zip -r functionapp.zip * .[^.]* -x "*.git*" "*.vscode*" "__pycache__/*"

# 2. Deploy
az functionapp deployment source config-zip \
  --resource-group <RESOURCE_GROUP> \
  --name <APP_NAME> \
  --src functionapp.zip

B. GitHub Actions CI/CD

Push your project to GitHub.
In Azure Portal → Function App → Deployment Center → GitHub Actions → Configure.
Azure generates .github/workflows/azure-functions.yml:

YamlCopy codename: Deploy Azure Function App

on:
  push:
    branches:
      - main

jobs:
  build-and-deploy:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
      - name: Setup Python
        uses: actions/setup-python@v4
        with:
          python-version: '3.10'
      - name: Install dependencies
        run: pip install -r requirements.txt
      - name: Azure Functions Action
        uses: Azure/functions-action@v1
        with:
          app-name: <APP_NAME>
          package: '.'
          publish-profile: ${{ secrets.AZURE_FUNCTIONAPP_PUBLISH_PROFILE }}

Get the Publish Profile from Azure Portal → Function App → Get Publish Profile and store it in GitHub Secrets as AZURE_FUNCTIONAPP_PUBLISH_PROFILE.

C. VS Code Deploy

Install Azure Functions extension.
Sign in to Azure in VS Code.
Right-click project → Deploy to Function App.

D. Local Git Deploy

BashCopy code# Enable local git deployment
az functionapp deployment source config-local-git \
  --name <APP_NAME> \
  --resource-group <RESOURCE_GROUP>

# Add remote
git remote add azure <DEPLOYMENT_GIT_URL>

# Push
git push azure main

3. Recommended Folder Structure (Recap)

Copy codeproject-root/
│
├── host.json
├── local.settings.json
├── requirements.txt
├── .funcignore
├── .gitignore
│
├── shared/
│   ├── index_manager.py
│   ├── sql_connector.py
│   ├── blob_state.py
│   └── event_processor.py
│
├── timer_sync/
│   ├── __init__.py
│   └── function.json
│
├── eventgrid_sync/
│   ├── __init__.py
│   └── function.json
│
├── get_index_status/
│   ├── __init__.py
│   └── function.json
│
└── manual_refresh/
    ├── __init__.py
    └── function.json

–

wrap this up with the final Azure Functions project structure and code for each file so you have a complete, working baseline for:

Timer-triggered sync (scheduled)
Event Grid-triggered sync (real-time)
HTTP API for status and manual refresh
Shared modules for SQL, Blob state, AI Search, and Event processing

This will be Python-based (Azure Functions v4, Python 3.10+), but the structure works for other runtimes too.

1. Final Folder Structure

Copy codeproject-root/
│
├── host.json
├── local.settings.json
├── requirements.txt
├── .funcignore
├── .gitignore
│
├── shared/
│   ├── __init__.py
│   ├── index_manager.py
│   ├── sql_connector.py
│   ├── blob_state.py
│   └── event_processor.py
│
├── timer_sync/
│   ├── __init__.py
│   └── function.json
│
├── eventgrid_sync/
│   ├── __init__.py
│   └── function.json
│
├── get_index_status/
│   ├── __init__.py
│   └── function.json
│
└── manual_refresh/
    ├── __init__.py
    └── function.json

2. Root Files

host.json

JsonCopy code{
  "version": "2.0",
  "logging": {
    "applicationInsights": {
      "samplingSettings": {
        "isEnabled": true,
        "excludedTypes": "Request"
      }
    }
  }
}

local.settings.json (do not commit to Git)

JsonCopy code{
  "IsEncrypted": false,
  "Values": {
    "AzureWebJobsStorage": "UseDevelopmentStorage=true",
    "FUNCTIONS_WORKER_RUNTIME": "python",
    "SQL_CONNECTION_STRING": "<your-sql-connection-string>",
    "BLOB_CONNECTION_STRING": "<your-blob-connection-string>",
    "BLOB_CONTAINER_NAME": "state",
    "BLOB_STATE_FILE": "last_sync.txt",
    "AI_SEARCH_ENDPOINT": "<your-ai-search-endpoint>",
    "AI_SEARCH_KEY": "<your-ai-search-key>",
    "AI_SEARCH_INDEX": "knowledge-index"
  }
}

requirements.txt

Copy codeazure-functions
azure-storage-blob
pyodbc
requests

3. Shared Modules

shared/index_manager.py

PythonCopy codeimport requests
import os

class IndexManager:
    def __init__(self):
        self.endpoint = os.environ["AI_SEARCH_ENDPOINT"]
        self.api_key = os.environ["AI_SEARCH_KEY"]
        self.index_name = os.environ["AI_SEARCH_INDEX"]

    def push_docs(self, docs):
        url = f"{self.endpoint}/indexes/{self.index_name}/docs/index?api-version=2021-04-30-Preview"
        headers = {"Content-Type": "application/json", "api-key": self.api_key}
        payload = {"value": docs}
        r = requests.post(url, headers=headers, json=payload)
        r.raise_for_status()
        return r.json()

    def delete_docs(self, ids):
        url = f"{self.endpoint}/indexes/{self.index_name}/docs/index?api-version=2021-04-30-Preview"
        headers = {"Content-Type": "application/json", "api-key": self.api_key}
        payload = {"value": [{"@search.action": "delete", "id": i} for i in ids]}
        r = requests.post(url, headers=headers, json=payload)
        r.raise_for_status()
        return r.json()

shared/sql_connector.py

PythonCopy codeimport pyodbc
import os

class SqlConnector:
    def __init__(self):
        self.conn_str = os.environ["SQL_CONNECTION_STRING"]

    def get_all_ids(self):
        with pyodbc.connect(self.conn_str) as conn:
            cursor = conn.cursor()
            cursor.execute("SELECT id FROM Documents")
            return [row.id for row in cursor.fetchall()]

    def get_changes(self, last_sync):
        with pyodbc.connect(self.conn_str) as conn:
            cursor = conn.cursor()
            cursor.execute("SELECT id, title, content FROM Documents WHERE lastModified > ?", last_sync)
            return [{"id": row.id, "title": row.title, "content": row.content} for row in cursor.fetchall()]

shared/blob_state.py

let’s proceed and finish the remaining files so you have the complete working code for your Azure Functions project.

We stopped at shared/blob_state.py, so let’s continue from there.

3. Shared Modules (continued)

shared/blob_state.py

PythonCopy codefrom azure.storage.blob import BlobServiceClient
import os
from datetime import datetime

class BlobState:
    def __init__(self):
        self.conn_str = os.environ["BLOB_CONNECTION_STRING"]
        self.container_name = os.environ["BLOB_CONTAINER_NAME"]
        self.state_file = os.environ["BLOB_STATE_FILE"]
        self.client = BlobServiceClient.from_connection_string(self.conn_str)
        self.container_client = self.client.get_container_client(self.container_name)

    def get_last_sync(self):
        try:
            blob_client = self.container_client.get_blob_client(self.state_file)
            data = blob_client.download_blob().readall().decode("utf-8")
            return datetime.fromisoformat(data.strip())
        except Exception:
            return None

    def set_last_sync(self, timestamp: datetime):
        blob_client = self.container_client.get_blob_client(self.state_file)
        blob_client.upload_blob(timestamp.isoformat(), overwrite=True)

shared/event_processor.py

PythonCopy codefrom .index_manager import IndexManager

class EventProcessor:
    def __init__(self):
        self.index_manager = IndexManager()

    def process_event(self, event):
        data = event.get("data", {})
        op_type = data.get("operationType")
        doc = data.get("document", {})

        if op_type in ("insert", "update"):
            doc["@search.action"] = "mergeOrUpload"
            return self.index_manager.push_docs([doc])
        elif op_type == "delete":
            return self.index_manager.delete_docs([doc.get("id")])
        else:
            return {"status": "ignored"}

4. Function Implementations

A. Timer Trigger — `timer_sync/init.py`

PythonCopy codeimport datetime
import azure.functions as func
from ..shared.sql_connector import SqlConnector
from ..shared.index_manager import IndexManager
from ..shared.blob_state import BlobState

def main(mytimer: func.TimerRequest) -> None:
    sql = SqlConnector()
    index = IndexManager()
    state = BlobState()

    last_sync = state.get_last_sync()
    now = datetime.datetime.utcnow()

    changes = sql.get_changes(last_sync) if last_sync else sql.get_changes(datetime.datetime(1970, 1, 1))
    if changes:
        index.push_docs(changes)
        state.set_last_sync(now)
        print(f"Indexed {len(changes)} documents.")
    else:
        print("No changes found.")

timer_sync/function.json

JsonCopy code{
  "scriptFile": "__init__.py",
  "bindings": [
    {
      "name": "mytimer",
      "type": "timerTrigger",
      "direction": "in",
      "schedule": "0 0 * * * *"
    }
  ]
}

B. Event Grid Trigger — `eventgrid_sync/init.py`

PythonCopy codeimport azure.functions as func
import json
from ..shared.event_processor import EventProcessor

def main(event: func.EventGridEvent):
    processor = EventProcessor()
    event_data = json.loads(event.get_body().decode())
    result = processor.process_event(event_data)
    print(f"Processed EventGrid event: {result}")

eventgrid_sync/function.json

JsonCopy code{
  "scriptFile": "__init__.py",
  "bindings": [
    {
      "name": "event",
      "type": "eventGridTrigger",
      "direction": "in"
    }
  ]
}

C. HTTP API — Get Index Status — `get_index_status/init.py`

PythonCopy codeimport azure.functions as func
from ..shared.blob_state import BlobState
from ..shared.sql_connector import SqlConnector

def main(req: func.HttpRequest) -> func.HttpResponse:
    state = BlobState()
    sql = SqlConnector()

    last_sync = state.get_last_sync()
    doc_count = len(sql.get_all_ids())

    return func.HttpResponse(
        body={
            "lastSync": last_sync.isoformat() if last_sync else None,
            "docCount": doc_count
        },
        mimetype="application/json"
    )

get_index_status/function.json

JsonCopy code{
  "scriptFile": "__init__.py",
  "bindings": [
    {
      "authLevel": "function",
      "type": "httpTrigger",
      "direction

proceed and finish the last pieces so you have the full working Azure Functions project with all files in place.

We stopped at get_index_status/function.json, so now we’ll complete that and add the manual refresh function.

C. HTTP API — Get Index Status (continued)

get_index_status/function.json

JsonCopy code{
  "scriptFile": "__init__.py",
  "bindings": [
    {
      "authLevel": "function",
      "type": "httpTrigger",
      "direction": "in",
      "name": "req",
      "methods": [ "get" ]
    },
    {
      "type": "http",
      "direction": "out",
      "name": "$return"
    }
  ]
}

D. HTTP API — Manual Refresh

manual_refresh/__init__.py

PythonCopy codeimport datetime
import azure.functions as func
from ..shared.sql_connector import SqlConnector
from ..shared.index_manager import IndexManager
from ..shared.blob_state import BlobState

def main(req: func.HttpRequest) -> func.HttpResponse:
    sql = SqlConnector()
    index = IndexManager()
    state = BlobState()

    now = datetime.datetime.utcnow()
    changes = sql.get_changes(datetime.datetime(1970, 1, 1))  # full refresh

    if changes:
        index.push_docs(changes)
        state.set_last_sync(now)
        return func.HttpResponse(
            body='{"message": "Manual refresh completed"}',
            mimetype="application/json"
        )
    else:
        return func.HttpResponse(
            body='{"message": "No documents to refresh"}',
            mimetype="application/json"
        )

manual_refresh/function.json

JsonCopy code{
  "scriptFile": "__init__.py",
  "bindings": [
    {
      "authLevel": "function",
      "type": "httpTrigger",
      "direction": "in",
      "name": "req",
      "methods": [ "post" ]
    },
    {
      "type": "http",
      "direction": "out",
      "name": "$return"
    }
  ]
}

E. .funcignore

This ensures unnecessary files aren’t deployed.

Copy code.git*
.vscode/
__pycache__/
local.settings.json
*.pyc

F. .gitignore

Copy code__pycache__/
*.pyc
local.settings.json
.env

G. How It All Works Together

Timer Trigger (timer_sync)
- Runs on schedule (every hour in this example).
- Checks last sync timestamp in Blob Storage.
- Pulls changed docs from SQL.
- Pushes them to Azure AI Search.
- Updates last sync timestamp in Blob.
Event Grid Trigger (eventgrid_sync)
- Fires instantly when a change event is sent from Event Grid.
- Processes insert/update/delete events.
- Updates AI Search index accordingly.
HTTP API — Get Index Status (get_index_status)
- Returns last sync timestamp and total doc count.
- Useful for monitoring and MCP Inspector testing.
HTTP API — Manual Refresh (manual_refresh)
- Forces a full reindex from SQL to AI Search.
- Updates last sync timestamp.

✅ With this structure, you can:

Run locally with func start
Deploy with func azure functionapp publish <APP_NAME> or Zip Deploy
Test with MCP Inspector using the JSON configs we built earlier
Trigger both scheduled and real-time indexing

If you want, I can now give you a full “first-run” test script that will:

Start the function locally
Call all endpoints in sequence
Simulate an Event Grid event
Verify AI Search results

proceed and create your full “first-run” test script so you can validate the entire Azure Functions + Event Grid + AI Search pipeline end-to-end in one go.

This script will:

Start the function app locally
Call the HTTP APIs (get_index_status, manual_refresh)
Simulate an Event Grid event
Query Azure AI Search to confirm the document is indexed

1. Prerequisites

Python 3.10+ installed
Azure Functions Core Tools installed (func --version should be v4)
Azure CLI installed and logged in (az login)
local.settings.json configured with your connection strings and keys
AI Search index already created in Azure

2. Save This Script as `test_pipeline.py`

Python

import subprocess
import time
import requests
import json
from datetime import datetime

# Local endpoints
GET_STATUS_URL = "http://localhost:7071/api/get_index_status"
MANUAL_REFRESH_URL = "http://localhost:7071/api/manual_refresh"

# Event Grid simulation (local function trigger)
EVENTGRID_URL = "http://localhost:7071/runtime/webhooks/EventGrid?functionName=eventgrid_sync"

# AI Search query
AI_SEARCH_ENDPOINT = "<YOUR_AI_SEARCH_ENDPOINT>"
AI_SEARCH_KEY = "<YOUR_AI_SEARCH_KEY>"
AI_SEARCH_INDEX = "<YOUR_AI_SEARCH_INDEX>"

def start_functions():
    print("🚀 Starting Azure Functions locally...")
    proc = subprocess.Popen(["func", "start"], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    time.sleep(5)  # Give it time to start
    return proc

def get_index_status():
    print("📊 Getting index status...")
    r = requests.get(GET_STATUS_URL)
    print("Response:", r.text)

def manual_refresh():
    print("🔄 Triggering manual refresh...")
    r = requests.post(MANUAL_REFRESH_URL)
    print("Response:", r.text)

def send_eventgrid_event():
    print("⚡ Sending Event Grid test event...")
    payload = [
        {
            "id": "12345",
            "eventType": "Microsoft.Storage.BlobCreated",
            "subject": "/blobServices/default/containers/test/blobs/doc.json",
            "eventTime": datetime.utcnow().isoformat() + "Z",
            "data": {
                "operationType": "insert",
                "document": {
                    "id": "doc-001",
                    "title": "Test Document",
                    "content": "This is a test",
                    "lastModified": datetime.utcnow().isoformat() + "Z"
                }
            },
            "dataVersion": "1.0"
        }
    ]
    r = requests.post(EVENTGRID_URL, json=payload)
    print("Response:", r.status_code, r.text)

def query_ai_search():
    print("🔍 Querying AI Search for 'Test Document'...")
    url = f"{AI_SEARCH_ENDPOINT}/indexes/{AI_SEARCH_INDEX}/docs/search?api-version=2021-04-30-Preview"
    headers = {
        "Content-Type": "application/json",
        "api-key": AI_SEARCH_KEY
    }
    payload = {"search": "Test Document"}
    r = requests.post(url, headers=headers, json=payload)
    print("Search Results:", json.dumps(r.json(), indent=2))

if __name__ == "__main__":
    proc = start_functions()
    try:
        get_index_status()
        manual_refresh()
        time.sleep(3)  # Wait for refresh to complete
        get_index_status()
        send_eventgrid_event()
        time.sleep(3)  # Wait for event to process
        query_ai_search()
    finally:
        print("🛑 Stopping Azure Functions...")
        proc.terminate()

3. How to Run

BashCopy codepython test_pipeline.py

4. Expected Output

Start Functions → You’ll see Azure Functions logs starting up.
Get Index Status → JSON with lastSync and docCount.
Manual Refresh → "Manual refresh completed".
Get Index Status Again → Updated lastSync.
Event Grid Test → HTTP 200 response.
AI Search Query → JSON with "Test Document" in results.

5. Why This Script Helps

Runs all triggers (timer excluded, since it’s schedule-based)
Confirms local dev environment works before deployment
Lets you debug without switching between tools
Can be adapted for cloud testing by replacing URLs with deployed endpoints

4. Post-Deployment Verification Checklist

After deployment, verify:

Function App Status:BashCopy codeaz functionapp show --name <APP_NAME> --resource-group <RESOURCE_GROUP> --query state Should be "Running".
Function Endpoints:
Test each HTTP function with curl or Postman.

–

PHASE 7 — UI Frontend Integration

Your UI can now call the Function App endpoint to trigger indexing or query the AI Search index.

Example: Trigger Indexing from UI

JavascriptCopy codeasync function triggerIndexing() {
  const response = await fetch(
    "https://<YOUR-FUNCTION-APP>.azurewebsites.net/api/IndexSqlToAISearch?lastSync=2026-09-01T00:00:00"

proceed and finish the UI integration so your MCP-based Azure Function App can be triggered from the frontend, and your Azure AI Search knowledge base can be queried directly from the UI.

We already have the indexing endpoint in place — now we’ll:

Show how the frontend can trigger indexing.
Show how the frontend can query the AI Search index for results.
Wrap it into a simple UI flow.

PHASE 7 (continued) — UI Frontend Integration

1. Trigger Indexing from UI

This will call your Azure Function to pull data from SQL and push it into Azure AI Search.

JavascriptCopy codeasync function triggerIndexing() {
  try {
    const response = await fetch(
      "https://<YOUR-FUNCTION-APP>.azurewebsites.net/api/IndexSqlToAISearch?lastSync=2026-09-01T00:00:00",
      { method: "GET" }
    );
    const data = await response.json();
    console.log("Indexing result:", data);
    alert(`Indexing completed: ${data.message}`);
  } catch (error) {
    console.error("Error triggering indexing:", error);
    alert("Indexing failed. Check console for details.");
  }
}

2. Query Azure AI Search from UI

We’ll create a separate Azure Function to query the AI Search index and return results to the frontend.

`query_search/init.py`

PythonCopy codeimport logging
import json
import azure.functions as func
import requests

SEARCH_ENDPOINT = "https://<YOUR-SEARCH-NAME>.search.windows.net"
SEARCH_API_KEY = "<YOUR-SEARCH-QUERY-KEY>"
SEARCH_INDEX = "knowledge-index"

def main(req: func.HttpRequest) -> func.HttpResponse:
    logging.info("Query AI Search Function triggered")

    query = req.params.get("q")
    if not query:
        return func.HttpResponse(
            json.dumps({"error": "Missing query parameter 'q'"}),
            mimetype="application/json",
            status_code=400
        )

    url = f"{SEARCH_ENDPOINT}/indexes/{SEARCH_INDEX}/docs/search?api-version=2023-07-01-Preview"
    headers = {
        "Content-Type": "application/json",
        "api-key": SEARCH_API_KEY
    }
    payload = {
        "search": query,
        "top": 10
    }

    try:
        response = requests.post(url, headers=headers, json=payload)
        response.raise_for_status()
        return func.HttpResponse(
            json.dumps(response.json()),
            mimetype="application/json",
            status_code=200
        )
    except Exception as e:
        logging.error(f"Search query failed: {str(e)}", exc_info=True)
        return func.HttpResponse(
            json.dumps({"error": str(e)}),
            mimetype="application/json",
            status_code=500
        )

3. Query from UI

JavascriptCopy codeasync function searchKnowledgeBase(query) {
  try {
    const response = await fetch(
      `https://<YOUR-FUNCTION-APP>.azurewebsites.net/api/QueryAISearch?q=${encodeURIComponent(query)}`
    );
    const results = await response.json();
    console.log("Search results:", results);

    const container = document.getElementById("results");
    container.innerHTML = "";
    results.value.forEach(doc => {
      const div = document.createElement("div");
      div.innerHTML = `<h3>${doc.title}</h3><p>${doc.content}</p>`;
      container.appendChild(div);
    });
  } catch (error) {
    console.error("Error searching knowledge base:", error);
    alert("Search failed. Check console for details.");
  }
}

4. Simple HTML UI

HtmlCopy code<!DOCTYPE html>
<html>
<head>
  <title>Knowledge Base Search</title>
</head>
<body>
  <h1>Knowledge Base</h1>
  <button onclick="triggerIndexing()">Reindex Data</button>
  


  <input type="text" id="searchBox" placeholder="Enter search term">
  <button onclick="searchKnowledgeBase(document.getElementById('searchBox').value)">Search</button>
  <div id="results"></div>

  <script src="app.js"></script>
</body>
</html>

PHASE 8 — How This Works Together

Reindex Data button → Calls IndexSqlToAISearch Function → Pulls from SQL → Pushes to AI Search.
Search button → Calls `

proceed and finish PHASE 8 so you have the full end-to-end flow for your MCP-based Azure Function App, Azure AI Search knowledge base, and UI frontend.

PHASE 8 — Full Workflow

1. End-to-End Flow

User clicks “Reindex Data” in the UI.
- Calls IndexSqlToAISearch Azure Function (MCP tool).
- Function connects to Azure SQL Database, retrieves new/updated rows, and pushes them into Azure AI Search.
- AI Search index (knowledge-index) is updated.
User enters a search term in the UI and clicks “Search”.
- Calls QueryAISearch Azure Function.
- Function sends the query to Azure AI Search using the query key.
- AI Search returns matching documents.
- UI renders results dynamically.

2. Deployment Steps

Backend

Create two Azure Functions in the same Function App:
- IndexSqlToAISearch (indexing)
- QueryAISearch (searching)
Add Application Settings in Azure Portal:
- SEARCH_ENDPOINT
- SEARCH_API_KEY (admin key for indexing)
- SEARCH_QUERY_KEY (query key for searching)
- SEARCH_INDEX
- SQL_SERVER
- SQL_DATABASE
- SQL_USERNAME
- SQL_PASSWORD
Deploy using:BashCopy codefunc azure functionapp publish <YOUR_FUNCTION_APP_NAME>

Frontend

Host your HTML + JS files in:
- Azure Static Web Apps
- Azure Blob Storage (static website)
- Or any web server
Update the Function App URLs in app.js:JavascriptCopy codeconst INDEX_URL = "https://<YOUR-FUNCTION-APP>.azurewebsites.net/api/IndexSqlToAISearch"; const SEARCH_URL = "https://<YOUR-FUNCTION-APP>.azurewebsites.net/api/QueryAISearch";

3. Security Best Practices

Use Azure Managed Identity instead of storing SQL credentials in plain text.
Restrict AI Search admin key usage to indexing function only.
Use CORS settings in Azure Functions to allow only your frontend domain.
Consider API Management in front of your Functions for rate limiting and monitoring.

4. Optional MCP Integration

If you want this to be agent-driven in Foundry:

Register IndexSqlToAISearch and QueryAISearch as MCP tools in your Foundry Agent configuration.
The agent can then:
- Trigger reindexing automatically on a schedule.
- Query the knowledge base to answer user questions in natural language.

✅ Now you have:

SQL → AI Search indexing function
AI Search query function
Frontend UI to trigger indexing and search
Deployment & security guidance
Optional MCP agent integration

If you want, I can now extend this so that the indexing function runs automatically on a schedule (e.g., every hour) without the user clicking “Reindex Data” — making it a fully automated knowledge base refresh.

proceed and add the scheduled auto-indexing enhancement so your SQL → Azure AI Search pipeline refreshes automatically without requiring the user to click “Reindex Data” in the UI.

This will make your knowledge base self-updating and always fresh for both the UI and any MCP/Foundry agent that uses it.

PHASE 9 — Scheduled Auto-Indexing

1. How It Works

Azure Functions supports Timer Triggers that run on a CRON schedule.
We’ll create a new function (ScheduledIndexSqlToAISearch) that runs the same indexing logic as your HTTP-triggered function, but automatically.
You can set the schedule to hourly, daily, or custom intervals.

2. Timer Trigger Function Code

Create a new function in your Function App:

`scheduled_index/init.py`

PythonCopy codeimport logging
import json
import pyodbc
import requests
from datetime import datetime, timedelta
import azure.functions as func

# Azure AI Search config
SEARCH_ENDPOINT = "https://<YOUR-SEARCH-NAME>.search.windows.net"
SEARCH_API_KEY = "<YOUR-SEARCH-ADMIN-KEY>"
SEARCH_INDEX = "knowledge-index"

# SQL config
SQL_SERVER = "<YOUR-SQL-SERVER>.database.windows.net"
SQL_DATABASE = "<YOUR-DATABASE>"
SQL_USERNAME = "<YOUR-USERNAME>"
SQL_PASSWORD = "<YOUR-PASSWORD>"
SQL_DRIVER = "{ODBC Driver 18 for SQL Server}"

# Store last sync in memory (for demo; use durable storage in production)
last_sync_time = None

def get_sql_data(last_sync=None):
    conn_str = f"DRIVER={SQL_DRIVER};SERVER={SQL_SERVER};DATABASE={SQL_DATABASE};UID={SQL_USERNAME};PWD={SQL_PASSWORD}"
    conn = pyodbc.connect(conn_str)
    cursor = conn.cursor()

    if last_sync:
        cursor.execute("""
            SELECT Id, Title, Content, LastModified 
            FROM KnowledgeTable 
            WHERE LastModified > ?
        """, last_sync)
    else:
        cursor.execute("""
            SELECT Id, Title, Content, LastModified 
            FROM KnowledgeTable
        """)

    rows = cursor.fetchall()
    conn.close()

    docs = []
    for row in rows:
        docs.append({
            "id": str(row.Id),
            "title": row.Title,
            "content": row.Content,
            "lastModified": row.LastModified.isoformat()
        })
    return docs

def push_to_ai_search(docs):
    url = f"{SEARCH_ENDPOINT}/indexes/{SEARCH_INDEX}/docs/index?api-version=2023-07-01-Preview"
    headers = {
        "Content-Type": "application/json",
        "api-key": SEARCH_API_KEY
    }
    payload = {
        "value": [{"@search.action": "mergeOrUpload", **doc} for doc in docs]
    }
    response = requests.post(url, headers=headers, json=payload)
    response.raise_for_status()
    return response.json()

def main(mytimer: func.TimerRequest) -> None:
    global last_sync_time
    logging.info("Scheduled SQL → AI Search indexing started")

    try:
        docs = get_sql_data(last_sync_time)

        if not docs:
            logging.info("No new documents found")
        else:
            result = push_to_ai_search(docs)
            logging.info(f"Indexed {len(docs)} documents successfully: {result}")

        # Update last sync time
        last_sync_time = datetime.utcnow()

    except Exception as e:
        logging.error(f"Error during scheduled indexing: {str(e)}", exc_info=True)

3. Function Schedule

In function.json for this function:

JsonCopy code{
  "scriptFile": "__init__.py",
  "bindings": [
    {
      "name": "mytimer",
      "type": "timerTrigger",
      "direction": "in",
      "schedule": "0 0 * * * *"
    }
  ]
}

Schedule format:

"0 0 * * * *" → Every hour on the hour
"0 0 0 * * *" → Every day at midnight UTC
"0 */30 * * * *" → Every 30 minutes

4. Deployment

Add this new function to your Function App project.
Deploy with:BashCopy codefunc azure functionapp publish <YOUR_FUNCTION_APP_NAME>
Azure will now run the indexing automatically based on your CRON schedule.

5. Benefits

Hands-free updates — no manual reindexing needed.
**Always fresh

proceed and wrap this up by making your scheduled auto-indexing setup production-ready, fully integrated with your UI frontend and MCP/Foundry agent so it’s a complete, self-maintaining knowledge base system.

PHASE 10 — Production-Ready Enhancements

1. Persisting Last Sync Time

Right now, our last_sync_time is stored in memory, which resets if the Function App restarts.
In production, we should store it in Azure Table Storage, Blob Storage, or Azure SQL itself.

Example: Store in Azure Blob Storage

PythonCopy codefrom azure.storage.blob import BlobServiceClient

BLOB_CONN_STR = "<YOUR-STORAGE-CONNECTION-STRING>"
BLOB_CONTAINER = "function-state"
BLOB_NAME = "last_sync.txt"

def get_last_sync():
    blob_service = BlobServiceClient.from_connection_string(BLOB_CONN_STR)
    blob_client = blob_service.get_blob_client(container=BLOB_CONTAINER, blob=BLOB_NAME)
    try:
        data = blob_client.download_blob().readall().decode("utf-8")
        return datetime.fromisoformat(data)
    except:
        return None

def set_last_sync(timestamp):
    blob_service = BlobServiceClient.from_connection_string(BLOB_CONN_STR)
    blob_client = blob_service.get_blob_client(container=BLOB_CONTAINER, blob=BLOB_NAME)
    blob_client.upload_blob(timestamp.isoformat(), overwrite=True)

Replace the in-memory last_sync_time with calls to get_last_sync() and set_last_sync().

2. UI Frontend Awareness

Since indexing is now automatic, the UI can:

Show last updated time by calling a small Function that reads the stored last_sync_time.
Disable the “Reindex” button unless a manual override is needed.

Example:

JavascriptCopy codeasync function getLastUpdated() {
  const res = await fetch("https://<YOUR-FUNCTION-APP>.azurewebsites.net/api/GetLastSync");
  const data = await res.json();
  document.getElementById("lastUpdated").innerText = `Last updated: ${data.lastSync}`;
}

3. MCP / Foundry Agent Integration

Now that the index is always fresh:

Register your QueryAISearch endpoint as an MCP tool in Foundry.
The agent can answer questions directly from the AI Search index without manual refresh.
Optionally, register the IndexSqlToAISearch endpoint as a tool for manual override.

Example MCP Tool Config (conceptual):

JsonCopy code{
  "name": "query_knowledge_base",
  "description": "Searches the Azure AI Search knowledge base for relevant documents",
  "endpoint": "https://<YOUR-FUNCTION-APP>.azurewebsites.net/api/QueryAISearch",
  "method": "GET",
  "parameters": [
    { "name": "q", "type": "string", "required": true }
  ]
}

4. Security Hardening

Use Managed Identity for SQL and Blob access instead of passwords.
Restrict AI Search admin key to indexing functions only.
Use query key for search functions.
Enable CORS in Azure Functions to allow only your frontend domain.
Optionally, put Azure API Management in front of your Functions for rate limiting and monitoring.

5. Final End-to-End Flow

Timer Trigger runs every hour → Pulls new/updated rows from SQL → Pushes to AI Search → Updates last_sync_time in Blob Storage.
UI Frontend queries AI Search via QueryAISearch Function → Displays results instantly.
Foundry Agent uses MCP tool to query the same AI Search index → Answers user questions in natural language.
Manual Override: UI “Reindex” button calls IndexSqlToAISearch if needed.

✅ Now you have:

Automated scheduled indexing from SQL to Azure AI Search.
Persistent last sync tracking.
UI integration for search and status display.
MCP/Foundry agent integration for conversational access.
Security best practices for production.

If you want, I can now extend this one step further so that deleted rows in SQL are also removed from Azure AI Search automatically during scheduled runs — making it a full sync instead of just incremental updates.

proceed and add the auto-deletion sync enhancement to your scheduled SQL → Azure AI Search indexing job, so that your search index always mirrors your SQL database exactly — no stale or orphaned documents.

PHASE 11 — Auto-Deletion Sync for Scheduled Indexing

1. Why This Matters

Without deletion sync:

Old records removed from SQL will still appear in search results.
Users may see outdated or irrelevant content.

With deletion sync:

The Azure AI Search index is a true reflection of your SQL database.
Search results are always accurate and trustworthy.

2. Updated Scheduled Function Logic

We’ll:

Retrieve all current IDs from SQL.
Retrieve all current IDs from Azure AI Search.
Compare and find IDs that exist in AI Search but not in SQL.
Delete those from AI Search.
Continue with incremental indexing for new/updated rows.

`scheduled_index/init.py` (Updated)


import logging
import json
import pyodbc
import requests
from datetime import datetime
import azure.functions as func
from azure.storage.blob import BlobServiceClient

# Azure AI Search config
SEARCH_ENDPOINT = "https://<YOUR-SEARCH-NAME>.search.windows.net"
SEARCH_API_KEY = "<YOUR-SEARCH-ADMIN-KEY>"
SEARCH_INDEX = "knowledge-index"

# SQL config
SQL_SERVER = "<YOUR-SQL-SERVER>.database.windows.net"
SQL_DATABASE = "<YOUR-DATABASE>"
SQL_USERNAME = "<YOUR-USERNAME>"
SQL_PASSWORD = "<YOUR-PASSWORD>"
SQL_DRIVER = "{ODBC Driver 18 for SQL Server}"

# Blob storage config for last sync persistence
BLOB_CONN_STR = "<YOUR-STORAGE-CONNECTION-STRING>"
BLOB_CONTAINER = "function-state"
BLOB_NAME = "last_sync.txt"

def get_last_sync():
    blob_service = BlobServiceClient.from_connection_string(BLOB_CONN_STR)
    blob_client = blob_service.get_blob_client(container=BLOB_CONTAINER, blob=BLOB_NAME)
    try:
        data = blob_client.download_blob().readall().decode("utf-8")
        return datetime.fromisoformat(data)
    except:
        return None

def set_last_sync(timestamp):
    blob_service = BlobServiceClient.from_connection_string(BLOB_CONN_STR)
    blob_client = blob_service.get_blob_client(container=BLOB_CONTAINER, blob=BLOB_NAME)
    blob_client.upload_blob(timestamp.isoformat(), overwrite=True)

def get_sql_data(last_sync=None):
    conn_str = f"DRIVER={SQL_DRIVER};SERVER={SQL_SERVER};DATABASE={SQL_DATABASE};UID={SQL_USERNAME};PWD={SQL_PASSWORD}"
    conn = pyodbc.connect(conn_str)
    cursor = conn.cursor()

    if last_sync:
        cursor.execute("""
            SELECT Id, Title, Content, LastModified 
            FROM KnowledgeTable 
            WHERE LastModified > ?
        """, last_sync)
    else:
        cursor.execute("""
            SELECT Id, Title, Content, LastModified 
            FROM KnowledgeTable
        """)

    rows = cursor.fetchall()
    conn.close()

    docs = []
    for row in rows:
        docs.append({
            "id": str(row.Id),
            "title": row.Title,
            "content": row.Content,
            "lastModified": row.LastModified.isoformat()
        })
    return docs

def get_all_sql_ids():
    conn_str = f"DRIVER={SQL_DRIVER};SERVER={SQL_SERVER};DATABASE={SQL_DATABASE};UID={SQL_USERNAME};PWD={SQL_PASSWORD}"
    conn = pyodbc.connect(conn_str)
    cursor = conn.cursor()
    cursor.execute("SELECT Id FROM KnowledgeTable")
    ids = [str(row.Id) for row in cursor.fetchall()]
    conn.close()
    return ids

def get_all_ai_search_ids():
    url = f"{SEARCH_ENDPOINT}/indexes/{SEARCH_INDEX}/docs?api-version=2023-07-01-Preview&$select=id&$top=1000"
    headers = {"api-key": SEARCH_API_KEY}
    ids = []
    while url:
        resp = requests.get(url, headers=headers)
        resp.raise_for_status()
        data = resp.json()
        ids.extend([doc["id"] for doc in data.get("value", [])])
        url = data.get("@odata.nextLink")
    return ids

def delete_from_ai_search(ids):
    if not ids:
        return
    url = f"{SEARCH_ENDPOINT}/indexes/{SEARCH_INDEX}/docs/index?api-version=2023-07-01-Preview"
    headers = {
        "Content-Type": "application/json",
        "api-key": SEARCH_API_KEY
    }
    payload = {
        "value": [{"@search.action": "delete", "id": doc_id} for doc

proceed and finish the auto-deletion sync enhancement code so your scheduled job can both remove stale documents and index new/updated ones in Azure AI Search.

PHASE 11 (continued) — Completing the Function

We left off inside the delete_from_ai_search() function.
Now we’ll complete it, integrate it into the scheduled run, and ensure the last sync time is updated.

Full Updated `scheduled_index/init.py`

PythonCopy codeimport logging
import json
import pyodbc
import requests
from datetime import datetime
import azure.functions as func
from azure.storage.blob import BlobServiceClient

# Azure AI Search config
SEARCH_ENDPOINT = "https://<YOUR-SEARCH-NAME>.search.windows.net"
SEARCH_API_KEY = "<YOUR-SEARCH-ADMIN-KEY>"
SEARCH_INDEX = "knowledge-index"

# SQL config
SQL_SERVER = "<YOUR-SQL-SERVER>.database.windows.net"
SQL_DATABASE = "<YOUR-DATABASE>"
SQL_USERNAME = "<YOUR-USERNAME>"
SQL_PASSWORD = "<YOUR-PASSWORD>"
SQL_DRIVER = "{ODBC Driver 18 for SQL Server}"

# Blob storage config for last sync persistence
BLOB_CONN_STR = "<YOUR-STORAGE-CONNECTION-STRING>"
BLOB_CONTAINER = "function-state"
BLOB_NAME = "last_sync.txt"

def get_last_sync():
    blob_service = BlobServiceClient.from_connection_string(BLOB_CONN_STR)
    blob_client = blob_service.get_blob_client(container=BLOB_CONTAINER, blob=BLOB_NAME)
    try:
        data = blob_client.download_blob().readall().decode("utf-8")
        return datetime.fromisoformat(data)
    except:
        return None

def set_last_sync(timestamp):
    blob_service = BlobServiceClient.from_connection_string(BLOB_CONN_STR)
    blob_client = blob_service.get_blob_client(container=BLOB_CONTAINER, blob=BLOB_NAME)
    blob_client.upload_blob(timestamp.isoformat(), overwrite=True)

def get_sql_data(last_sync=None):
    conn_str = f"DRIVER={SQL_DRIVER};SERVER={SQL_SERVER};DATABASE={SQL_DATABASE};UID={SQL_USERNAME};PWD={SQL_PASSWORD}"
    conn = pyodbc.connect(conn_str)
    cursor = conn.cursor()

    if last_sync:
        cursor.execute("""
            SELECT Id, Title, Content, LastModified 
            FROM KnowledgeTable 
            WHERE LastModified > ?
        """, last_sync)
    else:
        cursor.execute("""
            SELECT Id, Title, Content, LastModified 
            FROM KnowledgeTable
        """)

    rows = cursor.fetchall()
    conn.close()

    docs = []
    for row in rows:
        docs.append({
            "id": str(row.Id),
            "title": row.Title,
            "content": row.Content,
            "lastModified": row.LastModified.isoformat()
        })
    return docs

def get_all_sql_ids():
    conn_str = f"DRIVER={SQL_DRIVER};SERVER={SQL_SERVER};DATABASE={SQL_DATABASE};UID={SQL_USERNAME};PWD={SQL_PASSWORD}"
    conn = pyodbc.connect(conn_str)
    cursor = conn.cursor()
    cursor.execute("SELECT Id FROM KnowledgeTable")
    ids = [str(row.Id) for row in cursor.fetchall()]
    conn.close()
    return ids

def get_all_ai_search_ids():
    url = f"{SEARCH_ENDPOINT}/indexes/{SEARCH_INDEX}/docs?api-version=2023-07-01-Preview&$select=id&$top=1000"
    headers = {"api-key": SEARCH_API_KEY}
    ids = []
    while url:
        resp = requests.get(url, headers=headers)
        resp.raise_for_status()
        data = resp.json()
        ids.extend([doc["id"] for doc in data.get("value", [])])
        url = data.get("@odata.nextLink")
    return ids

def delete_from_ai_search(ids):
    if not ids:
        return
    url = f"{SEARCH_ENDPOINT}/indexes/{SEARCH_INDEX}/docs/index?api-version=2023-07-01-Preview"
    headers = {
        "Content-Type": "application/json",
        "api-key": SEARCH_API_KEY
    }
    payload = {
        "value": [{"@search.action": "delete", "id": doc_id} for doc_id in ids]
    }
    resp = requests.post(url, headers=headers, json=payload)
    resp.raise_for_status()
    logging.info(f"Deleted {len(ids)} stale documents from AI Search.")

def push_to_ai_search(docs):
    if not docs:
        return
    url = f"{SEARCH_ENDPOINT}/indexes/{SEARCH_INDEX}/docs/index?api-version=2023-07-01-Preview"
    headers = {
        "Content-Type": "application/json",
        "api-key": SEARCH_API_KEY
    }

proceed and finish the push_to_ai_search function and the main scheduled job so your auto-deletion + indexing pipeline is complete.

PHASE 11 (final) — Completed Scheduled Indexing with Auto-Deletion

Here’s the rest of the code:

PythonCopy codedef push_to_ai_search(docs):
    if not docs:
        return
    url = f"{SEARCH_ENDPOINT}/indexes/{SEARCH_INDEX}/docs/index?api-version=2023-07-01-Preview"
    headers = {
        "Content-Type": "application/json",
        "api-key": SEARCH_API_KEY
    }
    payload = {
        "value": [{"@search.action": "mergeOrUpload", **doc} for doc in docs]
    }
    resp = requests.post(url, headers=headers, json=payload)
    resp.raise_for_status()
    logging.info(f"Indexed {len(docs)} documents into AI Search.")

def main(mytimer: func.TimerRequest) -> None:
    logging.info("Scheduled SQL → AI Search sync started.")

    try:
        # 1. Get last sync time from Blob Storage
        last_sync = get_last_sync()
        logging.info(f"Last sync time: {last_sync}")

        # 2. Get all IDs from SQL and AI Search
        sql_ids = set(get_all_sql_ids())
        ai_ids = set(get_all_ai_search_ids())

        # 3. Find stale IDs to delete
        stale_ids = list(ai_ids - sql_ids)
        if stale_ids:
            delete_from_ai_search(stale_ids)
        else:
            logging.info("No stale documents to delete.")

        # 4. Get new/updated docs from SQL
        new_docs = get_sql_data(last_sync)
        if new_docs:
            push_to_ai_search(new_docs)
        else:
            logging.info("No new or updated documents to index.")

        # 5. Update last sync time
        set_last_sync(datetime.utcnow())
        logging.info("Sync completed successfully.")

    except Exception as e:
        logging.error(f"Error during scheduled sync: {str(e)}", exc_info=True)

How This Works

Every scheduled run:
- Reads the last sync time from Blob Storage.
- Fetches all IDs from SQL and AI Search.
- Deletes any IDs in AI Search that no longer exist in SQL.
- Fetches only new or updated rows from SQL since the last sync.
- Pushes them into AI Search.
- Updates the last sync timestamp in Blob Storage.
Result:
- AI Search index is always in sync with SQL.
- No stale or orphaned documents remain.
- Incremental updates keep it efficient.

Deployment Notes

Timer Schedule: In function.json, set your desired CRON expression:

JsonCopy code{
  "scriptFile": "__init__.py",
  "bindings": [
    {
      "name": "mytimer",
      "type": "timerTrigger",
      "direction": "in",
      "schedule": "0 0 * * * *"
    }
  ]
}

This example runs every hour.

Security:
- Use Managed Identity for SQL and Blob access.
- Store keys in Azure Key Vault.
- Restrict AI Search admin key to this function only.

Next Step

Now that we have:

Auto-deletion
Incremental indexing
Persistent last sync
Scheduled automation

We can connect this directly to your UI and MCP agent so that:

The UI always queries a fresh index.
The MCP agent can answer questions without worrying about stale data.

If you want, I can now proceed to show you how to wire this scheduled sync into your existing UI and MCP agent config so the frontend can also display last updated time and the agent can confirm freshness before answering.

proceed and wire your scheduled sync into both the UI frontend and the MCP/Foundry agent so everything stays fresh and transparent to the user.

PHASE 12 — UI + MCP Integration for Freshness Awareness

1. UI Frontend Integration

We’ll add a status endpoint to your Azure Functions so the UI can display:

Last updated time
Number of documents in the index

`get_index_status/init.py`

PythonCopy codeimport logging
import json
import azure.functions as func
import requests
from azure.storage.blob import BlobServiceClient
from datetime import datetime

SEARCH_ENDPOINT = "https://<YOUR-SEARCH-NAME>.search.windows.net"
SEARCH_API_KEY = "<YOUR-SEARCH-QUERY-KEY>"
SEARCH_INDEX = "knowledge-index"

BLOB_CONN_STR = "<YOUR-STORAGE-CONNECTION-STRING>"
BLOB_CONTAINER = "function-state"
BLOB_NAME = "last_sync.txt"

def get_last_sync():
    blob_service = BlobServiceClient.from_connection_string(BLOB_CONN_STR)
    blob_client = blob_service.get_blob_client(container=BLOB_CONTAINER, blob=BLOB_NAME)
    try:
        data = blob_client.download_blob().readall().decode("utf-8")
        return datetime.fromisoformat(data)
    except:
        return None

def get_index_count():
    url = f"{SEARCH_ENDPOINT}/indexes/{SEARCH_INDEX}/docs/$count?api-version=2023-07-01-Preview"
    headers = {"api-key": SEARCH_API_KEY}
    resp = requests.get(url, headers=headers)
    resp.raise_for_status()
    return int(resp.text)

def main(req: func.HttpRequest) -> func.HttpResponse:
    logging.info("Index status request received.")
    try:
        last_sync = get_last_sync()
        count = get_index_count()
        return func.HttpResponse(
            json.dumps({
                "lastSync": last_sync.isoformat() if last_sync else None,
                "docCount": count
            }),
            mimetype="application/json",
            status_code=200
        )
    except Exception as e:
        logging.error(f"Error fetching index status: {str(e)}", exc_info=True)
        return func.HttpResponse(
            json.dumps({"error": str(e)}),
            mimetype="application/json",
            status_code=500
        )

2. UI Changes

In your app.js:

JavascriptCopy codeasync function updateStatus() {
  try {
    const res = await fetch("https://<YOUR-FUNCTION-APP>.azurewebsites.net/api/GetIndexStatus");
    const data = await res.json();
    document.getElementById("status").innerText =
      `Last updated: ${data.lastSync || "Never"} | Documents: ${data.docCount}`;
  } catch (err) {
    console.error("Error fetching status:", err);
    document.getElementById("status").innerText = "Status unavailable";
  }
}

// Call on page load
updateStatus();

In your HTML:

HtmlCopy code<div id="status" style="font-size: 0.9em; color: gray;"></div>

3. MCP / Foundry Agent Integration

We’ll register two MCP tools:

query_knowledge_base → Calls QueryAISearch function.
get_index_status → Calls GetIndexStatus function.

Example MCP Tool Config:

JsonCopy code[
  {
    "name": "query_knowledge_base",
    "description": "Searches the Azure AI Search knowledge base for relevant documents",
    "endpoint": "https://<YOUR-FUNCTION-APP>.azurewebsites.net/api/QueryAISearch",
    "method": "GET",
    "parameters": [
      { "name": "q", "type": "string", "required": true }
    ]
  },
  {
    "name": "get_index_status",
    "description": "Returns the last sync time and document count for the knowledge base",
    "endpoint": "https://<YOUR-FUNCTION-APP>.azurewebsites.net/api/GetIndexStatus",
    "method": "GET",
    "parameters": []
  }
]

4. Agent Behavior

With this setup:

Before answering a user’s question, the MCP agent can call get_index_status to confirm the index is fresh.
If the index is stale, the agent can:
- Inform the user.
- Optionally trigger a manual reindex via IndexSqlToAISearch.

5. Benefits

Transparency: Users see when the knowledge base was last updated.
Trust: MCP agent can confirm data

proceed and wire your scheduled sync into both the UI frontend and the MCP/Foundry agent so everything stays fresh and transparent to the user.

PHASE 12 — UI + MCP Integration for Freshness Awareness

1. UI Frontend Integration

We’ll add a status endpoint to your Azure Functions so the UI can display:

Last updated time
Number of documents in the index

`get_index_status/init.py`

PythonCopy codeimport logging
import json
import azure.functions as func
import requests
from azure.storage.blob import BlobServiceClient
from datetime import datetime

SEARCH_ENDPOINT = "https://<YOUR-SEARCH-NAME>.search.windows.net"
SEARCH_API_KEY = "<YOUR-SEARCH-QUERY-KEY>"
SEARCH_INDEX = "knowledge-index"

BLOB_CONN_STR = "<YOUR-STORAGE-CONNECTION-STRING>"
BLOB_CONTAINER = "function-state"
BLOB_NAME = "last_sync.txt"

def get_last_sync():
    blob_service = BlobServiceClient.from_connection_string(BLOB_CONN_STR)
    blob_client = blob_service.get_blob_client(container=BLOB_CONTAINER, blob=BLOB_NAME)
    try:
        data = blob_client.download_blob().readall().decode("utf-8")
        return datetime.fromisoformat(data)
    except:
        return None

def get_index_count():
    url = f"{SEARCH_ENDPOINT}/indexes/{SEARCH_INDEX}/docs/$count?api-version=2023-07-01-Preview"
    headers = {"api-key": SEARCH_API_KEY}
    resp = requests.get(url, headers=headers)
    resp.raise_for_status()
    return int(resp.text)

def main(req: func.HttpRequest) -> func.HttpResponse:
    logging.info("Index status request received.")
    try:
        last_sync = get_last_sync()
        count = get_index_count()
        return func.HttpResponse(
            json.dumps({
                "lastSync": last_sync.isoformat() if last_sync else None,
                "docCount": count
            }),
            mimetype="application/json",
            status_code=200
        )
    except Exception as e:
        logging.error(f"Error fetching index status: {str(e)}", exc_info=True)
        return func.HttpResponse(
            json.dumps({"error": str(e)}),
            mimetype="application/json",
            status_code=500
        )

2. UI Changes

In your app.js:

JavascriptCopy codeasync function updateStatus() {
  try {
    const res = await fetch("https://<YOUR-FUNCTION-APP>.azurewebsites.net/api/GetIndexStatus");
    const data = await res.json();
    document.getElementById("status").innerText =
      `Last updated: ${data.lastSync || "Never"} | Documents: ${data.docCount}`;
  } catch (err) {
    console.error("Error fetching status:", err);
    document.getElementById("status").innerText = "Status unavailable";
  }
}

// Call on page load
updateStatus();

In your HTML:

HtmlCopy code<div id="status" style="font-size: 0.9em; color: gray;"></div>

3. MCP / Foundry Agent Integration

We’ll register two MCP tools:

query_knowledge_base → Calls QueryAISearch function.
get_index_status → Calls GetIndexStatus function.

Example MCP Tool Config:

JsonCopy code[
  {
    "name": "query_knowledge_base",
    "description": "Searches the Azure AI Search knowledge base for relevant documents",
    "endpoint": "https://<YOUR-FUNCTION-APP>.azurewebsites.net/api/QueryAISearch",
    "method": "GET",
    "parameters": [
      { "name": "q", "type": "string", "required": true }
    ]
  },
  {
    "name": "get_index_status",
    "description": "Returns the last sync time and document count for the knowledge base",
    "endpoint": "https://<YOUR-FUNCTION-APP>.azurewebsites.net/api/GetIndexStatus",
    "method": "GET",
    "parameters": []
  }
]

4. Agent Behavior

With this setup:

Before answering a user’s question, the MCP agent can call get_index_status to confirm the index is fresh.
If the index is stale, the agent can:
- Inform the user.
- Optionally trigger a manual reindex via IndexSqlToAISearch.

5. Benefits

Transparency: Users see when the knowledge base was last updated.
Trust: MCP agent can confirm data

proceed and finish the PHASE 12 integration so your UI and MCP/Foundry agent not only know when the index was last updated, but can also act if it’s stale — giving you a fully self-healing, self-aware knowledge base.

PHASE 13 — Freshness-Aware Actions

1. UI: Manual Refresh if Stale

We’ll enhance the UI so that if the last sync time is older than a threshold (e.g., 24 hours), it shows a “Refresh Now” button.

app.js

JavascriptCopy codeasync function updateStatus() {
  try {
    const res = await fetch("https://<YOUR-FUNCTION-APP>.azurewebsites.net/api/GetIndexStatus");
    const data = await res.json();

    const lastSync = data.lastSync ? new Date(data.lastSync) : null;
    const docCount = data.docCount || 0;

    document.getElementById("status").innerText =
      `Last updated: ${lastSync ? lastSync.toLocaleString() : "Never"} | Documents: ${docCount}`;

    // Show refresh button if stale
    if (!lastSync || (Date.now() - lastSync.getTime()) > 24 * 60 * 60 * 1000) {
      document.getElementById("refreshBtn").style.display = "inline-block";
    } else {
      document.getElementById("refreshBtn").style.display = "none";
    }
  } catch (err) {
    console.error("Error fetching status:", err);
    document.getElementById("status").innerText = "Status unavailable";
  }
}

async function manualRefresh() {
  try {
    const res = await fetch("https://<YOUR-FUNCTION-APP>.azurewebsites.net/api/IndexSqlToAISearch");
    const data = await res.json();
    alert(`Manual refresh triggered: ${data.message || "Check logs"}`);
    updateStatus();
  } catch (err) {
    alert("Manual refresh failed.");
  }
}

// Call on page load
updateStatus();

HTML

HtmlCopy code<div id="status" style="font-size: 0.9em; color: gray;"></div>
<button id="refreshBtn" style="display:none;" onclick="manualRefresh()">Refresh Now</button>

2. MCP Agent: Freshness Check Before Query

We’ll make the agent call get_index_status first, and only query if the index is fresh enough.

Example MCP Agent Logic (pseudo-code):

PythonCopy codedef answer_user_question(question):
    status = call_tool("get_index_status")
    last_sync = parse_datetime(status["lastSync"])
    if not last_sync or (datetime.utcnow() - last_sync).days > 1:
        return f"The knowledge base may be outdated (last updated {last_sync}). " \
               f"Would you like me to refresh it before answering?"
    else:
        return call_tool("query_knowledge_base", {"q": question})

3. Optional: Agent-Initiated Refresh

If the user says “yes” to refreshing:

PythonCopy codedef refresh_and_answer(question):
    call_tool("index_sql_to_ai_search")  # Manual reindex
    wait_for_completion()  # Optional: poll until done
    return call_tool("query_knowledge_base", {"q": question})

4. Benefits of This Setup

Self-healing: If the index is stale, the system can refresh automatically.
User trust: Both UI and agent are transparent about data freshness.
Efficiency: Avoids unnecessary reindexing when data is already fresh.
Consistency: UI and agent share the same freshness logic.

✅ Now you have:

Scheduled sync with auto-deletion + incremental indexing.
Persistent last sync tracking in Blob Storage.
UI freshness display + manual refresh option.
MCP agent freshness check before answering.
Optional agent-triggered refresh.

If you want, I can now proceed to the final phase where we add real-time change detection from SQL using Azure Event Grid or Change Data Capture (CDC) — so the index updates immediately when data changes, instead of waiting for the schedule.

That would make your knowledge base truly real-time.

let’s proceed and take your system to the final phase by adding real-time sync from SQL to Azure AI Search using Change Data Capture (CDC) and Azure Event Grid.

This will make your knowledge base update instantly when data changes — no waiting for the scheduled job.

PHASE 14 — Real-Time SQL → AI Search Sync

1. How It Works

Enable Change Data Capture (CDC) in Azure SQL Database.
Azure Data Factory or Azure Functions listens for CDC changes.
Event Grid triggers your indexing function immediately when:
- A row is inserted → Add to AI Search.
- A row is updated → Merge into AI Search.
- A row is deleted → Remove from AI Search.

2. Enabling CDC in Azure SQL

Run in SQL Server Management Studio (SSMS) or Azure Data Studio:

SqlCopy code-- Enable CDC at the database level
EXEC sys.sp_cdc_enable_db;

-- Enable CDC for your table
EXEC sys.sp_cdc_enable_table
    @source_schema = 'dbo',
    @source_name = 'KnowledgeTable',
    @role_name = NULL;

This creates change tables like cdc.dbo_KnowledgeTable_CT that log inserts, updates, and deletes.

3. Azure Function for Real-Time Processing

We’ll create a new function that:

Runs on a Timer Trigger every minute (or Event Grid if using CDC streaming).
Reads the CDC change table.
Pushes changes to AI Search immediately.

`realtime_index/init.py`

PythonCopy codeimport logging
import pyodbc
import requests
import azure.functions as func
from datetime import datetime, timedelta

SEARCH_ENDPOINT = "https://<YOUR-SEARCH-NAME>.search.windows.net"
SEARCH_API_KEY = "<YOUR-SEARCH-ADMIN-KEY>"
SEARCH_INDEX = "knowledge-index"

SQL_SERVER = "<YOUR-SQL-SERVER>.database.windows.net"
SQL_DATABASE = "<YOUR-DATABASE>"
SQL_USERNAME = "<YOUR-USERNAME>"
SQL_PASSWORD = "<YOUR-PASSWORD>"
SQL_DRIVER = "{ODBC Driver 18 for SQL Server}"

def get_cdc_changes():
    conn_str = f"DRIVER={SQL_DRIVER};SERVER={SQL_SERVER};DATABASE={SQL_DATABASE};UID={SQL_USERNAME};PWD={SQL_PASSWORD}"
    conn = pyodbc.connect(conn_str)
    cursor = conn.cursor()

    # Get changes from the last minute
    cursor.execute("""
        DECLARE @from_lsn binary(10), @to_lsn binary(10);
        SET @from_lsn = sys.fn_cdc_get_min_lsn('dbo_KnowledgeTable');
        SET @to_lsn = sys.fn_cdc_map_time_to_lsn('largest less than or equal', GETDATE());

        SELECT *
        FROM cdc.fn_cdc_get_all_changes_dbo_KnowledgeTable(@from_lsn, @to_lsn, 'all');
    """)

    rows = cursor.fetchall()
    conn.close()
    return rows

def push_to_ai_search(action, doc):
    url = f"{SEARCH_ENDPOINT}/indexes/{SEARCH_INDEX}/docs/index?api-version=2023-07-01-Preview"
    headers = {"Content-Type": "application/json", "api-key": SEARCH_API_KEY}
    payload = {"value": [{**doc, "@search.action": action}]}
    resp = requests.post(url, headers=headers, json=payload)
    resp.raise_for_status()

def main(mytimer: func.TimerRequest) -> None:
    logging.info("Real-time CDC sync started.")
    try:
        changes = get_cdc_changes()
        for change in changes:
            op = change.__getattribute__("__$operation")  # CDC operation code
            doc = {
                "id": str(change.Id),
                "title": change.Title,
                "content": change.Content,
                "lastModified": change.LastModified.isoformat()
            }
            if op == 1:  # Delete
                push_to_ai_search("delete", {"id": doc["id"]})
            elif op in (2, 4):  # Insert or Update
                push_to_ai_search("mergeOrUpload", doc)

        logging.info(f"Processed {len(changes)} CDC changes.")
    except Exception as e:
        logging.error(f"Error in real-time sync: {str(e)}", exc_info=True)

4. Event-Driven Alternative

Instead of polling CDC:

Use Azure Data Factory or Azure Synapse Link to stream changes to Event Grid.
Event Grid triggers your Azure Function instantly.
This removes the 1-minute polling delay.

proceed and finish the real-time sync setup by wiring in the Event Grid trigger so your Azure Function reacts instantly to SQL changes without polling.

PHASE 14 (continued) — Event-Driven Real-Time Sync

1. Why Event Grid Instead of Polling

Lower latency — changes propagate in seconds.
Lower cost — no constant database queries.
Scalable — handles bursts of changes without overloading SQL.

2. Architecture Overview

Azure SQL Change Data Capture (CDC) or Change Tracking detects row changes.
Azure Data Factory or Azure Synapse Link streams changes to Azure Storage or Azure Event Hubs.
Event Grid publishes an event when a new change file/message arrives.
Azure Function (Event Grid Trigger) processes the event and updates Azure AI Search.

3. Event Grid Trigger Function

Here’s an example Azure Function that listens for Event Grid events and updates AI Search:

`eventgrid_index/init.py`

PythonCopy codeimport logging
import json
import requests
import azure.functions as func

SEARCH_ENDPOINT = "https://<YOUR-SEARCH-NAME>.search.windows.net"
SEARCH_API_KEY = "<YOUR-SEARCH-ADMIN-KEY>"
SEARCH_INDEX = "knowledge-index"

def push_to_ai_search(action, doc):
    url = f"{SEARCH_ENDPOINT}/indexes/{SEARCH_INDEX}/docs/index?api-version=2023-07-01-Preview"
    headers = {"Content-Type": "application/json", "api-key": SEARCH_API_KEY}
    payload = {"value": [{**doc, "@search.action": action}]}
    resp = requests.post(url, headers=headers, json=payload)
    resp.raise_for_status()
    logging.info(f"{action} action completed for doc ID {doc.get('id')}")

def main(event: func.EventGridEvent):
    logging.info(f"Event received: {event.id}")
    try:
        data = event.get_json()

        # Example: Event contains change type and document payload
        change_type = data.get("operationType")  # e.g., "insert", "update", "delete"
        doc = data.get("document")

        if change_type == "delete":
            push_to_ai_search("delete", {"id": doc["id"]})
        else:
            push_to_ai_search("mergeOrUpload", doc)

    except Exception as e:
        logging.error(f"Error processing Event Grid event: {str(e)}", exc_info=True)

4. Event Grid Subscription Setup

Source:
- If using Azure Data Factory: configure a pipeline that outputs change data to Blob Storage or Event Hubs.
- If using Synapse Link: stream changes to a storage account.
Event Grid Topic:
- Create a topic that listens to Blob Storage or Event Hubs events.
Subscription:
- Create an Event Grid subscription that triggers your Azure Function when a new change event is published.

5. Benefits of Event-Driven Sync

Instant updates — no waiting for scheduled jobs.
Efficient — only processes changes, not the whole dataset.
Scalable — handles high change volumes without constant DB polling.
Unified — works alongside your scheduled sync as a safety net.

6. Recommended Hybrid Approach

Event Grid for real-time changes.
Scheduled sync (hourly/daily) as a catch-up in case any events are missed.
UI + MCP agent still use the GetIndexStatus endpoint to confirm freshness.

–

proceed with a step-by-step testing plan using Azure Functions Core Tools so you can validate your scheduled sync, real-time Event Grid sync, and HTTP API functions before and after deployment.

This will help you catch issues early and confirm that your folder structure + code works end-to-end.

1. Prerequisites

Azure Functions Core Tools installed:BashCopy codefunc --version Should match your runtime (v4 for Python 3.9+).
Azure CLI installed and logged in:BashCopy codeaz login
Local.settings.json configured with:
- SQL connection string
- AI Search endpoint + key
- Blob Storage connection string
- Event Grid test payloads (optional)

2. Local Testing with Core Tools

A. Start the Function App Locally

From your project root:

BashCopy codefunc start

You should see output like:

Copy codeFunctions:

        timer_sync: timerTrigger
        eventgrid_sync: eventGridTrigger
        get_index_status: [GET,POST] http://localhost:7071/api/get_index_status
        manual_refresh: [POST] http://localhost:7071/api/manual_refresh

B. Test HTTP Functions Locally

1. Get Index Status

BashCopy codecurl http://localhost:7071/api/get_index_status

Expected JSON:

JsonCopy code{
  "lastSync": "2026-09-10T12:34:56Z",
  "docCount": 1234
}

2. Trigger Manual Refresh

BashCopy codecurl -X POST http://localhost:7071/api/manual_refresh

Expected:

JsonCopy code{
  "message": "Manual refresh started"
}

C. Test Timer Trigger Locally

You can manually invoke the timer function without waiting for the schedule:

BashCopy codefunc run timer_sync

Expected logs:

Copy code[Information] Fetching last sync time from Blob...
[Information] Found 5 stale docs, deleting...
[Information] Indexed 12 new/updated docs.
[Information] Updated last sync timestamp.

D. Test Event Grid Trigger Locally

Create a sample Event Grid payload in eventgrid_test.json:

JsonCopy code[
  {
    "id": "12345",
    "eventType": "Microsoft.Storage.BlobCreated",
    "subject": "/blobServices/default/containers/test/blobs/doc.json",
    "eventTime": "2026-09-10T12:00:00Z",
    "data": {
      "operationType": "insert",
      "document": {
        "id": "doc-001",
        "title": "Test Document",
        "content": "This is a test",
        "lastModified": "2026-09-10T12:00:00Z"
      }
    },
    "dataVersion": "1.0"
  }
]

Run:

BashCopy codefunc run eventgrid_sync --input "@eventgrid_test.json"

Expected logs:

Copy code[Information] Event received: 12345
[Information] mergeOrUpload action completed for doc ID doc-001

3. Testing After Deployment

A. Get Function URLs

BashCopy codeaz functionapp function list \
  --name <APP_NAME> \
  --resource-group <RESOURCE_GROUP> \
  --query "[].{name:name, url:invokeUrlTemplate}"

B. Test HTTP Functions in Azure

BashCopy codecurl "https://<APP_NAME>.azurewebsites.net/api/get_index_status?code=<FUNCTION_KEY>"

BashCopy codecurl -X POST "https://<APP_NAME>.azurewebsites.net/api/manual_refresh?code=<FUNCTION_KEY>"

C. Trigger Timer Function in Azure

Azure doesn’t let you manually run a timer trigger directly, but you can:

Temporarily change the CRON schedule in function.json to run every minute:

JsonCopy code"schedule": "0 * * * * *"

Redeploy, wait for it to run, then revert.

D. Trigger Event Grid Function in Azure

Go to Azure Portal → Event Grid Topic → Test.
Paste your sample payload.
Send test event.
Check Function App logs:

BashCopy codeaz functionapp log tail --name <APP_NAME> --resource-group <RESOURCE_GROUP>

—

Testing MCP

step-by-step instructions to test your deployed Azure Functions (scheduled sync, real-time Event Grid sync, and HTTP APIs) through your MCP Inspector, so you can confirm that your MCP/Foundry Agent can call them and get the expected results.

I’ll walk you through local MCP Inspector testing and cloud testing after deployment.

1. Prerequisites

MCP Inspector installed and running.
Your MCP Agent configured with:
- The HTTP tool pointing to your Azure Function endpoints.
- Authentication (function keys or managed identity).
Your Function App deployed and running in Azure (or locally via func start).
Function Keys retrieved from Azure Portal → Function → Get Function URL.

2. Local Testing with MCP Inspector

A. Start Functions Locally

BashCopy codefunc start

You should see:

Copy codeFunctions:

        get_index_status: [GET,POST] http://localhost:7071/api/get_index_status
        manual_refresh: [POST] http://localhost:7071/api/manual_refresh

B. Configure MCP Inspector

Open MCP Inspector.
Go to Tools → Add Tool.
Add a new HTTP tool:
- Name: GetIndexStatus
- Method: GET
- URL: http://localhost:7071/api/get_index_status
- Auth: None (for local testing).
Add another HTTP tool:
- Name: ManualRefresh
- Method: POST
- URL: http://localhost:7071/api/manual_refresh
- Auth: None (for local testing).

C. Test in MCP Inspector

Get Index Status:
1. Select GetIndexStatus tool.
2. Click Run.
3. Expected output:JsonCopy code{ "lastSync": "2026-09-10T12:34:56Z", "docCount": 1234 }
Manual Refresh:
1. Select ManualRefresh tool.
2. Click Run.
3. Expected output:JsonCopy code{ "message": "Manual refresh started" }

3. Cloud Testing with MCP Inspector

A. Get Function URLs

BashCopy codeaz functionapp function list \
  --name <APP_NAME> \
  --resource-group <RESOURCE_GROUP> \
  --query "[].{name:name, url:invokeUrlTemplate}"

Example:

Copy codehttps://my-func-app.azurewebsites.net/api/get_index_status?code=abc123...

B. Configure MCP Inspector for Cloud

In MCP Inspector, edit your HTTP tools:
- URL: Replace localhost:7071 with your Azure Function URL.
- Auth: If using function keys, append ?code=<FUNCTION_KEY> to the URL.
Save changes.

C. Run Cloud Tests

Get Index Status:
- Expected output should match Azure’s live index state.
Manual Refresh:
- Should trigger a reindex in Azure AI Search.
- You can confirm by checking Azure Function logs:BashCopy codeaz functionapp log tail --name <APP_NAME> --resource-group <RESOURCE_GROUP>

4. Testing Real-Time Event Grid via MCP

If your MCP Agent has a tool for triggering test events:

Create a JSON payload matching your Event Grid schema.
Call the Event Grid test endpoint from MCP Inspector.
Check logs to confirm the Event Grid-triggered function ran.

5. Pass/Fail Criteria

✅ Pass if:

MCP Inspector can call both HTTP functions and get valid JSON.
Manual refresh triggers indexing in Azure AI Search.
Get Index Status reflects updated lastSync after refresh.
Event Grid test events result in AI Search updates.

❌ Fail if:

MCP Inspector returns HTTP 401/403 → Check function key or auth.
MCP Inspector times out → Check Function App is running and accessible.
Output JSON is empty or incorrect → Check function logic and bindings.

If you want, I can now give you a ready-to-import MCP Inspector tool configuration JSON so you don’t have to manually add each tool — you can just import and start testing.

proceed and give you a ready-to-import MCP Inspector tool configuration JSON so you can instantly test your Azure Functions from MCP Inspector without manually creating each tool.

I’ll make two versions:

Local Testing Config (for func start)
Cloud Testing Config (for deployed Azure Functions)

1. Local MCP Inspector Config (No Auth)

Save this as mcp_tools_local.json and import into MCP Inspector.

JsonCopy code{
  "tools": [
    {
      "name": "GetIndexStatusLocal",
      "type": "http",
      "method": "GET",
      "url": "http://localhost:7071/api/get_index_status",
      "headers": {
        "Content-Type": "application/json"
      },
      "description": "Retrieve the last sync timestamp and document count from the local Azure Function."
    },
    {
      "name": "ManualRefreshLocal",
      "type": "http",
      "method": "POST",
      "url": "http://localhost:7071/api/manual_refresh",
      "headers": {
        "Content-Type": "application/json"
      },
      "description": "Trigger a manual refresh of the index from the local Azure Function."
    }
  ]
}

2. Cloud MCP Inspector Config (With Function Keys)

Save this as mcp_tools_cloud.json and replace placeholders:

<APP_NAME> → Your Azure Function App name
<FUNCTION_KEY> → Your function key from Azure Portal → Function → Get Function URL

JsonCopy code{
  "tools": [
    {
      "name": "GetIndexStatusCloud",
      "type": "http",
      "method": "GET",
      "url": "https://<APP_NAME>.azurewebsites.net/api/get_index_status?code=<FUNCTION_KEY>",
      "headers": {
        "Content-Type": "application/json"
      },
      "description": "Retrieve the last sync timestamp and document count from the deployed Azure Function."
    },
    {
      "name": "ManualRefreshCloud",
      "type": "http",
      "method": "POST",
      "url": "https://<APP_NAME>.azurewebsites.net/api/manual_refresh?code=<FUNCTION_KEY>",
      "headers": {
        "Content-Type": "application/json"
      },
      "description": "Trigger a manual refresh of the index from the deployed Azure Function."
    }
  ]
}

3. How to Import into MCP Inspector

Open MCP Inspector.
Go to Tools → Import Tools.
Select either mcp_tools_local.json or mcp_tools_cloud.json.
The tools will appear in your tool list.

4. Testing Steps in MCP Inspector

A. Local

Run:BashCopy codefunc start
In MCP Inspector, run GetIndexStatusLocal → Expect JSON with lastSync and docCount.
Run ManualRefreshLocal → Expect "message": "Manual refresh started".

B. Cloud

Ensure your Function App is running in Azure:BashCopy codeaz functionapp show --name <APP_NAME> --resource-group <RESOURCE_GROUP> --query state
In MCP Inspector, run GetIndexStatusCloud → Expect live Azure index status.
Run ManualRefreshCloud → Check Azure logs:BashCopy codeaz functionapp log tail --name <APP_NAME> --resource-group <RESOURCE_GROUP>

5. Bonus — Event Grid Test Tool for MCP Inspector

If you want to test real-time sync from MCP Inspector, add this to your config:

JsonCopy code{
  "name": "TestEventGrid",
  "type": "http",
  "method": "POST",
  "url": "https://<EVENT_GRID_TOPIC_ENDPOINT>?api-version=2018-01-01",
  "headers": {
    "Content-Type": "application/json",
    "aeg-sas-key": "<EVENT_GRID_KEY>"
  },
  "body": "[{\"id\":\"12345\",\"eventType\":\"Microsoft.Storage.BlobCreated\",\"subject\":\"/blobServices/default/containers/test/blobs/doc.json\",\"eventTime\":\"2026-09-10T12:00:00Z\",\"data\":{\"operationType\":\"insert\",\"document\":{\"id\":\"doc-001\",\"title\":\"Test Document\",\"content\":\"This is a test\",\"lastModified\":\"2026-09-10T12:00:00Z\"}},\"dataVersion\":\"1.0\"}]",
  "description": "Send a test Event Grid event to trigger the real-time sync

proceed and finish the MCP Inspector testing setup so you can fully validate your Azure Functions + Event Grid + AI Search pipeline end-to-end from inside MCP Inspector.

We’ll now:

Complete the Event Grid test tool config
Show how to import & run it in MCP Inspector
Explain how to verify the results in Azure logs & AI Search
Give you a full “Test Run Script” so you can repeat this anytime

1. Complete Event Grid Test Tool Config (MCP Inspector)

Save this as mcp_tools_eventgrid.json and replace placeholders:

<EVENT_GRID_TOPIC_ENDPOINT> → From Azure Portal → Event Grid Topic → Overview → Topic Endpoint
<EVENT_GRID_KEY> → From Azure Portal → Event Grid Topic → Access Keys → Key 1 or Key 2

JsonCopy code{
  "tools": [
    {
      "name": "TestEventGrid",
      "type": "http",
      "method": "POST",
      "url": "<EVENT_GRID_TOPIC_ENDPOINT>?api-version=2018-01-01",
      "headers": {
        "Content-Type": "application/json",
        "aeg-sas-key": "<EVENT_GRID_KEY>"
      },
      "body": "[{\"id\":\"12345\",\"eventType\":\"Microsoft.Storage.BlobCreated\",\"subject\":\"/blobServices/default/containers/test/blobs/doc.json\",\"eventTime\":\"2026-09-10T12:00:00Z\",\"data\":{\"operationType\":\"insert\",\"document\":{\"id\":\"doc-001\",\"title\":\"Test Document\",\"content\":\"This is a test\",\"lastModified\":\"2026-09-10T12:00:00Z\"}},\"dataVersion\":\"1.0\"}]",
      "description": "Send a test Event Grid event to trigger the real-time sync Azure Function."
    }
  ]
}

2. Import into MCP Inspector

Open MCP Inspector.
Go to Tools → Import Tools.
Select mcp_tools_eventgrid.json.
The tool TestEventGrid will appear in your tool list.

3. Run the Event Grid Test

In MCP Inspector, select TestEventGrid.
Click Run.
Expected HTTP response:JsonCopy code{} (Event Grid usually returns an empty body with HTTP 200 if accepted.)

4. Verify in Azure

Open a terminal and tail the logs for your Event Grid-triggered function:

BashCopy codeaz functionapp log tail \
  --name <APP_NAME> \
  --resource-group <RESOURCE_GROUP>

You should see:

Copy code[Information] Event received: 12345
[Information] mergeOrUpload action completed for doc ID doc-001

5. Confirm in Azure AI Search

Run:

BashCopy codecurl -X POST \
  -H "Content-Type: application/json" \
  -H "api-key: <AI_SEARCH_ADMIN_KEY>" \
  "https://<AI_SEARCH_SERVICE>.search.windows.net/indexes/<INDEX_NAME>/docs/search?api-version=2021-04-30-Preview" \
  -d '{"search": "Test Document"}'

Expected result:

JsonCopy code{
  "value": [
    {
      "id": "doc-001",
      "title": "Test Document",
      "content": "This is a test"
    }
  ]
}

6. Full MCP Inspector Test Run Script

Here’s the order I recommend for a full validation:

Get Index Status (Local or Cloud tool) → Confirm baseline lastSync and docCount.
Manual Refresh → Confirm "Manual refresh started" and watch logs for indexing.
Get Index Status again → Confirm lastSync updated and docCount changed if new docs were added.
Test Event Grid → Send test event from MCP Inspector.
Tail Logs → Confirm Event Grid-triggered function processed the event.
Search AI Index → Confirm the new document is searchable.

7. Why This Works Well

MCP Inspector acts as your agent-side integration tester — no need to switch tools.
You can simulate both manual and automated triggers from one place.
You can chain tools in MCP Inspector to run the full sequence automatically.

–

proceed and give you a single combined MCP Inspector configuration JSON that includes:

Local testing tools (no auth, for func start)
Cloud testing tools (with function keys)
Event Grid test tool (to simulate real-time sync)

This way, you can import one file into MCP Inspector and have everything ready for both local and deployed testing.

1. Combined MCP Inspector Config JSON

Save this as mcp_tools_all.json and replace placeholders:

<APP_NAME> → Your Azure Function App name
<FUNCTION_KEY> → Function key from Azure Portal → Function → Get Function URL
<EVENT_GRID_TOPIC_ENDPOINT> → From Azure Portal → Event Grid Topic → Overview → Topic Endpoint
<EVENT_GRID_KEY> → From Azure Portal → Event Grid Topic → Access Keys → Key 1 or Key 2

JsonCopy code{
  "tools": [
    {
      "name": "GetIndexStatusLocal",
      "type": "http",
      "method": "GET",
      "url": "http://localhost:7071/api/get_index_status",
      "headers": {
        "Content-Type": "application/json"
      },
      "description": "Retrieve the last sync timestamp and document count from the local Azure Function."
    },
    {
      "name": "ManualRefreshLocal",
      "type": "http",
      "method": "POST",
      "url": "http://localhost:7071/api/manual_refresh",
      "headers": {
        "Content-Type": "application/json"
      },
      "description": "Trigger a manual refresh of the index from the local Azure Function."
    },
    {
      "name": "GetIndexStatusCloud",
      "type": "http",
      "method": "GET",
      "url": "https://<APP_NAME>.azurewebsites.net/api/get_index_status?code=<FUNCTION_KEY>",
      "headers": {
        "Content-Type": "application/json"
      },
      "description": "Retrieve the last sync timestamp and document count from the deployed Azure Function."
    },
    {
      "name": "ManualRefreshCloud",
      "type": "http",
      "method": "POST",
      "url": "https://<APP_NAME>.azurewebsites.net/api/manual_refresh?code=<FUNCTION_KEY>",
      "headers": {
        "Content-Type": "application/json"
      },
      "description": "Trigger a manual refresh of the index from the deployed Azure Function."
    },
    {
      "name": "TestEventGrid",
      "type": "http",
      "method": "POST",
      "url": "<EVENT_GRID_TOPIC_ENDPOINT>?api-version=2018-01-01",
      "headers": {
        "Content-Type": "application/json",
        "aeg-sas-key": "<EVENT_GRID_KEY>"
      },
      "body": "[{\"id\":\"12345\",\"eventType\":\"Microsoft.Storage.BlobCreated\",\"subject\":\"/blobServices/default/containers/test/blobs/doc.json\",\"eventTime\":\"2026-09-10T12:00:00Z\",\"data\":{\"operationType\":\"insert\",\"document\":{\"id\":\"doc-001\",\"title\":\"Test Document\",\"content\":\"This is a test\",\"lastModified\":\"2026-09-10T12:00:00Z\"}},\"dataVersion\":\"1.0\"}]",
      "description": "Send a test Event Grid event to trigger the real-time sync Azure Function."
    }
  ]
}

2. How to Import

Open MCP Inspector.
Go to Tools → Import Tools.
Select mcp_tools_all.json.
All five tools will appear in your tool list.

3. Recommended Test Flow in MCP Inspector

Local Test (before deployment)
- Run GetIndexStatusLocal → Confirm baseline.
- Run ManualRefreshLocal → Confirm "Manual refresh started".
- Run GetIndexStatusLocal again → Confirm lastSync updated.
Cloud Test (after deployment)
- Run GetIndexStatusCloud → Confirm live Azure index status.
- Run ManualRefreshCloud → Check Azure logs for indexing.
Event Grid Test
- Run TestEventGrid → Expect HTTP 200.
- Tail logs:BashCopy codeaz functionapp log tail --name <APP_NAME> --resource-group <RESOURCE_GROUP> Confirm event processed.
- Search AI Search index to confirm new doc exists.

1. System Design Overview

2. High-Level Flow Diagram

3. Context Diagram (C4 Level 1)

4. Container Diagram (C4 Level 2)

5. Class Diagram (Core Components)

6. Sequence Flow (Real-Time Event)

7. Sequence Flow — Scheduled Sync

8. Sequence Flow — Real-Time Event

9. Deployment View (C4 Level 3)

10. Component Responsibilities

11. Final Integration Notes

1. Context Diagram (C4 Level 1) — Mermaid

2. Container Diagram (C4 Level 2) — Mermaid

3. Class Diagram — PlantUML

4. Sequence Diagram — Scheduled Sync — PlantUML

5. Sequence Diagram — Real-Time Event — PlantUML

6. Deployment Diagram — PlantUML

How to Generate the Images

Updated Code — function_app.py

Completed function_app.py

Final Completed function_app.py

Final Completed function_app.py

PHASE 1 — Architecture Overview

PHASE 2 — Prerequisites

PHASE 3 — Azure AI Search Setup

PHASE 4 — Azure SQL Database Setup

PHASE 5 — Azure Function App MCP Tool

function_app/__init__.py

PHASE 5 (continued) — Complete Azure Function

function_app/__init__.py (Final Version)

PHASE 6 — Deployment Steps

1. Deployment Strategy Matrix

2. Step-by-Step for Each Method

A. Zip Deploy via Azure CLI

B. GitHub Actions CI/CD

C. VS Code Deploy

D. Local Git Deploy

3. Recommended Folder Structure (Recap)

1. Final Folder Structure

2. Root Files

3. Shared Modules

3. Shared Modules (continued)

4. Function Implementations

A. Timer Trigger — timer_sync/__init__.py

B. Event Grid Trigger — eventgrid_sync/__init__.py

C. HTTP API — Get Index Status — get_index_status/__init__.py

C. HTTP API — Get Index Status (continued)

D. HTTP API — Manual Refresh

E. .funcignore

F. .gitignore

G. How It All Works Together

1. Prerequisites

2. Save This Script as test_pipeline.py

3. How to Run

4. Expected Output

5. Why This Script Helps

4. Post-Deployment Verification Checklist

PHASE 7 — UI Frontend Integration

Example: Trigger Indexing from UI

PHASE 7 (continued) — UI Frontend Integration

1. Trigger Indexing from UI

2. Query Azure AI Search from UI

query_search/__init__.py

3. Query from UI

4. Simple HTML UI

PHASE 8 — How This Works Together

PHASE 8 — Full Workflow

1. End-to-End Flow

2. Deployment Steps

Backend

Frontend

3. Security Best Practices

4. Optional MCP Integration

PHASE 9 — Scheduled Auto-Indexing

1. How It Works

2. Timer Trigger Function Code

scheduled_index/__init__.py

3. Function Schedule

4. Deployment

Updated Code — `function_app.py`

Completed `function_app.py`

Final Completed `function_app.py`

Final Completed `function_app.py`

`function_app/init.py`

`function_app/init.py` (Final Version)

A. Timer Trigger — `timer_sync/init.py`

B. Event Grid Trigger — `eventgrid_sync/init.py`

C. HTTP API — Get Index Status — `get_index_status/init.py`

2. Save This Script as `test_pipeline.py`

`query_search/init.py`

`scheduled_index/init.py`

`scheduled_index/init.py` (Updated)

Full Updated `scheduled_index/init.py`

`get_index_status/init.py`

`get_index_status/init.py`

`realtime_index/init.py`

`eventgrid_index/init.py`

Testing MCP