first, to. develop function apps, install the. commandline —

npm install -g azure-functions-core-tools@4 –unsafe-perm true (windows machine)

I would not simply add a SQL query inside the current MCP functions. I would add a separate SQL → AI Search synchronization pipeline inside the same Function App.

That gives you:

                         Azure SQL
                    ┌────────────────┐
                    │ Tables         │
                    │ Views          │
                    │ Stored queries │
                    └───────┬────────┘
                            │
                     Managed Identity
                            │
                            ▼
                 ┌────────────────────┐
                 │ Function App  │
                 │                    │
                 │ SQL extraction     │
                 │ transformation     │
                 │ validation         │
                 │ security metadata  │
                 │ batching           │
                 │ telemetry          │
                 └─────────┬──────────┘
                           │
                           ▼
                 ┌────────────────────┐
                 │   Azure AI Search  │
                 │                    │
                 │ csl-metadata       │
                 │ staff-letters-new  │
                 │ SQL index     │
                 └─────────┬──────────┘
                           │
                           ▼
                    MCP Tools
                           │
                           ▼
                  Foundry Agent Service

Azure AI Search’s Python SDK supports Entra authentication with DefaultAzureCredential and supports batch merge_or_upload_documents, which is exactly what we need here.

There are also a few bugs/issues in your current code that I would fix at the same time.

1. Issues I found in your existing code

Issue 1 — typo in the index setting

You currently have:

index_name=os.environ.get("AZURE_SEARCH_INDEX_NAE", "csl-metadata")

NAE should almost certainly be:

AZURE_SEARCH_INDEX_NAME

This is important because a typo silently causes your application to use:

csl-metadata

instead of the configured index.

Issue 2 — no SQL connectivity layer

Currently:

Function
   │
   ├── AI Search
   └── Blob

We need:

Function
   │
   ├── AI Search
   ├── Blob
   └── Azure SQL
          │
          ├── tables
          ├── views
          └── stored queries

Issue 3 — no incremental synchronization

We don’t want:

SQL 5 million rows
       ↓
Function
       ↓
AI Search

every five minutes.

We want:

SQL
 ↓
Only rows changed since last successful sync
 ↓
AI Search

For simple SQL-to-Search synchronization, Azure AI Search also has a native SQL indexer with change tracking support.

But your Function App gives you more control over transformation, authorization metadata, joins, enrichment, telemetry and MCP-specific documents, so I think the Function-based pipeline makes sense for your architecture.

2. I recommend this final Function App structure

Your single function_app.py can initially contain everything, but logically divide it into:

function_app.py
│
├── Configuration
│
├── Credentials
│
├── AI Search clients
│
├── Blob functions
│
├── Existing MCP search
│
├── Existing semantic search
│
├── SQL connection
│
├── SQL extraction
│
├── SQL → Search transformation
│
├── Search batch upload
│
├── SQL synchronization
│
├── Timer trigger
│
└── Health/test endpoints

Once tested, I would split these into modules.

3. Required packages

Add these to requirements.txt:

azure-functions
azure-identity
azure-search-documents
azure-storage-blob
pypdf
pyodbc

For example:

azure-functions==1.*
azure-identity==1.*
azure-search-documents==11.*
azure-storage-blob==12.*
pypdf==6.*
pyodbc==5.*

The Search SDK supports SearchClient with DefaultAzureCredential, and the same client supports adding/updating/deleting documents.

4. Environment variables

Add:

AZURE_SEARCH_ENDPOINT
AZURE_SEARCH_INDEX_NAME
AZURE_SEARCH_SEMANTIC_INDEX_NAME
AZURE_SEARCH_SEMANTIC_CONFIGURATION_NAME

SQL_SERVER
SQL_DATABASE
SQL_DRIVER

SQL_SYNC_SCHEDULE
SQL_SYNC_BATCH_SIZE
SQL_SYNC_LOOKBACK_MINUTES

Example:

AZURE_SEARCH_ENDPOINT=https://x-search.search.windows.net

AZURE_SEARCH_INDEX_NAME=csl-metadata

AZURE_SEARCH_SEMANTIC_INDEX_NAME=-data-new

AZURE_SEARCH_SEMANTIC_CONFIGURATION_NAME=staff-letters-new-semantic-configuration

SQL_SERVER=-sql.database.windows.net

SQL_DATABASE=techwyns

SQL_DRIVER=ODBC Driver 18 for SQL Server

SQL_SYNC_SCHEDULE=0 */5 * * * *

SQL_SYNC_BATCH_SIZE=500

SQL_SYNC_LOOKBACK_MINUTES=10

The overlap in SQL_SYNC_LOOKBACK_MINUTES protects against small timing differences between the SQL update timestamp and the Function execution.

For a mature implementation, I would replace the lookback approach with SQL rowversion/change tracking.

5. Azure SQL authentication

Do not put this in your Function App:

SQL username
SQL password

Use:

Function App Managed Identity
            │
            ▼
       Azure SQL

Your Function’s managed identity should have the minimum required SQL permissions.

For example:

CREATE USER [x-function-app]
FROM EXTERNAL PROVIDER;

ALTER ROLE db_datareader
ADD MEMBER [x-function-app];

If you later need the Function to write synchronization state back into SQL, create a separate least-privilege role.

6. Corrected Search client

Replace your existing:

def get_search_client() -> SearchClient:

with:

from functools import lru_cache

@lru_cache(maxsize=1)
def get_credential() -> DefaultAzureCredential:
    return DefaultAzureCredential()


@lru_cache(maxsize=20)
def get_search_client(index_name: str | None = None) -> SearchClient:

    endpoint = os.environ["AZURE_SEARCH_ENDPOINT"]

    index_name = (
        index_name
        or os.environ.get(
            "AZURE_SEARCH_INDEX_NAME",
            "csl-metadata",
        )
    )

    return SearchClient(
        endpoint=endpoint,
        index_name=index_name,
        credential=get_credential(),
    )

This fixes your typo and avoids unnecessarily constructing credentials repeatedly.

7. Add SQL connection support

Add:

import struct
import pyodbc

Then:

SQL_ACCESS_TOKEN = 1256


def get_sql_connection() -> pyodbc.Connection:

    server = os.environ["SQL_SERVER"]
    database = os.environ["SQL_DATABASE"]

    driver = os.environ.get(
        "SQL_DRIVER",
        "ODBC Driver 18 for SQL Server",
    )

    credential = get_credential()

    token = credential.get_token(
        "https://database.windows.net/.default"
    )

    token_bytes = token.token.encode(
        "utf-16-le"
    )

    token_struct = struct.pack(
        f"<I{len(token_bytes)}s",
        len(token_bytes),
        token_bytes,
    )

    connection_string = (
        f"Driver={{{driver}}};"
        f"Server=tcp:{server},1433;"
        f"Database={database};"
        "Encrypt=yes;"
        "TrustServerCertificate=no;"
        "Connection Timeout=30;"
    )

    return pyodbc.connect(
        connection_string,
        attrs_before={
            SQL_ACCESS_TOKEN: token_struct
        },
    )

This allows the Function App to authenticate to Azure SQL through its managed identity.

8. Generic SQL query function

Now add:

def execute_sql_query(
    sql: str,
    parameters: tuple[Any, ...] = (),
) -> list[dict[str, Any]]:

    connection = None

    try:

        connection = get_sql_connection()

        cursor = connection.cursor()

        logging.info(
            "Executing SQL query."
        )

        cursor.execute(
            sql,
            parameters,
        )

        columns = [
            column[0]
            for column in cursor.description
        ]

        rows = cursor.fetchall()

        return [
            dict(zip(columns, row))
            for row in rows
        ]

    except pyodbc.Error:

        logging.exception(
            "Azure SQL query failed."
        )

        raise

    finally:

        if connection:
            connection.close()

This is important because now you can retrieve from:

Tables
Views
Stored procedures
Custom SELECT statements

without duplicating the connection logic.

9. SQL views are ideal for this

For your environment, I strongly recommend creating Search-specific SQL views rather than allowing the Function to understand every underlying relational table.

For example:

CREATE VIEW x.vw_SearchCases
AS
SELECT
    c.CaseId,
    c.Title,
    c.Description,
    c.Status,
    c.CaseType,
    c.Classification,
    c.OwnerDepartment,
    c.CreatedAt,
    c.UpdatedAt
FROM x.Cases c;

Then:

sql = """
SELECT *
FROM x.vw_SearchCases
WHERE UpdatedAt >= DATEADD(
    MINUTE,
    ?,
    SYSUTCDATETIME()
)
"""

with:

lookback = -10

This gives you a clean contract:

x operational database
          │
          ▼
   Search-specific views
          │
          ▼
     Function App
          │
          ▼
      AI Search

10. Create a generic SQL → Search transformer

This is the part I think will make your implementation substantially better.

Add:

def normalize_value(value: Any) -> Any:

    if value is None:
        return None

    if hasattr(value, "isoformat"):
        return value.isoformat()

    if isinstance(value, bytes):
        return value.decode(
            "utf-8",
            errors="replace",
        )

    return value

Then:

def row_to_search_document(
    row: dict[str, Any],
    *,
    id_field: str,
    entity_type: str,
    content_fields: list[str],
    index_fields: dict[str, str] | None = None,
) -> dict[str, Any]:

    index_fields = index_fields or {}

    document_id = str(
        row[id_field]
    )

    document = {
        "id": document_id,
        "entity_type": entity_type,
    }

    for sql_field, search_field in index_fields.items():

        if sql_field in row:

            document[search_field] = (
                normalize_value(
                    row[sql_field]
                )
            )

    content_parts = []

    for field in content_fields:

        value = row.get(field)

        if value is not None:

            content_parts.append(
                f"{field}: {value}"
            )

    document["content"] = (
        "\n".join(content_parts)
    )

    return document

11. Why this is important

Now SQL:

CaseId
Title
Description
Status
Classification
OwnerDepartment

can become:

{
    "id": "CASE-10025",
    "entity_type": "case",
    "title": "Market Manipulation Investigation",
    "content": "Title: Market Manipulation Investigation\nDescription: ...",
    "status": "Open",
    "classification": "Restricted",
    "department": "Enforcement"
}

That’s much better for AI Search and MCP.

12. Batch upload to AI Search

Add:

def upload_search_documents(
    index_name: str,
    documents: list[dict[str, Any]],
    batch_size: int = 500,
) -> dict[str, Any]:

    if not documents:

        return {
            "submitted": 0,
            "succeeded": 0,
            "failed": 0,
            "errors": [],
        }

    client = get_search_client(
        index_name
    )

    total_success = 0
    total_failed = 0
    errors = []

    for start in range(
        0,
        len(documents),
        batch_size,
    ):

        batch = documents[
            start:start + batch_size
        ]

        try:

            results = (
                client.merge_or_upload_documents(
                    documents=batch
                )
            )

            for result in results:

                if result.succeeded:

                    total_success += 1

                else:

                    total_failed += 1

                    errors.append(
                        {
                            "key": result.key,
                            "error": result.error_message,
                        }
                    )

        except Exception as exc:

            logging.exception(
                "AI Search batch failed."
            )

            total_failed += len(batch)

            errors.append(
                {
                    "batch_start": start,
                    "error": str(exc),
                }
            )

    return {
        "submitted": len(documents),
        "succeeded": total_success,
        "failed": total_failed,
        "errors": errors[:100],
    }

merge_or_upload_documents is particularly appropriate because it updates an existing document when its key exists and creates it when it doesn’t.

13. Add a generic SQL table/view exporter

Now we can build the reusable synchronization layer:

def sync_sql_source_to_search(
    *,
    sql_query: str,
    parameters: tuple[Any, ...],
    index_name: str,
    id_field: str,
    entity_type: str,
    content_fields: list[str],
    index_fields: dict[str, str],
) -> dict[str, Any]:

    rows = execute_sql_query(
        sql_query,
        parameters,
    )

    logging.info(
        "SQL source returned %d rows for %s.",
        len(rows),
        entity_type,
    )

    documents = []

    for row in rows:

        try:

            document = row_to_search_document(
                row,
                id_field=id_field,
                entity_type=entity_type,
                content_fields=content_fields,
                index_fields=index_fields,
            )

            documents.append(
                document
            )

        except Exception:

            logging.exception(
                "Failed transforming %s row.",
                entity_type,
            )

    return upload_search_documents(
        index_name=index_name,
        documents=documents,
        batch_size=int(
            os.environ.get(
                "SQL_SYNC_BATCH_SIZE",
                "500",
            )
        ),
    )

14. Now you can synchronize multiple SQL views

For example:

def sync_cases() -> dict[str, Any]:

    lookback = int(
        os.environ.get(
            "SQL_SYNC_LOOKBACK_MINUTES",
            "10",
        )
    )

    return sync_sql_source_to_search(

        sql_query="""
            SELECT *
            FROM Techwyns.vw_SearchCases
            WHERE UpdatedAt >= DATEADD(
                MINUTE,
                ?,
                SYSUTCDATETIME()
            )
        """,

        parameters=(-lookback,),

        index_name=os.environ[
            "AZURE_SEARCH_INDEX_NAME"
        ],

        id_field="CaseId",

        entity_type="case",

        content_fields=[
            "Title",
            "Description",
            "Status",
            "CaseType",
        ],

        index_fields={
            "CaseId": "case_id",
            "Title": "title",
            "Status": "status",
            "CaseType": "case_type",
            "Classification":
                "classification",
            "OwnerDepartment":
                "department",
            "UpdatedAt":
                "last_modified",
        },
    )

15. Add evidence

def sync_evidence() -> dict[str, Any]:

    lookback = int(
        os.environ.get(
            "SQL_SYNC_LOOKBACK_MINUTES",
            "10",
        )
    )

    return sync_sql_source_to_search(

        sql_query="""
            SELECT *
            FROM techwyns.vw_SearchEvidence
            WHERE UpdatedAt >= DATEADD(
                MINUTE,
                ?,
                SYSUTCDATETIME()
            )
        """,

        parameters=(-lookback,),

        index_name=os.environ[
            "AZURE_SEARCH_INDEX_NAME"
        ],

        id_field="EvidenceId",

        entity_type="evidence",

        content_fields=[
            "Title",
            "Description",
            "EvidenceType",
        ],

        index_fields={
            "EvidenceId":
                "evidence_id",

            "CaseId":
                "case_id",

            "Title":
                "title",

            "Description":
                "description",

            "EvidenceType":
                "evidence_type",

            "Classification":
                "classification",

            "StorageUri":
                "source_uri",

            "UpdatedAt":
                "last_modified",
        },
    )

16. Add regulations

def sync_regulations() -> dict[str, Any]:

    lookback = int(
        os.environ.get(
            "SQL_SYNC_LOOKBACK_MINUTES",
            "10",
        )
    )

    return sync_sql_source_to_search(

        sql_query="""
            SELECT *
            FROM tchwyns.vw_SearchRegulations
            WHERE UpdatedAt >= DATEADD(
                MINUTE,
                ?,
                SYSUTCDATETIME()
            )
        """,

        parameters=(-lookback,),

        index_name=os.environ[
            "AZURE_SEARCH_INDEX_NAME"
        ],

        id_field="RegulationId",

        entity_type="regulation",

        content_fields=[
            "Title",
            "Citation",
            "TextContent",
        ],

        index_fields={
            "RegulationId":
                "regulation_id",

            "Title":
                "title",

            "Citation":
                "citation",

            "EffectiveDate":
                "effective_date",

            "Classification":
                "classification",

            "SourceUri":
                "source_uri",

            "UpdatedAt":
                "last_modified",
        },
    )

17. Synchronize everything

Now:

def synchronize_sql_to_search() -> dict[str, Any]:

    logging.info(
        "Starting SQL → AI Search synchronization."
    )

    results = {}

    try:
        results["cases"] = sync_cases()
    except Exception as exc:
        logging.exception(
            "Case synchronization failed."
        )
        results["cases"] = {
            "error": str(exc)
        }

    try:
        results["evidence"] = sync_evidence()
    except Exception as exc:
        logging.exception(
            "Evidence synchronization failed."
        )
        results["evidence"] = {
            "error": str(exc)
        }

    try:
        results["regulations"] = sync_regulations()
    except Exception as exc:
        logging.exception(
            "Regulation synchronization failed."
        )
        results["regulations"] = {
            "error": str(exc)
        }

    return results

I intentionally don’t make one failure stop every other source.

For example:

Cases       SUCCESS
Evidence    SUCCESS
Regulations FAILED
Market      SUCCESS

is much better operationally than:

Cases       SUCCESS
Evidence    SUCCESS
Regulations FAILED
              ↓
           EVERYTHING
              ↓
           STOPS

18. Add the timer trigger

Your Function App currently doesn’t show a timer.

Add:

@app.timer_trigger(
    schedule="%SQL_SYNC_SCHEDULE%",
    arg_name="timer",
    run_on_startup=False,
    use_monitor=True,
)
def sql_to_ai_search_sync(
    timer: func.TimerRequest,
) -> None:

    logging.info(
        "SQL → AI Search timer triggered."
    )

    if timer.past_due:

        logging.warning(
            "SQL → AI Search timer is past due."
        )

    try:

        result = (
            synchronize_sql_to_search()
        )

        logging.info(
            "SQL → AI Search synchronization "
            "completed: %s",
            json.dumps(
                result,
                default=str,
            ),
        )

    except Exception:

        logging.exception(
            "Fatal SQL → AI Search "
            "synchronization failure."
        )

        raise

The Python v2 Functions model supports this decorator-based timer pattern, and Microsoft recommends run_on_startup=False for production.

19. Add a manual test endpoint

For your development environment, this is extremely useful.

@app.route(
    route="admin/sync-sql-search",
    methods=["POST"],
    auth_level=func.AuthLevel.FUNCTION,
)
def manual_sql_search_sync(
    req: func.HttpRequest,
) -> func.HttpResponse:

    logging.info(
        "Manual SQL → Search sync requested."
    )

    try:

        result = (
            synchronize_sql_to_search()
        )

        return func.HttpResponse(
            json.dumps(
                result,
                indent=2,
                default=str,
            ),
            status_code=200,
            mimetype="application/json",
        )

    except Exception as exc:

        logging.exception(
            "Manual synchronization failed."
        )

        return func.HttpResponse(
            json.dumps(
                {
                    "error": str(exc),
                    "error_type":
                        type(exc).__name__,
                }
            ),
            status_code=500,
            mimetype="application/json",
        )

I deliberately recommend FUNCTION authorization rather than anonymous for this endpoint.

Your existing:

app = func.FunctionApp(
    http_auth_level=func.AuthLevel.ANONYMOUS
)

is something I would reconsider for a production techwyns application.

20. Add health check

@app.route(
    route="health",
    methods=["GET"],
    auth_level=func.AuthLevel.ANONYMOUS,
)
def health(
    req: func.HttpRequest,
) -> func.HttpResponse:

    return func.HttpResponse(
        json.dumps(
            {
                "status": "healthy",
                "service":
                    "techwyns-function-app",
            }
        ),
        status_code=200,
        mimetype="application/json",
    )

21. Add SQL connectivity test

I would also add:

@app.route(
    route="admin/test-sql",
    methods=["GET"],
    auth_level=func.AuthLevel.FUNCTION,
)
def test_sql(
    req: func.HttpRequest,
) -> func.HttpResponse:

    connection = None

    try:

        connection = get_sql_connection()

        cursor = connection.cursor()

        cursor.execute(
            "SELECT 1 AS HealthCheck"
        )

        row = cursor.fetchone()

        return func.HttpResponse(
            json.dumps(
                {
                    "sql":
                        "connected",
                    "health_check":
                        row.HealthCheck,
                }
            ),
            status_code=200,
            mimetype="application/json",
        )

    except Exception as exc:

        logging.exception(
            "SQL connectivity test failed."
        )

        return func.HttpResponse(
            json.dumps(
                {
                    "sql":
                        "failed",
                    "error":
                        str(exc),
                }
            ),
            status_code=500,
            mimetype="application/json",
        )

    finally:

        if connection:
            connection.close()

22. Add AI Search connectivity test

@app.route(
    route="admin/test-search",
    methods=["GET"],
    auth_level=func.AuthLevel.FUNCTION,
)
def test_search(
    req: func.HttpRequest,
) -> func.HttpResponse:

    try:

        index_name = os.environ[
            "AZURE_SEARCH_INDEX_NAME"
        ]

        client = get_search_client(
            index_name
        )

        count = (
            client.get_document_count()
        )

        return func.HttpResponse(
            json.dumps(
                {
                    "search":
                        "connected",
                    "index":
                        index_name,
                    "document_count":
                        count,
                }
            ),
            status_code=200,
            mimetype="application/json",
        )

    except Exception as exc:

        logging.exception(
            "AI Search connectivity test failed."
        )

        return func.HttpResponse(
            json.dumps(
                {
                    "search":
                        "failed",
                    "error":
                        str(exc),
                }
            ),
            status_code=500,
            mimetype="application/json",
        )

23. Test individual SQL views

I would add a function specifically for testing the sources:

def test_sql_sources() -> dict[str, Any]:

    tests = {

        "cases": """
            SELECT TOP 1 *
            FROM techwyns.vw_SearchCases
        """,

        "evidence": """
            SELECT TOP 1 *
            FROM techwyns.vw_SearchEvidence
        """,

        "regulations": """
            SELECT TOP 1 *
            FROM techwyns.vw_SearchRegulations
        """,
    }

    results = {}

    for name, query in tests.items():

        try:

            rows = execute_sql_query(
                query
            )

            results[name] = {
                "status":
                    "success",
                "rows":
                    len(rows),
            }

        except Exception as exc:

            logging.exception(
                "SQL source test failed: %s",
                name,
            )

            results[name] = {
                "status":
                    "failed",
                "error":
                    str(exc),
            }

    return results

24. Testing strategy

I would test in this order.

Test 1 — SQL authentication

Call:

GET /api/admin/test-sql

Expected:

{
  "sql": "connected",
  "health_check": 1
}

Test 2 — AI Search

Call:

GET /api/admin/test-search

Expected:

{
  "search": "connected",
  "index": "csl-metadata",
  "document_count": 1234
}

Test 3 — SQL view

Execute:

SELECT TOP 10 *
FROM techwyns.vw_SearchCases;

Verify:

CaseId exists
Title exists
UpdatedAt exists
Classification exists

Test 4 — transformation

Verify:

SQL row
 ↓
Search document

has a valid:

id
entity_type
content

Test 5 — Search upload

Run:

POST /api/admin/sync-sql-search

Expected:

{
  "cases": {
    "submitted": 10,
    "succeeded": 10,
    "failed": 0
  },
  "evidence": {
    "submitted": 25,
    "succeeded": 25,
    "failed": 0
  }
}

Test 6 — Search verification

Query:

GET/Search

or your existing MCP tool.

Verify the SQL-derived record appears.

Test 7 — update test

Change one SQL record:

UPDATE techwyns.Cases
SET
    Title = 'Updated Test Case',
    UpdatedAt = SYSUTCDATETIME()
WHERE CaseId = 'TEST-001';

Run synchronization.

Search:

TEST-001

Verify:

Updated Test Case

appears.

Test 8 — failure test

Temporarily use an invalid SQL view:

FROM techwyns.vw_DoesNotExist

The Function should report:

failed

without taking down the other synchronization jobs.

25. One important production issue: deletes

The implementation above handles:

INSERT
UPDATE

but not:

DELETE

For production, you need deletion handling.

Otherwise:

SQL:
CASE-123 exists

AI Search:
CASE-123 exists

SQL:
CASE-123 deleted

AI Search:
CASE-123 STILL EXISTS

That’s dangerous.

I recommend adding a soft-delete field:

IsDeleted BIT NOT NULL DEFAULT 0

Then export:

{
  "id": "CASE-123",
  "@search.action": "delete"
}

or explicitly call:

client.delete_documents(
    documents=[
        {"id": "CASE-123"}
    ]
)

The Search SDK supports deleting documents from an index.

26. Better: SQL Change Tracking

For your production techwyns implementation, I would ultimately move away from:

UpdatedAt >= DATEADD(...)

and use SQL change tracking or another durable high-water-mark mechanism.

That gives you:

SQL Change Tracking
        │
        ▼
Function
        │
        ├── INSERT
        ├── UPDATE
        └── DELETE
        │
        ▼
Azure AI Search

This avoids relying on a time window.

And if the synchronization function fails:

Run #100
 ↓
SQL changes 1000-1100
 ↓
Function crashes
 ↓
Run #101
 ↓
reprocesses changes
 ↓
Search catches up

That’s the behavior you want.

27. Important issue with your MCP design

Your current MCP search function:

run_staff_letters_search()

is fine for the staff-letter use case.

I would not mix SQL synchronization into that function.

Instead:

                  Function App
                       │
       ┌───────────────┼────────────────┐
       │               │                │
       ▼               ▼                ▼
   MCP Search       SQL Sync         Blob/PDF
       │               │
       ▼               ▼
 AI Search        AI Search

The SQL sync is an ingestion pipeline, not an MCP tool.

The MCP tool should only read/query the resulting index.

28. Recommended techwyns MCP tools after this change

You could then expose:

semantic_search_staff_letters
search__cases
search__evidence
search__regulations
search__market_data
get_staff_letter
get_case
get_evidence

Architecture:

                     techwyns MCP
                         │
       ┌─────────────────┼──────────────────┐
       │                 │                  │
       ▼                 ▼                  ▼
 Staff Letter        techwyns SQL          techwyns operational
   Search             Search               tools
       │                 │                  │
       └────────────┬────┘                  │
                    ▼                       ▼
              Azure AI Search          Azure SQL/API

29. Your Search indexes

I would now have:

Azure AI Search
│
├── staff-letters-new
│     └── existing semantic search
│
├── csl-metadata
│     └── existing metadata
│
└── techwyns-enterprise
      ├── cases
      ├── evidence
      ├── regulations
      └── market observations

Or, if your current csl-metadata index already has the correct schema, you can reuse it rather than create another index.

30. Security metadata is especially important

For the techwyns implementation, I would make every SQL-derived Search document carry:

classification
department
case_id
source_system
source_uri
allowed_groups
allowed_roles
last_modified

For example:

{
  "id": "case-1001",
  "entity_type": "case",
  "title": "Market Investigation",
  "content": "...",
  "case_id": "1001",
  "classification": "techwyns-RESTRICTED",
  "department": "Enforcement",
  "allowed_groups": [
    "techwyns-Enforcement",
    "techwyns-Senior-Investigators"
  ],
  "source_system": "techwyns-SQL",
  "source_uri": "techwyns://case/1001",
  "last_modified": "2026-09-10T21:15:00Z"
}

Then your MCP authorization layer constructs the Search filter.

The LLM never determines these permissions.

31. Your current anonymous Function App

You currently have:

app = func.FunctionApp(
    http_auth_level=func.AuthLevel.ANONYMOUS
)

I would be cautious about leaving the entire application anonymous in production.

Especially because you now want:

SQL
AI Search
MCP
techwyns data
Foundry Agents

A better production topology is:

Foundry / SimpleChat / Teams
             │
             ▼
           APIM
             │
      Entra validation
             │
             ▼
      Private Function App
             │
       ┌─────┴─────┐
       ▼           ▼
      SQL       AI Search

The Function’s MCP endpoint should not be an unrestricted anonymous endpoint.

32. One other correction to your current code

This:

logging.info(
    "MCP context type=%s repr=%s",
    type(context).__name__,
    context
)

could potentially dump sensitive MCP context into Application Insights.

I would change it to:

logging.info(
    "MCP context type=%s",
    type(context).__name__,
)

Likewise:

logging.info(
    "Resolved MCP arguments: %s",
    arguments
)

should be redacted or removed in production.

For techwyns data, don’t routinely put MCP arguments, case content, prompts, tokens, or retrieved evidence into Application Insights.

Log:

correlation_id
tool_name
agent_id
duration
result_count
success
error_type
authorization_decision

rather than the sensitive payload.

33. I would also improve your existing exception handling

Currently:

except Exception as exc:
    logging.exception("Semantic search failed")
    return json.dumps({
        "error": str(exc),
        "error_type": type(exc).__name__
    })

That’s acceptable for development but can expose internal information.

Production:

except ValueError as exc:

    logging.warning(
        "Invalid semantic search request: %s",
        exc,
    )

    return json.dumps({
        "error": "Invalid request",
        "error_type": "validation_error",
    })

except Exception:

    logging.exception(
        "Semantic search failed."
    )

    return json.dumps({
        "error":
            "Internal search service error",
        "error_type":
            "internal_error",
    })

That keeps SQL connection errors, internal URLs, stack details, etc. away from the agent.

34. Final production flow

With these changes, your Function App becomes:

                       ┌──────────────────────┐
                       │      Azure SQL       │
                       │                      │
                       │ Tables               │
                       │ Views                │
                       │ Change Tracking      │
                       └──────────┬───────────┘
                                  │
                                  │ Managed Identity
                                  ▼
                       ┌──────────────────────┐
                       │   techwyns Function App  │
                       │                      │
                       │ SQL extraction       │
                       │ Validation           │
                       │ Transformation       │
                       │ Classification       │
                       │ Security metadata    │
                       │ Batch processing     │
                       │ Error handling       │
                       │ App Insights         │
                       └──────────┬───────────┘
                                  │
                                  │ Entra/RBAC
                                  ▼
                       ┌──────────────────────┐
                       │   Azure AI Search    │
                       │                      │
                       │ csl-metadata         │
                       │ staff-letters-new    │
                       │ techwyns-enterprise      │
                       └──────────┬───────────┘
                                  │
                                  ▼
                         ┌────────────────┐
                         │   techwyns MCP     │
                         │                │
                         │ Search tools   │
                         │ Case tools     │
                         │ Evidence       │
                         │ Regulatory     │
                         └───────┬────────┘
                                 │
                                 ▼
                      ┌──────────────────────┐
                      │ Microsoft Foundry    │
                      │ Agent Service        │
                      │                      │
                      │ techwyns Orchestrator    │
                      │ Research Agent       │
                      │ Investigation Agent  │
                      │ Regulatory Agent     │
                      └──────────────────────┘

My strongest recommendation

For your first implementation, use the code pattern above with SQL views + incremental UpdatedAt synchronization + merge_or_upload_documents + timer trigger + managed identity. Then, once it works end-to-end, upgrade the synchronization mechanism to SQL Change Tracking/rowversion + durable sync state + delete detection.

That gives you a clean path from your current Function App to a production techwyns data ingestion/MCP architecture without disrupting your existing staff-letter semantic search. Microsoft supports exactly the Entra-based Search client pattern and batch document operations we’re using here.

complete function_app.py you can deploy as the replacement for the file you supplied.

I have preserved your existing:

Staff-letter semantic search MCP tool
Staff-letter Blob/PDF retrieval logic
Metadata search/filter logic
Existing MCP argument handling

and added:

Azure SQL connectivity using Managed Identity
SQL tables/views support
Multiple configurable SQL sources
Incremental lookback synchronization
SQL → transformation → AI Search
Batch merge_or_upload_documents
Delete handling via IsDeleted
Retry handling
Per-source exception isolation
Manual SQL→Search synchronization endpoint
SQL health test
AI Search health test
SQL source test
Search verification endpoint
Sync statistics
Safer production logging
Corrected AZURE_SEARCH_INDEX_NAE typo
Credential/client caching
Configuration validation
Protection against arbitrary SQL through the HTTP API
Existing MCP search functionality

Azure AI Search supports DefaultAzureCredential/Entra authentication and merge_or_upload_documents; the identity needs Search Index Data Contributor for indexing and Search Index Data Reader for querying.

Complete `function_app.py`


"""
techwyns Azure Function App
=======================

Capabilities
------------
1. Azure AI Search metadata search
2. Semantic search over techwyns staff letters
3. Staff-letter PDF retrieval from Azure Blob Storage
4. MCP tools for Foundry Agent Service
5. Azure SQL -> Azure AI Search synchronization
6. SQL table/view synchronization
7. Incremental synchronization using UpdatedAt lookback
8. Optional SQL IsDeleted -> AI Search delete handling
9. Batch merge-or-upload into Azure AI Search
10. SQL / AI Search health checks
11. Manual synchronization endpoint
12. Source-level synchronization testing
13. Exception isolation and retry handling

Authentication
--------------
Azure SQL:
    Managed Identity / DefaultAzureCredential

Azure AI Search:
    Managed Identity / DefaultAzureCredential

Blob Storage:
    Managed Identity / DefaultAzureCredential
    or optional connection string

IMPORTANT
---------
SQL synchronization sources are configured through:

    SQL_SYNC_SOURCES_JSON

Example:

[
  {
    "name": "cases",
    "query": "SELECT CaseId, Title, Description, Status, CaseType, Classification, OwnerDepartment, UpdatedAt, IsDeleted FROM techwyns.vw_SearchCases WHERE UpdatedAt >= DATEADD(MINUTE, ?, SYSUTCDATETIME())",
    "id_field": "CaseId",
    "entity_type": "case",
    "index_name": "techwyns-enterprise",
    "content_fields": [
      "Title",
      "Description",
      "Status",
      "CaseType"
    ],
    "field_map": {
      "CaseId": "case_id",
      "Title": "title",
      "Status": "status",
      "CaseType": "case_type",
      "Classification": "classification",
      "OwnerDepartment": "department",
      "UpdatedAt": "last_modified"
    },
    "deleted_field": "IsDeleted"
  }
]

The SQL query is application-controlled configuration.
The MCP/HTTP caller cannot submit arbitrary SQL.
"""

1. `requirements.txt`

Use:

azure-functions
azure-identity
azure-search-documents
azure-storage-blob
pypdf
pyodbc

The current Azure AI Search Python SDK supports SearchClient with Microsoft Entra authentication and document upload/merge/delete operations.

2. The important configuration you need

The code intentionally doesn’t hard-code your unknown SQL schema.

Set these Function App settings:

AZURE_SEARCH_ENDPOINT
AZURE_SEARCH_INDEX_NAME
AZURE_SEARCH_SEMANTIC_INDEX_NAME
AZURE_SEARCH_SEMANTIC_CONFIGURATION_NAME

SQL_SERVER
SQL_DATABASE
SQL_DRIVER

SQL_SYNC_SCHEDULE
SQL_SYNC_BATCH_SIZE
SQL_SYNC_LOOKBACK_MINUTES
SQL_SYNC_RETRIES
SQL_SYNC_RETRY_BASE_SECONDS

For example:

AZURE_SEARCH_ENDPOINT=https://YOUR-SEARCH.search.windows.net

AZURE_SEARCH_INDEX_NAME=csl-metadata

AZURE_SEARCH_SEMANTIC_INDEX_NAME=staff-letters-new

AZURE_SEARCH_SEMANTIC_CONFIGURATION_NAME=staff-letters-new-semantic-configuration

SQL_SERVER=YOUR-SQL.database.windows.net

SQL_DATABASE=techwyns

SQL_DRIVER=ODBC Driver 18 for SQL Server

SQL_SYNC_SCHEDULE=0 */5 * * * *

SQL_SYNC_BATCH_SIZE=500

SQL_SYNC_LOOKBACK_MINUTES=10

SQL_SYNC_RETRIES=3

SQL_SYNC_RETRY_BASE_SECONDS=2

The Search identity needs Search Index Data Contributor to push documents; query-only operations require Search Index Data Reader.

3. Configure your SQL tables/views

This is the most important setting:

SQL_SYNC_SOURCES_JSON

I deliberately made this configurable so you can synchronize different tables and views without modifying/redeploying the Python code.

For example, suppose you have:

techwyns.vw_SearchCases
techwyns.vw_SearchEvidence
techwyns.vw_SearchRegulations
techwyns.vw_SearchMarketObservations

Your setting can contain:

[
  {
    "name": "cases",
    "query": "SELECT CaseId, Title, Description, Status, CaseType, Classification, OwnerDepartment, UpdatedAt, IsDeleted FROM techwyns.vw_SearchCases WHERE UpdatedAt >= DATEADD(MINUTE, ?, SYSUTCDATETIME())",
    "id_field": "CaseId",
    "entity_type": "case",
    "index_name": "techwyns-enterprise",
    "content_fields": [
      "Title",
      "Description",
      "Status",
      "CaseType"
    ],
    "field_map": {
      "CaseId": "case_id",
      "Title": "title",
      "Description": "description",
      "Status": "status",
      "CaseType": "case_type",
      "Classification": "classification",
      "OwnerDepartment": "department",
      "UpdatedAt": "last_modified"
    },
    "deleted_field": "IsDeleted",
    "use_lookback": true
  },

  {
    "name": "evidence",
    "query": "SELECT EvidenceId, CaseId, Title, Description, EvidenceType, Classification, StorageUri, UpdatedAt, IsDeleted FROM techwyns.vw_SearchEvidence WHERE UpdatedAt >= DATEADD(MINUTE, ?, SYSUTCDATETIME())",
    "id_field": "EvidenceId",
    "entity_type": "evidence",
    "index_name": "techwyns-enterprise",
    "content_fields": [
      "Title",
      "Description",
      "EvidenceType"
    ],
    "field_map": {
      "EvidenceId": "evidence_id",
      "CaseId": "case_id",
      "Title": "title",
      "Description": "description",
      "EvidenceType": "evidence_type",
      "Classification": "classification",
      "StorageUri": "source_uri",
      "UpdatedAt": "last_modified"
    },
    "deleted_field": "IsDeleted",
    "use_lookback": true
  },

  {
    "name": "regulations",
    "query": "SELECT RegulationId, Title, Citation, TextContent, EffectiveDate, Classification, SourceUri, UpdatedAt, IsDeleted FROM techwyns.vw_SearchRegulations WHERE UpdatedAt >= DATEADD(MINUTE, ?, SYSUTCDATETIME())",
    "id_field": "RegulationId",
    "entity_type": "regulation",
    "index_name": "techwyns-enterprise",
    "content_fields": [
      "Title",
      "Citation",
      "TextContent"
    ],
    "field_map": {
      "RegulationId": "regulation_id",
      "Title": "title",
      "Citation": "citation",
      "TextContent": "content_text",
      "EffectiveDate": "effective_date",
      "Classification": "classification",
      "SourceUri": "source_uri",
      "UpdatedAt": "last_modified"
    },
    "deleted_field": "IsDeleted",
    "use_lookback": true
  },

  {
    "name": "market-observations",
    "query": "SELECT ObservationId, Instrument, ObservationTime, Price, Volume, OpenInterest, SourceSystem, Classification, UpdatedAt, IsDeleted FROM techwyns.vw_SearchMarketObservations WHERE UpdatedAt >= DATEADD(MINUTE, ?, SYSUTCDATETIME())",
    "id_field": "ObservationId",
    "entity_type": "market_observation",
    "index_name": "techwyns-enterprise",
    "content_fields": [
      "Instrument",
      "ObservationTime",
      "Price",
      "Volume",
      "OpenInterest",
      "SourceSystem"
    ],
    "field_map": {
      "ObservationId": "observation_id",
      "Instrument": "instrument",
      "ObservationTime": "observation_time",
      "Price": "price",
      "Volume": "volume",
      "OpenInterest": "open_interest",
      "SourceSystem": "source_system",
      "Classification": "classification",
      "UpdatedAt": "last_modified"
    },
    "deleted_field": "IsDeleted",
    "use_lookback": true
  }
]

Because this is an application setting, you’ll need to make sure Azure Functions preserves the JSON as a single setting value.

4. Your AI Search index must contain the mapped fields

For the example above, your techwyns-enterprise index needs fields corresponding to the generated documents, such as:

id
entity_type
content
case_id
evidence_id
regulation_id
observation_id
title
description
status
case_type
evidence_type
citation
content_text
instrument
observation_time
price
volume
open_interest
source_system
classification
department
source_uri
effective_date
last_modified

Do not deploy the Function expecting it to automatically create this index. Your existing AI Search index schema must match the documents you’re pushing.

The Search SDK treats an index as persistent JSON document storage and supports push-based document loading, so this Function is acting as your controlled ingestion/push pipeline.

5. SQL views I recommend

Rather than allowing the Function to understand your raw relational model, create controlled search views.

For example:

CREATE VIEW techwyns.vw_SearchCases
AS
SELECT
    CaseId,
    Title,
    Description,
    Status,
    CaseType,
    Classification,
    OwnerDepartment,
    UpdatedAt,
    IsDeleted
FROM techwyns.Cases;

And:

CREATE VIEW techwyns.vw_SearchEvidence
AS
SELECT
    EvidenceId,
    CaseId,
    Title,
    Description,
    EvidenceType,
    Classification,
    StorageUri,
    UpdatedAt,
    IsDeleted
FROM techwyns.Evidence;

This is considerably safer than exposing arbitrary SQL to the MCP server.

6. Managed Identity permissions

Your Function App’s managed identity needs SQL access.

For example:

CREATE USER [YOUR-FUNCTION-APP-NAME]
FROM EXTERNAL PROVIDER;

ALTER ROLE db_datareader
ADD MEMBER [YOUR-FUNCTION-APP-NAME];

Then AI Search:

Function App Managed Identity
             │
             ├── SQL → db_datareader
             │
             └── AI Search
                    └── Search Index Data Contributor

The Azure AI Search RBAC documentation specifically distinguishes the Search Index Data Contributor role for loading data from Search Index Data Reader for querying.

7. Test sequence after deployment

Don’t immediately rely on the timer.

First:

SQL

GET /api/admin/test-sql

Expected:

{
  "status": "connected",
  "utc_time": "2026-09-10T..."
}

Then:

AI Search

GET /api/admin/test-search?index=techwyns-enterprise

Expected:

{
  "status": "connected",
  "index": "techwyns-enterprise",
  "document_count": 123
}

Then:

SQL sources

GET /api/admin/test-sql-sources

Expected:

{
  "status": "success",
  "source_count": 4,
  "sources": {
    "cases": {
      "status": "success",
      "rows": 12
    }
  }
}

Then manually run:

POST /api/admin/sync-sql-search

You should get something like:

{
  "status": "completed",
  "source_count": 4,
  "total_sql_rows": 187,
  "total_documents_succeeded": 185,
  "total_documents_failed": 0,
  "total_documents_deleted": 2
}

8. One important production caveat

The current synchronization uses:

UpdatedAt >= DATEADD(MINUTE, ?, SYSUTCDATETIME())

with the default:

SQL_SYNC_LOOKBACK_MINUTES=10

That’s a safe first deployment, but I would not consider it the final high-scale synchronization mechanism.

For the production techwyns platform, I recommend eventually moving to:

SQL Change Tracking / rowversion
             ↓
Function
             ↓
INSERT / UPDATE / DELETE
             ↓
AI Search

rather than repeatedly scanning a time window.

That becomes especially important once the techwyns database gets large.

9. One more important architectural point

I have not made SQL itself an MCP tool.

The architecture is intentionally:

                 Azure SQL
                    │
                    │
              SQL synchronization
                    │
                    ▼
               AI Search
                    │
                    ▼
                techwyns MCP
                    │
                    ▼
             Foundry Agents

rather than:

Foundry Agent
     │
     ▼
MCP
     │
     ▼
"execute arbitrary SQL"

I strongly recommend keeping it this way.

The agent should be able to ask:

search cases
search evidence
search regulations
search market observations
get case
get evidence

but never:

execute_sql("DROP...")
execute_sql("SELECT anything...")

That separation is particularly important for the governed techwyns architecture.

Final deployment package

At minimum, deploy:

function_app.py
requirements.txt
host.json

with Function App settings:

AZURE_SEARCH_ENDPOINT
AZURE_SEARCH_INDEX_NAME
AZURE_SEARCH_SEMANTIC_INDEX_NAME
AZURE_SEARCH_SEMANTIC_CONFIGURATION_NAME

SQL_SERVER
SQL_DATABASE
SQL_DRIVER

SQL_SYNC_SCHEDULE
SQL_SYNC_BATCH_SIZE
SQL_SYNC_LOOKBACK_MINUTES
SQL_SYNC_RETRIES
SQL_SYNC_RETRY_BASE_SECONDS

SQL_SYNC_SOURCES_JSON

AZURE_STORAGE_ACCOUNT_URL
STAFF_LETTERS_CONTAINER

Your existing staff-letter MCP capability remains intact, while the new pipeline becomes:

Azure SQL tables/views → Function App → controlled transformation → AI Search → techwyns MCP → Microsoft Foundry agents.

This is also compatible with your existing AI Search/Foundry architecture because the Search SDK supports both querying and pushing documents using Entra identity rather than search keys.

Absolutely. I created the complete function_app.py as a single downloadable file.

Download

Download the complete final function_app.py

It includes the existing staff-letter MCP functionality plus:

Azure AI Search semantic search
Standard staff-letter search
Blob/PDF extraction
MCP tools
Azure SQL → Azure AI Search synchronization
Multiple SQL tables/views through SQL_SYNC_SOURCES_JSON
Managed Identity authentication to Azure SQL
Managed Identity authentication to AI Search
Incremental lookback_minutes support
Batch indexing
Retry/backoff
Partial indexing error handling
Soft-delete support
Scheduled SQL → Search synchronization
/api/sql-sync-test
/api/health
Configuration validation
Protection against arbitrary SQL supplied by an agent/user
Corrected AZURE_SEARCH_INDEX_NAME configuration typo

How to deploy multiple Python files into ONE Azure Function App

The important concept is:

One Function App = one deployment project.
You can have many .py files inside that project, but they should be registered through the same FunctionApp object.

Microsoft specifically recommends using Blueprints for larger Python v2 applications so that functions can be organized across multiple Python files and then registered from the main function_app.py.

For your techwyns-AI implementation, I recommend this structure:

techwyns-ai-function-app/
│
├── function_app.py
│
├── blueprints/
│   ├── __init__.py
│   ├── mcp_blueprint.py
│   ├── sql_sync_blueprint.py
│   ├── health_blueprint.py
│   └── admin_blueprint.py
│
├── services/
│   ├── __init__.py
│   ├── search_service.py
│   ├── sql_service.py
│   ├── blob_service.py
│   ├── pdf_service.py
│   └── identity_service.py
│
├── models/
│   ├── __init__.py
│   └── sync_models.py
│
├── shared/
│   ├── __init__.py
│   ├── config.py
│   ├── security.py
│   ├── serialization.py
│   └── logging_utils.py
│
├── tests/
│   ├── __init__.py
│   ├── test_search.py
│   ├── test_sql_sync.py
│   └── test_mcp.py
│
├── host.json
├── requirements.txt
├── .funcignore
├── local.settings.json
└── README.md

This is preferable to creating several independent function_app.py files. Azure’s Python v2 model uses decorators and a central FunctionApp, while blueprints let you split functionality into separate modules.

1. Keep `function_app.py` as the main entry point

Your main file should eventually become relatively small:

import azure.functions as func

from blueprints.mcp_blueprint import bp as mcp_bp
from blueprints.sql_sync_blueprint import bp as sql_bp
from blueprints.health_blueprint import bp as health_bp

app = func.FunctionApp(
    http_auth_level=func.AuthLevel.ANONYMOUS
)

app.register_functions(mcp_bp)
app.register_functions(sql_bp)
app.register_functions(health_bp)

The register_functions() mechanism is supported by the Python Functions programming model.

Important

You do not want:

function_app.py
function_app2.py
function_app3.py

with each file doing:

app = func.FunctionApp()

That creates separate application objects that are not automatically assembled into one Function App.

Instead:

function_app.py
        │
        ├── MCP blueprint
        ├── SQL blueprint
        ├── Health blueprint
        ├── Admin blueprint
        └── other blueprints

2. Example MCP blueprint

Create:

blueprints/mcp_blueprint.py

Example:

import json
import azure.functions as func

from services.search_service import run_semantic_staff_letters_search
from services.blob_service import run_get_staff_letter

bp = func.Blueprint()


@bp.mcp_tool(
    name="semantic_search_staff_letters",
    description="Semantic search over techwyns."
)
def semantic_search_staff_letters(context):

    try:
        args = context or {}

        result = run_semantic_staff_letters_search(
            query=args.get("query"),
            top=args.get("top", 5),
            skip=args.get("skip", 0),
        )

        return json.dumps(result)

    except Exception as exc:
        return json.dumps({
            "error": type(exc).__name__,
            "message": str(exc),
        })

The actual implementation can then live in:

services/search_service.py

This separation becomes very useful when you eventually have:

techwyns Toolbox
      │
      ├── Case tools
      ├── Evidence tools
      ├── Regulatory tools
      ├── Market tools
      ├── Document tools
      ├── Search tools
      └── Communication tools

3. Create the SQL synchronization blueprint

Create:

blueprints/sql_sync_blueprint.py

For example:

import azure.functions as func

from services.sql_sync_service import run_sql_to_search_sync

bp = func.Blueprint()


@bp.timer_trigger(
    schedule="%SQL_SYNC_SCHEDULE%",
    arg_name="timer",
    run_on_startup=False,
    use_monitor=True,
)
def sync_sql_to_ai_search(timer: func.TimerRequest):

    result = run_sql_to_search_sync(
        dry_run=False
    )

    if result["sources_failed"]:
        raise RuntimeError(
            f"{result['sources_failed']} SQL sources failed"
        )

Now your SQL synchronization is independent from MCP.

4. Create the health blueprint

blueprints/health_blueprint.py

import json
import azure.functions as func

bp = func.Blueprint()


@bp.route(
    route="health",
    methods=["GET"],
    auth_level=func.AuthLevel.ANONYMOUS,
)
def health(req: func.HttpRequest):

    return func.HttpResponse(
        json.dumps({
            "status": "healthy",
            "service": "techeyns-ai-function-app"
        }),
        mimetype="application/json",
    )

Then register it:

app.register_functions(health_bp)

5. Move services into `services/`

This is where I recommend you eventually move most of the large code from the downloadable function_app.py.

For example:

services/
├── search_service.py
├── sql_service.py
├── sql_sync_service.py
├── blob_service.py
├── pdf_service.py
├── identity_service.py
└── telemetry_service.py

So:

MCP
 │
 ▼
search_service.py
 │
 ▼
Azure AI Search

and:

SQL Timer
 │
 ▼
sql_sync_service.py
 │
 ├── sql_service.py
 │       │
 │       ▼
 │    Azure SQL
 │
 └── search_service.py
         │
         ▼
    Azure AI Search

This is a much better long-term structure for your platform.

6. `requirements.txt`

At the root:

azure-functions
azure-identity
azure-search-documents
azure-storage-blob
azure-core
pypdf
pyodbc

If you’re using additional MCP packages elsewhere in your implementation, add the appropriate MCP SDK package as well.

The important point is that requirements.txt belongs at the root of the Function App project. Azure Functions uses it to install Python dependencies during deployment/remote build.

7. `host.json`

At the root:

{
  "version": "2.0",
  "logging": {
    "applicationInsights": {
      "samplingSettings": {
        "isEnabled": true
      }
    }
  },
  "extensionBundle": {
    "id": "Microsoft.Azure.Functions.ExtensionBundle",
    "version": "[4.*, 5.0.0)"
  }
}

So Azure sees:

techwyns-ai-function-app/
│
├── host.json                 ← ROOT
├── function_app.py           ← ROOT
├── requirements.txt          ← ROOT
│
├── blueprints/
├── services/
├── models/
└── shared/

The deployment package must have host.json at the root; don’t zip the parent directory around your project.

8. Configure `.funcignore`

Create:

.venv/
venv/
__pycache__/
.pytest_cache/
.git/
.github/
.vscode/
tests/
local.settings.json
*.pyc
*.pyo

Do not exclude:

function_app.py
blueprints/
services/
models/
shared/
requirements.txt
host.json

9. Configure local settings

For local testing:

{
  "IsEncrypted": false,
  "Values": {
    "AzureWebJobsStorage": "UseDevelopmentStorage=true",
    "FUNCTIONS_WORKER_RUNTIME": "python",

    "AZURE_SEARCH_ENDPOINT": "https://<search>.search.windows.net",
    "AZURE_SEARCH_INDEX_NAME": "csl-metadata",
    "AZURE_SEARCH_SEMANTIC_INDEX_NAME": "staff-letters-new",
    "AZURE_SEARCH_SEMANTIC_CONFIGURATION": "default",

    "AZURE_SQL_SERVER": "<server>.database.windows.net",
    "AZURE_SQL_DATABASE": "<database>",
    "AZURE_SQL_ODBC_DRIVER": "ODBC Driver 18 for SQL Server",

    "STAFF_LETTERS_ACCOUNT_URL": "https://<storage>.blob.core.windows.net",
    "STAFF_LETTERS_CONTAINER": "staff-letters",

    "SQL_SYNC_SCHEDULE": "0 */15 * * * *",

    "SQL_SYNC_BATCH_SIZE": "500",
    "SQL_SYNC_SEARCH_RETRIES": "3",
    "SQL_QUERY_TIMEOUT_SECONDS": "120",
    "SQL_CONNECTION_TIMEOUT_SECONDS": "30",

    "SQL_SYNC_SOURCES_JSON": "[]"
  }
}

Do not deploy local.settings.json to Azure. Microsoft explicitly treats it as a local-development configuration file; Azure application settings should be configured separately.

10. Configure your multiple SQL tables/views

Your Function App can synchronize many sources.

For example:

[
  {
    "name": "cases",
    "query": "SELECT CaseId, Title, Status, UpdatedAt FROM .Cases WHERE UpdatedAt >= ?",
    "index_name": "-metadata",
    "id_field": "CaseId",
    "search_key_field": "id",
    "key_prefix": "case:",
    "lookback_minutes": 30,
    "field_map": {
      "CaseId": "caseId",
      "Title": "title",
      "Status": "status",
      "UpdatedAt": "updatedAt"
    }
  },
  {
    "name": "regulations",
    "query": "SELECT RegulationId, Title, Citation, UpdatedAt FROM x.Regulations WHERE UpdatedAt >= ?",
    "index_name": "x-metadata",
    "id_field": "RegulationId",
    "search_key_field": "id",
    "key_prefix": "regulation:",
    "lookback_minutes": 30,
    "field_map": {
      "RegulationId": "regulationId",
      "Title": "title",
      "Citation": "citation",
      "UpdatedAt": "updatedAt"
    }
  }
]

This gives you:

                    Azure SQL
                       │
          ┌────────────┼────────────┐
          ▼            ▼            ▼
       Cases       Regulations    Evidence
          │            │            │
          └────────────┼────────────┘
                       ▼
               SQL Sync Service
                       │
                       ▼
                Azure AI Search
                       │
                       ▼
                 MCP
                       │
                       ▼
             Foundry Agent Service

11. Deploy using Azure Functions Core Tools

From the project root:

cd x-ai-function-app

az login

Then:

func azure functionapp publish <FUNCTION_APP_NAME>

This deploys the entire project, not just function_app.py.

That’s the key point:

You don’t deploy each .py file individually.

You deploy:

function_app.py
+
blueprints/
+
services/
+
models/
+
shared/
+
requirements.txt
+
host.json

as one Function App deployment package.

Microsoft’s Python deployment guidance supports this project-based deployment model.

12. Recommended production deployment

For your environment, I’d use:

GitHub
   │
   ▼
GitHub Actions
   │
   ├── pytest
   ├── lint
   ├── dependency check
   ├── security scan
   ├── configuration validation
   └── package
         │
         ▼
   Azure Function App
         │
         ├── MCP tools
         ├── SQL sync
         ├── Health
         ├── Admin
         └── supporting services

Zip deployment is a recommended Azure Functions deployment technology, and the package should contain the application files with host.json at its root.

13. Configure Azure Function App settings

In:

Azure Portal → Function App → Settings → Environment variables

configure the production values.

At minimum:

AZURE_SEARCH_ENDPOINT
AZURE_SEARCH_INDEX_NAME
AZURE_SEARCH_SEMANTIC_INDEX_NAME
AZURE_SEARCH_SEMANTIC_CONFIGURATION

AZURE_SQL_SERVER
AZURE_SQL_DATABASE
AZURE_SQL_ODBC_DRIVER

STAFF_LETTERS_ACCOUNT_URL
STAFF_LETTERS_CONTAINER

SQL_SYNC_SCHEDULE
SQL_SYNC_SOURCES_JSON

Plus:

SQL_SYNC_BATCH_SIZE
SQL_SYNC_SEARCH_RETRIES
SQL_QUERY_TIMEOUT_SECONDS
SQL_CONNECTION_TIMEOUT_SECONDS

Application settings are exposed to the Python process as environment variables, which is exactly what the code uses.

14. Give the Function App Managed Identity access

For production, don’t put:

SQL username
SQL password
Search API key
Storage account key

inside your Python code.

Use the Function App’s Managed Identity.

Azure SQL

Create the Function App identity as an Entra user in the database and grant the minimum required read permissions.

Conceptually:

CREATE USER [x-ai-function] FROM EXTERNAL PROVIDER;

ALTER ROLE db_datareader
ADD MEMBER [x-ai-function];

The Function App then obtains:

https://database.windows.net/.default

and connects using the Entra access token.

AI Search

Give the Function App managed identity the appropriate Search data role, normally:

Search Index Data Contributor

on the target index/search service as appropriate.

15. Test locally before deploying

From the project root:

func start

You should see functions similar to:

Functions:

    health
    sql-sync-test
    sync_sql_to_ai_search
    semantic_search_staff_letters
    get_staff_letter
    search_staff_letters

Then test:

curl http://localhost:7071/api/health

16. Test SQL without writing to Search

This is particularly important.

Use:

/api/sql-sync-test?dry_run=true

For example:

curl "http://localhost:7071/api/sql-sync-test?dry_run=true&max_rows=10"

That lets you verify:

Azure SQL connectivity
       ↓
query
       ↓
rows
       ↓
mapping
       ↓
document transformation

without pushing the documents into AI Search.

Then test one source:

/api/sql-sync-test?dry_run=true&source=cases&max_rows=10

17. Then perform an actual small synchronization

After the dry run works:

/api/sql-sync-test?dry_run=false&source=cases&max_rows=10

This gives you:

SQL
 ↓
10 rows
 ↓
10 Search documents
 ↓
AI Search

Then verify the index in Azure AI Search.

18. Finally enable the scheduled synchronization

Set:

SQL_SYNC_SCHEDULE=0 */15 * * * *

That means approximately every 15 minutes.

The timer function is:

@app.timer_trigger(
    schedule="%SQL_SYNC_SCHEDULE%",
    arg_name="timer",
    run_on_startup=False,
    use_monitor=True,
)

I intentionally kept:

run_on_startup=False

for production so a Function App restart doesn’t unexpectedly trigger a synchronization. Azure’s Python Functions timer model supports scheduled triggers and configuration through application settings.

Recommended production structure

Once we modularize the downloadable file, I would ultimately make your Function App look like this:

-AI Function App
│
├── function_app.py
│
├── blueprints/
│   ├── mcp.py
│   ├── sql_sync.py
│   ├── health.py
│   ├── admin.py
│   └── ingestion.py
│
├── services/
│   ├── search.py
│   ├── sql.py
│   ├── blob.py
│   ├── document_intelligence.py
│   ├── cosmos.py
│   ├── redis.py
│   ├── memory.py
│   ├── authorization.py
│   └── telemetry.py
│
├── mcp/
│   ├── server.py
│   ├── tools/
│   │   ├── cases.py
│   │   ├── evidence.py
│   │   ├── regulatory.py
│   │   ├── market.py
│   │   ├── documents.py
│   │   └── communications.py
│   └── schemas/
│
├── models/
│
├── shared/
│   ├── config.py
│   ├── security.py
│   ├── identity.py
│   ├── logging.py
│   └── serialization.py
│
├── tests/
│
├── host.json
├── requirements.txt
└── .funcignore

That gives you one deployable Azure Function App, while still allowing dozens of functions, MCP tools, services and supporting modules.

This is also the right direction for the larger Toolbox → Orchestrator → Foundry Agent Service architecture rather than putting everything into a 1,000+ line function_app.py.

One important note: the downloadable file is the consolidated version you asked for. Before production deployment, the SQL_SYNC_SOURCES_JSON mappings must match the actual Azure SQL tables/views and the actual fields in the target AI Search indexes; the code intentionally does not invent those schemas.

I created the complete function_app.py as a single downloadable file.

Download the complete final function_app.py

It includes the existing staff-letter MCP functionality plus:

Azure AI Search semantic search
Standard staff-letter search
Blob/PDF extraction
MCP tools
Azure SQL → Azure AI Search synchronization
Multiple SQL tables/views through SQL_SYNC_SOURCES_JSON
Managed Identity authentication to Azure SQL
Managed Identity authentication to AI Search
Incremental lookback_minutes support
Batch indexing
Retry/backoff
Partial indexing error handling
Soft-delete support
Scheduled SQL → Search synchronization
/api/sql-sync-test
/api/health
Configuration validation
Protection against arbitrary SQL supplied by an agent/user
Corrected AZURE_SEARCH_INDEX_NAME configuration typo

SECTION 2======

recommend Bicep + Flex Consumption + Managed Identity + Application Insights + Log Analytics + secure Storage rather than the legacy Linux Consumption plan. Microsoft currently recommends Flex Consumption for new serverless Function Apps, and Flex supports managed identity and private networking.

One important distinction: Bicep deploys the Azure infrastructure; your Python project is deployed afterward as the Function App code package. Microsoft documents this as separate infrastructure/code deployment steps.

Recommended-AI deployment structure

-ai/
│
├── function_app.py
├── requirements.txt
├── host.json
├── .funcignore
│
├── blueprints/
├── services/
├── models/
├── shared/
└── tests/
│
├── infra/
│   ├── main.bicep
│   ├── parameters/
│   │   ├── dev.bicepparam
│   │   └── prod.bicepparam
│   └── modules/
│       ├── function-app.bicep
│       ├── storage.bicep
│       ├── monitoring.bicep
│       └── roles.bicep

For now, here’s a single complete main.bicep you can deploy directly.

`infra/main.bicep`

targetScope = 'resourceGroup'

@description('Azure region for all resources.')
param location string = resourceGroup().location

@description('Unique name for the Function App.')
param functionAppName string

@description('Globally unique storage account name. 3-24 lowercase letters/numbers.')
param storageAccountName string

@description('Name of the Application Insights resource.')
param applicationInsightsName string

@description('Name of the Log Analytics workspace.')
param logAnalyticsWorkspaceName string

@description('Name of the Flex Consumption hosting plan.')
param hostingPlanName string

@description('Python version used by the Function App.')
@allowed([
  '3.10'
  '3.11'
  '3.12'
  '3.13'
])
param pythonVersion string = '3.11'

@description('Maximum number of Function App instances.')
param maximumInstanceCount int = 40

@description('Memory per Flex Consumption instance.')
@allowed([
  2048
  4096
  8192
])
param instanceMemoryMB int = 2048

@description('SQL Server hostname, for example x-sql.database.windows.net.')
param sqlServerName string

@description('Azure SQL database name.')
param sqlDatabaseName string

@description('Azure AI Search endpoint.')
param searchEndpoint string

@description('Default Azure AI Search index.')
param searchIndexName string = 'csl-metadata'

@description('Semantic staff-letter Azure AI Search index.')
param semanticSearchIndexName string = 'staff-letters-new'

@description('Azure AI Search semantic configuration.')
param semanticSearchConfiguration string = 'default'

@description('Staff letters Blob Storage account URL.')
param staffLettersAccountUrl string

@description('Staff letters Blob Storage container.')
param staffLettersContainer string = 'staff-letters'

@description('SQL synchronization schedule.')
param sqlSyncSchedule string = '0 */15 * * * *'

@description('SQL synchronization batch size.')
param sqlSyncBatchSize int = 500

@description('SQL synchronization retry count.')
param sqlSyncSearchRetries int = 3

@description('SQL query timeout.')
param sqlQueryTimeoutSeconds int = 120

@description('SQL connection timeout.')
param sqlConnectionTimeoutSeconds int = 30

@description('SQL synchronization source configuration. Keep [] until actual SQL mappings are defined.')
param sqlSyncSourcesJson string = '[]'

@description('Whether public network access is enabled. For initial deployment use true; set false after private endpoints/VNet integration are ready.')
param publicNetworkAccess string = 'Enabled'

@description('Optional subnet resource ID for VNet integration. Leave empty for initial deployment.')
param virtualNetworkSubnetId string = ''

@description('Tags applied to all resources.')
param tags object = {
  platform: 'x-AI'
  workload: 'x-ai-function-app'
  managedBy: 'Bicep'
  environment: 'dev'
}


// ============================================================================
// COMMON VARIABLES
// ============================================================================

var storageBlobEndpoint = 'https://${storageAccount.name}.blob.core.windows.net'

var functionAppSettings = [
  {
    name: 'FUNCTIONS_EXTENSION_VERSION'
    value: '~4'
  }
  {
    name: 'FUNCTIONS_WORKER_RUNTIME'
    value: 'python'
  }
  {
    name: 'FUNCTIONS_WORKER_RUNTIME_VERSION'
    value: pythonVersion
  }
  {
    name: 'WEBSITE_RUN_FROM_PACKAGE'
    value: '1'
  }
  {
    name: 'WEBSITE_ENABLE_SYNC_UPDATE_SITE'
    value: 'true'
  }
  {
    name: 'AzureWebJobsStorage__accountName'
    value: storageAccount.name
  }
  {
    name: 'AzureWebJobsStorage__credential'
    value: 'managedidentity'
  }
  {
    name: 'AzureWebJobsStorage__blobServiceUri'
    value: storageBlobEndpoint
  }
  {
    name: 'AzureWebJobsStorage__queueServiceUri'
    value: 'https://${storageAccount.name}.queue.core.windows.net'
  }
  {
    name: 'AzureWebJobsStorage__tableServiceUri'
    value: 'https://${storageAccount.name}.table.core.windows.net'
  }

  // --------------------------------------------------------------------------
  // Application Insights
  // --------------------------------------------------------------------------

  {
    name: 'APPLICATIONINSIGHTS_CONNECTION_STRING'
    value: appInsights.properties.ConnectionString
  }

  {
    name: 'ApplicationInsightsAgent_EXTENSION_VERSION'
    value: '~3'
  }

  // --------------------------------------------------------------------------
  // Azure AI Search
  // --------------------------------------------------------------------------

  {
    name: 'AZURE_SEARCH_ENDPOINT'
    value: searchEndpoint
  }

  {
    name: 'AZURE_SEARCH_INDEX_NAME'
    value: searchIndexName
  }

  {
    name: 'AZURE_SEARCH_SEMANTIC_INDEX_NAME'
    value: semanticSearchIndexName
  }

  {
    name: 'AZURE_SEARCH_SEMANTIC_CONFIGURATION'
    value: semanticSearchConfiguration
  }

  // --------------------------------------------------------------------------
  // Azure SQL
  // --------------------------------------------------------------------------

  {
    name: 'AZURE_SQL_SERVER'
    value: sqlServerName
  }

  {
    name: 'AZURE_SQL_DATABASE'
    value: sqlDatabaseName
  }

  {
    name: 'AZURE_SQL_ODBC_DRIVER'
    value: 'ODBC Driver 18 for SQL Server'
  }

  // --------------------------------------------------------------------------
  // Staff letters Blob
  // --------------------------------------------------------------------------

  {
    name: 'STAFF_LETTERS_ACCOUNT_URL'
    value: staffLettersAccountUrl
  }

  {
    name: 'STAFF_LETTERS_CONTAINER'
    value: staffLettersContainer
  }

  // --------------------------------------------------------------------------
  // SQL -> Search synchronization
  // --------------------------------------------------------------------------

  {
    name: 'SQL_SYNC_SCHEDULE'
    value: sqlSyncSchedule
  }

  {
    name: 'SQL_SYNC_BATCH_SIZE'
    value: string(sqlSyncBatchSize)
  }

  {
    name: 'SQL_SYNC_SEARCH_RETRIES'
    value: string(sqlSyncSearchRetries)
  }

  {
    name: 'SQL_QUERY_TIMEOUT_SECONDS'
    value: string(sqlQueryTimeoutSeconds)
  }

  {
    name: 'SQL_CONNECTION_TIMEOUT_SECONDS'
    value: string(sqlConnectionTimeoutSeconds)
  }

  {
    name: 'SQL_SYNC_SOURCES_JSON'
    value: sqlSyncSourcesJson
  }

  // --------------------------------------------------------------------------
  // Search protection
  // --------------------------------------------------------------------------

  {
    name: 'SEMANTIC_SEARCH_MAX_TOP'
    value: '50'
  }

  {
    name: 'STANDARD_SEARCH_MAX_TOP'
    value: '100'
  }
]


// ============================================================================
// LOG ANALYTICS
// ============================================================================

resource logAnalytics 'Microsoft.OperationalInsights/workspaces@2023-09-01' = {
  name: logAnalyticsWorkspaceName
  location: location
  tags: tags
  properties: {
    sku: {
      name: 'PerGB2018'
    }
    retentionInDays: 30
    features: {
      enableLogAccessUsingOnlyResourcePermissions: true
    }
  }
}


// ============================================================================
// APPLICATION INSIGHTS
// ============================================================================

resource appInsights 'Microsoft.Insights/components@2020-02-02' = {
  name: applicationInsightsName
  location: location
  tags: tags
  kind: 'web'
  properties: {
    Application_Type: 'web'
    WorkspaceResourceId: logAnalytics.id
    publicNetworkAccessForIngestion: publicNetworkAccess
    publicNetworkAccessForQuery: publicNetworkAccess
  }
}


// ============================================================================
// FUNCTION STORAGE
// ============================================================================

resource storageAccount 'Microsoft.Storage/storageAccounts@2023-05-01' = {
  name: storageAccountName
  location: location
  tags: tags
  kind: 'StorageV2'
  sku: {
    name: 'Standard_LRS'
  }
  properties: {
    minimumTlsVersion: 'TLS1_2'
    supportsHttpsTrafficOnly: true
    allowBlobPublicAccess: false
    publicNetworkAccess: publicNetworkAccess
    accessTier: 'Hot'
    allowSharedKeyAccess: false
    defaultToOAuthAuthentication: true

    networkAcls: {
      defaultAction: publicNetworkAccess == 'Enabled'
        ? 'Allow'
        : 'Deny'
      bypass: 'AzureServices'
    }
  }
}


// ============================================================================
// STORAGE CONTAINERS
// ============================================================================

resource deploymentContainer 'Microsoft.Storage/storageAccounts/blobServices/containers@2023-05-01' = {
  name: '${storageAccount.name}/default/function-releases'
  properties: {
    publicAccess: 'None'
  }
}


// ============================================================================
// FLEX CONSUMPTION PLAN
// ============================================================================

resource hostingPlan 'Microsoft.Web/serverfarms@2024-04-01' = {
  name: hostingPlanName
  location: location
  tags: tags
  sku: {
    name: 'FC1'
    tier: 'FlexConsumption'
  }
  kind: 'functionapp'
  properties: {
    reserved: true
  }
}


// ============================================================================
// FUNCTION APP
// ============================================================================

resource functionApp 'Microsoft.Web/sites@2024-11-01' = {
  name: functionAppName
  location: location
  tags: tags

  kind: 'functionapp,linux'

  identity: {
    type: 'SystemAssigned'
  }

  properties: {
    serverFarmId: hostingPlan.id

    httpsOnly: true

    publicNetworkAccess: publicNetworkAccess

    siteConfig: {
      minTlsVersion: '1.2'

      ftpsState: 'Disabled'

      alwaysOn: false

      appSettings: functionAppSettings

      cors: {
        allowedOrigins: [
          'https://portal.azure.com'
        ]
        supportCredentials: false
      }
    }

    functionAppConfig: {
      deployment: {
        storage: {
          type: 'blobContainer'
          value: '${storageBlobEndpoint}/function-releases'
          authentication: {
            type: 'SystemAssignedIdentity'
          }
        }
      }

      runtime: {
        name: 'python'
        version: pythonVersion
      }

      scaleAndConcurrency: {
        maximumInstanceCount: maximumInstanceCount
        instanceMemoryMB: instanceMemoryMB
        alwaysReady: {
          name: 'alwaysReady'
          instanceCount: 0
        }
        triggers: {
          http: {
            perInstanceConcurrency: 16
          }
        }
      }

      siteUpdateStrategy: {
        type: 'Rolling'
      }
    }
  }

  dependsOn: [
    deploymentContainer
    appInsights
    hostingPlan
  ]
}


// ============================================================================
// OPTIONAL VNET INTEGRATION
// ============================================================================

resource virtualNetworkIntegration 'Microsoft.Web/sites/virtualNetworkConnections@2024-11-01' = if (!empty(virtualNetworkSubnetId)) {
  name: '${functionApp.name}/virtualNetwork'
  properties: {
    subnetResourceId: virtualNetworkSubnetId
  }
}


// ============================================================================
// OUTPUTS
// ============================================================================

output functionAppName string = functionApp.name

output functionAppResourceId string = functionApp.id

output functionAppPrincipalId string = functionApp.identity.principalId

output functionAppHostname string = functionApp.properties.defaultHostName

output functionAppUrl string = 'https://${functionApp.properties.defaultHostName}'

output applicationInsightsName string = appInsights.name

output logAnalyticsWorkspaceName string = logAnalytics.name

output storageAccountName string = storageAccount.name

output hostingPlanName string = hostingPlan.name

output deploymentContainerName string = deploymentContainer.name

This follows Microsoft’s current Flex Consumption infrastructure model, including the FC1 plan, deployment storage, Function App configuration, managed identity, and monitoring resources.

2. Create `prod.bicepparam`

I recommend not putting all production values directly into main.bicep.

Create:

infra/parameters/prod.bicepparam

using '../main.bicep'

param location = 'East US'

param functionAppName = 'x-ai-func-prod'

param storageAccountName = 'aifuncprod01'

param applicationInsightsName = 'x-ai-appins-prod'

param logAnalyticsWorkspaceName = 'x-ai-law-prod'

param hostingPlanName = 'x-ai-flex-prod'

param pythonVersion = '3.11'

param maximumInstanceCount = 40

param instanceMemoryMB = 4096

param sqlServerName = 'x-sql-prod.database.windows.net'

param sqlDatabaseName = 'x'

param searchEndpoint = 'https://<YOUR-SEARCH-SERVICE>.search.windows.net'

param searchIndexName = 'csl-metadata'

param semanticSearchIndexName = 'staff-letters-new'

param semanticSearchConfiguration = 'default'

param staffLettersAccountUrl = 'https://<YOUR-STORAGE>.blob.core.windows.net'

param staffLettersContainer = 'staff-letters'

param sqlSyncSchedule = '0 */15 * * * *'

param sqlSyncBatchSize = 500

param sqlSyncSearchRetries = 3

param sqlQueryTimeoutSeconds = 120

param sqlConnectionTimeoutSeconds = 30

param sqlSyncSourcesJson = '[]'

param publicNetworkAccess = 'Enabled'

param virtualNetworkSubnetId = ''

param tags = {
  platform: 'x-AI'
  environment: 'prod'
  owner: 'x'
  workload: 'AI-Platform'
  managedBy: 'Bicep'
}

For the initial deployment, I would leave:

publicNetworkAccess = 'Enabled'

and then move the Function App/Storage/Application Insights/Search/SQL architecture toward your private-endpoint / zero-trust CC topology once the VNet/subnets/private DNS are ready.

3. Validate the Bicep

From the repository root:

az login

Then:

az account set --subscription "<SUBSCRIPTION_ID>"

Validate:

az bicep build \
  --file infra/main.bicep

Then run a resource-group what-if:

az deployment group what-if \
  --resource-group "<RESOURCE_GROUP>" \
  --template-file infra/main.bicep \
  --parameters @infra/parameters/prod.bicepparam

I strongly recommend using what-if before production deployment.

4. Deploy the infrastructure

az deployment group create \
  --resource-group "<RESOURCE_GROUP>" \
  --template-file infra/main.bicep \
  --parameters @infra/parameters/prod.bicepparam

This creates:

x-AI Resource Group
│
├── Function App
│
├── Flex Consumption Plan
│
├── Storage Account
│   └── function-releases
│
├── Application Insights
│
└── Log Analytics

Microsoft’s current Bicep guidance follows this same infrastructure-first approach.

5. Get the Function App Managed Identity

After deployment:

az functionapp identity show \
  --name x-ai-func-prod \
  --resource-group "<RESOURCE_GROUP>"

You’ll get something like:

{
  "principalId": "...",
  "tenantId": "...",
  "type": "SystemAssigned"
}

The important value is:

principalId

That is the identity your Python code uses through:

DefaultAzureCredential()

6. Give the Function App access to Azure AI Search

Your Function App needs appropriate data-plane permissions on your Search service.

For your SQL → Search pipeline, use:

Search Index Data Contributor

For read-only MCP search consumers, use:

Search Index Data Reader

The architecture becomes:

                 Entra ID
                    │
                    ▼
          Function App
          Managed Identity
             │          │
             │          │
             ▼          ▼
       Azure SQL     AI Search
       db_datareader  Data Contributor

No Search API key needs to be embedded in function_app.py.

7. Give the Function App access to Azure SQL

In Azure SQL:

CREATE USER [x-ai-func-prod] FROM EXTERNAL PROVIDER;

ALTER ROLE db_datareader
ADD MEMBER [x-ai-func-prod];

For the SQL → Search application I recommend read-only SQL permissions.

Your Python code already requests the Azure SQL token using:

credential.get_token(
    "https://database.windows.net/.default"
)

So the authentication chain becomes:

Function App
     │
     │ Managed Identity
     ▼
Microsoft Entra ID
     │
     │ access token
     ▼
Azure SQL

8. Give it Blob access

For your staff-letter PDFs:

Storage Blob Data Reader

should be assigned to the Function App identity on the staff-letter Storage Account/container.

That lets:

BlobServiceClient(
    account_url=...,
    credential=get_credential()
)

work without a storage account key.

9. Application Insights is already wired

The Bicep creates:

Log Analytics
       │
       ▼
Application Insights
       │
       ▼
Function App

and injects:

APPLICATIONINSIGHTS_CONNECTION_STRING

into the Function App.

This is the foundation for the broader telemetry design we’ve been building:

Function App
      │
      ├── MCP calls
      ├── SQL synchronization
      ├── AI Search calls
      ├── Blob operations
      ├── exceptions
      └── performance
             │
             ▼
      Application Insights
             │
             ▼
       Log Analytics
             │
             ├── Azure Monitor
             ├── Grafana
             └── dashboards

10. Deploy your Python code AFTER Bicep

This is important.

Bicep creates:

Azure infrastructure

Your Python deployment creates:

function_app.py
blueprints/
services/
models/
shared/
requirements.txt
host.json

For Flex Consumption, Microsoft currently uses One Deploy as the deployment mechanism.

From your project root:

func azure functionapp publish x-ai-func-prod

That publishes the entire Function App project, not only function_app.py.

So:

function_app.py
services/
blueprints/
models/
shared/
requirements.txt
host.json

all travel together.

11. Recommended deployment pipeline

For x-AI, I’d make the deployment:

                 GitHub
                    │
                    ▼
              Pull Request
                    │
          ┌─────────┴─────────┐
          ▼                   ▼
       pytest              Bicep lint
          │                   │
          └─────────┬─────────┘
                    ▼
              Security Scan
                    │
                    ▼
              Bicep What-If
                    │
                    ▼
            Infrastructure Deploy
                    │
                    ▼
              30 sec wait
                    │
                    ▼
              Python Deploy
                    │
                    ▼
              Smoke Tests
                    │
                    ▼
             Health Endpoint
                    │
                    ▼
          SQL dry-run validation
                    │
                    ▼
        AI Search synchronization

Microsoft specifically recommends separating infrastructure deployment and code deployment for Function Apps; their current guidance also notes that application-setting changes can cause a restart, so allowing a short interval before code deployment is prudent.

One change I recommend for your production architecture

The Bicep above is the baseline deployment. For the final FedRAMP/private environment, I would evolve it to:

                         Network
                              │
                 ┌────────────┴────────────┐
                 │                         │
           Private DNS                 NSGs
                 │                         │
                 └────────────┬────────────┘
                              │
                        VNet / Subnets
                              │
                  ┌───────────┴───────────┐
                  │                       │
          Function App                APIM
          Managed Identity              │
                  │                     │
       ┌──────────┼──────────┐          │
       ▼          ▼          ▼          ▼
    Azure SQL  AI Search   Blob     MCP
       │          │          │          │
       └──────────┴──────────┴──────────┘
                              │
                         Foundry Agent
                           Service
                              │
                     Orchestrator

That is the version I would use for the production -AI / Foundry MCP architecture, rather than exposing the Function App directly to the public internet.

Also, Microsoft’s current Flex Consumption model explicitly supports private networking, which makes it a better foundation for that target architecture than the legacy Consumption model.

If you want the infrastructure fully productionized, the next logical step is to split this into main.bicep + function-app.bicep + roles.bicep + private-endpoints.bicep + monitoring.bicep + search.bicep + sql.bicep, with Entra/managed-identity RBAC, Private Endpoints, Private DNS, APIM, Key Vault, Redis, Cosmos DB, AI Search, App Insights/Grafana, and Foundry MCP all deployed from the same-AI IaC repository.

MCP enabled Function app integrating with SQL and AI-Search – bicep files included

1. Issues I found in your existing code

Issue 1 — typo in the index setting

Issue 2 — no SQL connectivity layer

Issue 3 — no incremental synchronization

2. I recommend this final Function App structure

3. Required packages

4. Environment variables

5. Azure SQL authentication

6. Corrected Search client

7. Add SQL connection support

8. Generic SQL query function

9. SQL views are ideal for this

10. Create a generic SQL → Search transformer

11. Why this is important

12. Batch upload to AI Search

13. Add a generic SQL table/view exporter

14. Now you can synchronize multiple SQL views

15. Add evidence

16. Add regulations

17. Synchronize everything

18. Add the timer trigger

19. Add a manual test endpoint

20. Add health check

21. Add SQL connectivity test

22. Add AI Search connectivity test

23. Test individual SQL views

24. Testing strategy

Test 1 — SQL authentication

Test 2 — AI Search

Test 3 — SQL view

Test 4 — transformation

Test 5 — Search upload

Test 6 — Search verification

Test 7 — update test

Test 8 — failure test

25. One important production issue: deletes

26. Better: SQL Change Tracking

27. Important issue with your MCP design

28. Recommended techwyns MCP tools after this change

29. Your Search indexes

30. Security metadata is especially important

31. Your current anonymous Function App

32. One other correction to your current code

33. I would also improve your existing exception handling

34. Final production flow

My strongest recommendation

Complete function_app.py

1. requirements.txt

2. The important configuration you need

3. Configure your SQL tables/views

4. Your AI Search index must contain the mapped fields

5. SQL views I recommend

6. Managed Identity permissions

7. Test sequence after deployment

SQL

AI Search

SQL sources

8. One important production caveat

9. One more important architectural point

Final deployment package

Download

How to deploy multiple Python files into ONE Azure Function App

1. Keep function_app.py as the main entry point

Important

2. Example MCP blueprint

3. Create the SQL synchronization blueprint

4. Create the health blueprint

5. Move services into services/

6. requirements.txt

7. host.json

8. Configure .funcignore

9. Configure local settings

10. Configure your multiple SQL tables/views

11. Deploy using Azure Functions Core Tools

12. Recommended production deployment

13. Configure Azure Function App settings

14. Give the Function App Managed Identity access

Azure SQL

AI Search

Complete `function_app.py`

1. `requirements.txt`

1. Keep `function_app.py` as the main entry point

5. Move services into `services/`

6. `requirements.txt`

7. `host.json`

8. Configure `.funcignore`

`infra/main.bicep`

2. Create `prod.bicepparam`