Robotics, Maths, Artificial Intelligence, Engineering, Science and Technology Lab

Add us on Social Media

Instructions to setup mcp server with function app for microsoft foundry agent tools with app insights

admin

|

|

walk through this in a clear, step-by-step way so you can set up a new MCP (Microsoft Foundry Custom Plugin) using an Azure Function App, integrate it with Microsoft Foundry Tools, and enable Application Insights for monitoring.

I’ll break it into three main phases:

  1. Azure Function App Setup
  2. MCP (Custom Tool) Registration in Foundry
  3. Application Insights Integration

1. Create and Configure the Azure Function App

Step 1 — Create a Resource Group

az group create --name my-foundry-mcp-rg --location eastus

Step 2 — Create a Storage Account

BashCopy codeaz storage account create \
  --name mystorageacctmcp \
  --location eastus \
  --resource-group my-foundry-mcp-rg \
  --sku Standard_LRS

Step 3 — Create the Function App

az functionapp create \
  --resource-group my-foundry-mcp-rg \
  --consumption-plan-location eastus \
  --runtime python \
  --runtime-version 3.10 \
  --functions-version 4 \
  --name my-foundry-mcp-func \
  --storage-account mystorageacctmcp

2. Build the MCP Logic in the Function App

Example: HttpTrigger Function for File Retrieval

PythonCopy code

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

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

    source_type = req.params.get('sourceType')
    if not source_type:
        return func.HttpResponse("Missing 'sourceType' parameter", status_code=400)

    results = []

    if source_type.lower() == "blob":
        blob_service_client = BlobServiceClient.from_connection_string(
            "<BLOB_CONNECTION_STRING>"
        )
        container_client = blob_service_client.get_container_client("my-container")
        for blob in container_client.list_blobs():
            results.append({
                "sourceType": "Blob",
                "fileName": blob.name.split("/")[-1],
                "filePath": blob.name
            })

    elif source_type.lower() == "sharepoint":
        # Placeholder for SharePoint retrieval logic
        results.append({
            "sourceType": "SharePoint",
            "fileName": "example.docx",
            "filePath": "/Shared Documents/example.docx"
        })

    return func.HttpResponse(
        json.dumps(results),
        mimetype="application/json",
        status_code=200
    )

Deploy this function:

func azure functionapp publish my-foundry-mcp-func

3. Enable Application Insights

Step 1 — Create Application Insights

BashCopy codeaz monitor app-insights component create \
  --app my-foundry-mcp-ai \
  --location eastus \
  --resource-group my-foundry-mcp-rg

Step 2 — Link to Function App

az functionapp update \
  --name my-foundry-mcp-func \
  --resource-group my-foundry-mcp-rg \
  --set applicationInsights.key=<YOUR_APP_INSIGHTS_KEY>

Step 3 — Add Telemetry in Code

from opencensus.ext.azure.log_exporter import AzureLogHandler

logger = logging.getLogger(__name__)
logger.addHandler(AzureLogHandler(
    connection_string='InstrumentationKey=<YOUR_APP_INSIGHTS_KEY>')
)

logger.warning("Custom MCP function executed.")

4. Register the MCP in Microsoft Foundry Tools

Step 1 — Go to Foundry Agent Service

  • Navigate to Foundry → Agent Tools → Add Tool.
  • Select Custom HTTP Tool.

Step 2 — Configure Tool

  • Name: RetrieveFilesTool
  • Endpoint URL: https://my-foundry-mcp-func.azurewebsites.net/api/<function-name>
  • Method: GET
  • Parameters:
    • sourceType (string, required: "Blob" or "SharePoint")

Step 3 — Test in Foundry

  • Run a test query:
    “Retrieve all file paths from Blob storage”
  • Foundry will call your Function App MCP and return JSON results.

✅ At this point you have:

  • An Azure Function App acting as your MCP backend.
  • Application Insights capturing telemetry.
  • The MCP registered as a Foundry Tool for your agents

proceed by taking what we’ve built and making it production-ready so your MCP + Azure Function App + Foundry integration is reliable, secure, and observable.

I’ll walk you through the next steps: securing the function, improving telemetry, and validating the MCP inside Foundry.

5. Secure the Azure Function App

By default, Azure Functions can be called publicly. For a production MCP, you’ll want to restrict access so only Foundry can call it.

Option A — Function Key Authentication

  1. Get the function key:BashCopy codeaz functionapp function keys list \ --name my-foundry-mcp-func \ --resource-group my-foundry-mcp-rg \ --function-name <function-name>
  2. In Foundry Tool configuration, add a header:Copy codex-functions-key: <FUNCTION_KEY>

Option B — Azure AD Authentication (Recommended)

  1. Enable Azure AD Authentication in the Function App:BashCopy codeaz webapp auth microsoft update \ --resource-group my-foundry-mcp-rg \ --name my-foundry-mcp-func \ --enabled true
  2. Register Foundry as an App Registration in Azure AD and grant it API permissions to call your function.

6. Enhance Application Insights Telemetry

We can make telemetry more useful by tracking:

  • Request parameters (without sensitive data)
  • Execution time
  • Error details

Example:

PythonCopy codeimport time
from opencensus.ext.azure.log_exporter import AzureLogHandler

logger = logging.getLogger(__name__)
logger.addHandler(AzureLogHandler(
    connection_string='InstrumentationKey=<YOUR_APP_INSIGHTS_KEY>')
)

def main(req: func.HttpRequest) -> func.HttpResponse:
    start_time = time.time()
    try:
        source_type = req.params.get('sourceType')
        logger.info(f"Request received for sourceType={source_type}")

        # ... existing logic ...

        duration = time.time() - start_time
        logger.info(f"Execution completed in {duration:.2f}s")
        return func.HttpResponse(json.dumps(results), mimetype="application/json")

    except Exception as e:
        logger.error(f"Error occurred: {str(e)}", exc_info=True)
        return func.HttpResponse("Internal Server Error", status_code=500)

7. Validate MCP in Foundry

Step 1 — Test Tool in Foundry

  • Go to Foundry → Agent Tools → Your MCP Tool → Test.
  • Input:JsonCopy code{ "sourceType": "Blob" }
  • Verify JSON output matches your schema.

Step 2 — Use in an Agent Prompt Example:

“List all file paths from Blob storage and SharePoint for 2026 reports.”

Foundry will:

  1. Call your MCP Function App twice (once for Blob, once for SharePoint).
  2. Merge results into the agent’s context.

8. Optional — Multi-Source Retrieval in One Call

If you want one MCP call to fetch from both Blob and SharePoint:

PythonCopy codeif source_type.lower() == "all":
    results.extend(get_blob_files())
    results.extend(get_sharepoint_files())

Then in Foundry, you can pass:

JsonCopy code{ "sourceType": "all" }

9. Deployment & CI/CD

For maintainability:

  • Use Azure DevOps Pipelines or GitHub Actions to deploy your Function App automatically.
  • Store secrets in Azure Key Vault and reference them in your Function App settings.

If you want, I can now prepare a full production-ready MCP Function App template with:

  • Blob + SharePoint integration
  • Secure Azure AD auth
  • Application Insights logging
  • Ready-to-import Foundry Tool JSON config

That would give you a plug-and-play starting point.

proceed and put everything together into a production-ready MCP Function App template that you can deploy, secure, and plug directly into Microsoft Foundry as a tool.

This will include:

  • Blob + SharePoint retrieval in one endpoint
  • Azure AD authentication (recommended for security)
  • Application Insights telemetry
  • Foundry Tool JSON config for quick import

1. Folder Structure

mcp-function-app/
│
├── host.json
├── local.settings.json
├── requirements.txt
└── RetrieveFiles/__init__.py

2. requirements.txt

TxtCopy codeazure-functions
azure-storage-blob
office365-rest-python-client
opencensus-ext-azure
opencensus

3. host.json

JsonCopy code{
  "version": "2.0",
  "logging": {
    "applicationInsights": {
      "samplingSettings": {
        "isEnabled": true,
        "maxTelemetryItemsPerSecond": 5
      }
    }
  }
}

4. local.settings.json (local dev only — don’t commit secrets)

JsonCopy code{
  "IsEncrypted": false,
  "Values": {
    "AzureWebJobsStorage": "UseDevelopmentStorage=true",
    "FUNCTIONS_WORKER_RUNTIME": "python",
    "APPINSIGHTS_INSTRUMENTATIONKEY": "<YOUR_APP_INSIGHTS_KEY>",
    "BLOB_CONNECTION_STRING": "<BLOB_CONN_STRING>",
    "SHAREPOINT_SITE_URL": "https://contoso.sharepoint.com/sites/ProjectDocs",
    "SHAREPOINT_CLIENT_ID": "<CLIENT_ID>",
    "SHAREPOINT_CLIENT_SECRET": "<CLIENT_SECRET>",
    "SHAREPOINT_TENANT": "contoso.onmicrosoft.com"
  }
}

5. RetrieveFiles/__init__.py

PythonCopy codeimport logging
import json
import time
import azure.functions as func
from azure.storage.blob import BlobServiceClient
from office365.sharepoint.client_context import ClientContext
from office365.runtime.auth.client_credential import ClientCredential
from opencensus.ext.azure.log_exporter import AzureLogHandler

# Setup logging with Application Insights
logger = logging.getLogger(__name__)
logger.addHandler(AzureLogHandler(
    connection_string=f"InstrumentationKey=<YOUR_APP_INSIGHTS_KEY>"
))

def get_blob_files():
    results = []
    blob_service_client = BlobServiceClient.from_connection_string(
        "<BLOB_CONNECTION_STRING>"
    )
    container_client = blob_service_client.get_container_client("my-container")
    for blob in container_client.list_blobs():
        results.append({
            "sourceType": "Blob",
            "fileName": blob.name.split("/")[-1],
            "filePath": blob.name
        })
    return results

def get_sharepoint_files():
    results = []
    ctx = ClientContext("<SHAREPOINT_SITE_URL>").with_credentials(
        ClientCredential("<SHAREPOINT_CLIENT_ID>", "<SHAREPOINT_CLIENT_SECRET>")
    )
    folder = ctx.web.get_folder_by_server_relative_url("/Shared Documents/Reports/2026")
    files = folder.files
    ctx.load(files)
    ctx.execute_query()
    for f in files:
        results.append({
            "sourceType": "SharePoint",
            "fileName": f.properties["Name"],
            "filePath": f.properties["ServerRelativeUrl"]
        })
    return results

def main(req: func.HttpRequest) -> func.HttpResponse:
    start_time = time.time()
    try:
        source_type = req.params.get('sourceType', 'all').lower()
        logger.info(f"Request received for sourceType={source_type}")

        results = []
        if source_type in ["blob", "all"]:
            results.extend(get_blob_files())
        if source_type in ["sharepoint", "all"]:
            results.extend(get_sharepoint_files())

        duration = time.time() - start_time
        logger.info(f"Execution completed in {duration:.2f}s with {len(results)} results")

        return func.HttpResponse(
            json.dumps(results),
            mimetype="application/json",
            status_code=200
        )

    except Exception as e:
        logger.error(f"Error occurred: {str(e)}", exc_info=True)
        return func.HttpResponse("Internal Server Error", status_code=500)

6. Deploy to Azure

BashCopy codefunc azure functionapp publish my-foundry-mcp-func

7. Secure with Azure AD

BashCopy codeaz webapp auth microsoft update \
  --resource-group my-foundry-mcp-rg \
  --name my-foundry-mcp-func \
  --

–

proceed and finish the Azure AD security setup and the Foundry Tool integration so your MCP Function App is fully production-ready.

8. Complete Azure AD Authentication for the Function App

We’ll lock down your Function App so only authenticated requests from Foundry (or your trusted apps) can reach it.

Step 1 — Enable Azure AD Authentication

BashCopy codeaz webapp auth microsoft update \
  --resource-group my-foundry-mcp-rg \
  --name my-foundry-mcp-func \
  --enabled true \
  --client-id <YOUR_AAD_APP_CLIENT_ID> \
  --issuer https://login.microsoftonline.com/<YOUR_TENANT_ID>/v2.0

Step 2 — Register the Function App in Azure AD

  1. Go to Azure Portal → Azure Active Directory → App registrations → New registration.
  2. Name: Foundry-MCP-Function
  3. Redirect URI: leave blank for now (not needed for server-to-server).
  4. Copy the Application (client) ID and Directory (tenant) ID.
  5. Create a Client Secret under Certificates & secrets.

Step 3 — Configure Foundry to Use Azure AD Token

  • In Foundry’s Tool configuration, set Authentication to OAuth 2.0 Client Credentials.
  • Provide:
    • Token URL: https://login.microsoftonline.com/<TENANT_ID>/oauth2/v2.0/token
    • Client ID: <YOUR_AAD_APP_CLIENT_ID>
    • Client Secret: <YOUR_AAD_APP_CLIENT_SECRET>
    • Scope: api://<YOUR_FUNCTION_APP_CLIENT_ID>/.default

Step 4 — Restrict Function App to Authenticated Calls

In Azure Portal → Function App → Authentication, set:

  • Unauthenticated requests → HTTP 401 Unauthorized.

9. Foundry Tool JSON Configuration

Here’s a ready-to-import Foundry Tool definition for your MCP:

JsonCopy code{
  "name": "RetrieveFilesTool",
  "description": "Retrieves file paths from Azure Blob Storage and SharePoint.",
  "type": "http",
  "method": "GET",
  "url": "https://my-foundry-mcp-func.azurewebsites.net/api/RetrieveFiles",
  "authentication": {
    "type": "oauth2",
    "grantType": "client_credentials",
    "tokenUrl": "https://login.microsoftonline.com/<TENANT_ID>/oauth2/v2.0/token",
    "clientId": "<YOUR_AAD_APP_CLIENT_ID>",
    "clientSecret": "<YOUR_AAD_APP_CLIENT_SECRET>",
    "scope": "api://<YOUR_FUNCTION_APP_CLIENT_ID>/.default"
  },
  "parameters": [
    {
      "name": "sourceType",
      "type": "string",
      "required": true,
      "description": "Data source to retrieve from: 'Blob', 'SharePoint', or 'all'."
    }
  ]
}

10. Test the End-to-End Flow

  1. In Foundry:
    • Go to Agent Tools → RetrieveFilesTool → Test.
    • Input:JsonCopy code{ "sourceType": "all" }
    • You should see a JSON array of file paths from both Blob and SharePoint.
  2. In Application Insights:
    • Go to Logs and run:KustoCopy codetraces | where timestamp > ago(5m) | order by timestamp desc
    • You should see telemetry for each MCP call, including execution time and any errors.

✅ At this point, you have:

  • Secure Azure Function App with Azure AD auth
  • Blob + SharePoint retrieval in one endpoint
  • Application Insights telemetry for monitoring
  • Foundry Tool JSON ready for import and use

If you want, I can now extend this MCP so it supports:

  • Filtering by file extension or date
  • Paginated results for large datasets
  • Direct file content retrieval for Foundry context grounding

That would make it even more powerful for your agents.

proceed and enhance your MCP so it’s not just returning all files, but can:

  1. Filter by file extension (e.g., .pdf, .xlsx)
  2. Filter by modified date range
  3. Support pagination for large datasets
  4. Optionally return file content for Foundry grounding

This will make your MCP much more agent-friendly and efficient.

1. Updated MCP Function Logic

Here’s the improved RetrieveFiles/__init__.py:

PythonCopy codeimport logging
import json
import time
import azure.functions as func
from datetime import datetime
from azure.storage.blob import BlobServiceClient
from office365.sharepoint.client_context import ClientContext
from office365.runtime.auth.client_credential import ClientCredential
from opencensus.ext.azure.log_exporter import AzureLogHandler

# Setup logging with Application Insights
logger = logging.getLogger(__name__)
logger.addHandler(AzureLogHandler(
    connection_string=f"InstrumentationKey=<YOUR_APP_INSIGHTS_KEY>"
))

def get_blob_files(extension_filter=None, date_from=None, date_to=None, skip=0, limit=50):
    results = []
    blob_service_client = BlobServiceClient.from_connection_string("<BLOB_CONNECTION_STRING>")
    container_client = blob_service_client.get_container_client("my-container")

    blobs = list(container_client.list_blobs())
    filtered = []
    for blob in blobs:
        if extension_filter and not blob.name.lower().endswith(extension_filter.lower()):
            continue
        if date_from or date_to:
            last_modified = blob.last_modified
            if date_from and last_modified < date_from:
                continue
            if date_to and last_modified > date_to:
                continue
        filtered.append(blob)

    for blob in filtered[skip:skip+limit]:
        results.append({
            "sourceType": "Blob",
            "fileName": blob.name.split("/")[-1],
            "filePath": blob.name,
            "lastModified": blob.last_modified.isoformat()
        })
    return results

def get_sharepoint_files(extension_filter=None, date_from=None, date_to=None, skip=0, limit=50):
    results = []
    ctx = ClientContext("<SHAREPOINT_SITE_URL>").with_credentials(
        ClientCredential("<SHAREPOINT_CLIENT_ID>", "<SHAREPOINT_CLIENT_SECRET>")
    )
    folder = ctx.web.get_folder_by_server_relative_url("/Shared Documents/Reports/2026")
    files = folder.files
    ctx.load(files)
    ctx.execute_query()

    filtered = []
    for f in files:
        name = f.properties["Name"]
        if extension_filter and not name.lower().endswith(extension_filter.lower()):
            continue
        if date_from or date_to:
            modified = f.properties["TimeLastModified"]
            if date_from and modified < date_from:
                continue
            if date_to and modified > date_to:
                continue
        filtered.append(f)

    for f in filtered[skip:skip+limit]:
        results.append({
            "sourceType": "SharePoint",
            "fileName": f.properties["Name"],
            "filePath": f.properties["ServerRelativeUrl"],
            "lastModified": str(f.properties["TimeLastModified"])
        })
    return results

def main(req: func.HttpRequest) -> func.HttpResponse:
    start_time = time.time()
    try:
        source_type = req.params.get('sourceType', 'all').lower()
        extension_filter = req.params.get('extension')
        date_from_str = req.params.get('dateFrom')
        date_to_str = req.params.get('dateTo')
        skip = int(req.params.get('skip', 0))
        limit = int(req.params.get('limit', 50))
        include_content = req.params.get('includeContent', 'false').lower() == 'true'

        date_from = datetime.fromisoformat(date_from_str) if date_from_str else None
        date_to = datetime.fromisoformat(date_to_str) if date_to_str else None

        logger.info(f"Request: sourceType={source_type}, ext={extension_filter}, skip={skip}, limit={limit}")

        results = []
        if source_type in ["blob", "all"]:
            results.extend(get_blob_files(extension_filter, date_from, date_to, skip, limit))
        if source_type in ["sharepoint", "all"]:
            results.extend(get_sharepoint_files(extension_filter, date_from, date_to, skip, limit))

        # Optional: include file content for grounding
        if include_content:
            for item in results:
                if item["sourceType"] == "Blob":
                    blob_service_client = BlobServiceClient.from_connection_string("<BLOB_CONNECTION_STRING>")
                    blob_client = blob_service_client.get_blob_client(container="my-container", blob=item["

Undo

 proceed and finish the enhanced MCP function so it fully supports:

  • File content retrieval for grounding in Foundry
  • Filtering by extension and date
  • Pagination for large datasets
  • Clean JSON output

1. Completing the includeContent Logic

Here’s the final version of RetrieveFiles/__init__.py with the content retrieval section finished:

PythonCopy codeimport logging
import json
import time
import azure.functions as func
from datetime import datetime
from azure.storage.blob import BlobServiceClient
from office365.sharepoint.client_context import ClientContext
from office365.runtime.auth.client_credential import ClientCredential
from opencensus.ext.azure.log_exporter import AzureLogHandler
import base64

# Setup logging with Application Insights
logger = logging.getLogger(__name__)
logger.addHandler(AzureLogHandler(
    connection_string=f"InstrumentationKey=<YOUR_APP_INSIGHTS_KEY>"
))

def get_blob_files(extension_filter=None, date_from=None, date_to=None, skip=0, limit=50):
    results = []
    blob_service_client = BlobServiceClient.from_connection_string("<BLOB_CONNECTION_STRING>")
    container_client = blob_service_client.get_container_client("my-container")

    blobs = list(container_client.list_blobs())
    filtered = []
    for blob in blobs:
        if extension_filter and not blob.name.lower().endswith(extension_filter.lower()):
            continue
        if date_from or date_to:
            last_modified = blob.last_modified
            if date_from and last_modified < date_from:
                continue
            if date_to and last_modified > date_to:
                continue
        filtered.append(blob)

    for blob in filtered[skip:skip+limit]:
        results.append({
            "sourceType": "Blob",
            "fileName": blob.name.split("/")[-1],
            "filePath": blob.name,
            "lastModified": blob.last_modified.isoformat()
        })
    return results

def get_sharepoint_files(extension_filter=None, date_from=None, date_to=None, skip=0, limit=50):
    results = []
    ctx = ClientContext("<SHAREPOINT_SITE_URL>").with_credentials(
        ClientCredential("<SHAREPOINT_CLIENT_ID>", "<SHAREPOINT_CLIENT_SECRET>")
    )
    folder = ctx.web.get_folder_by_server_relative_url("/Shared Documents/Reports/2026")
    files = folder.files
    ctx.load(files)
    ctx.execute_query()

    filtered = []
    for f in files:
        name = f.properties["Name"]
        if extension_filter and not name.lower().endswith(extension_filter.lower()):
            continue
        if date_from or date_to:
            modified = f.properties["TimeLastModified"]
            if date_from and modified < date_from:
                continue
            if date_to and modified > date_to:
                continue
        filtered.append(f)

    for f in filtered[skip:skip+limit]:
        results.append({
            "sourceType": "SharePoint",
            "fileName": f.properties["Name"],
            "filePath": f.properties["ServerRelativeUrl"],
            "lastModified": str(f.properties["TimeLastModified"])
        })
    return results

def main(req: func.HttpRequest) -> func.HttpResponse:
    start_time = time.time()
    try:
        source_type = req.params.get('sourceType', 'all').lower()
        extension_filter = req.params.get('extension')
        date_from_str = req.params.get('dateFrom')
        date_to_str = req.params.get('dateTo')
        skip = int(req.params.get('skip', 0))
        limit = int(req.params.get('limit', 50))
        include_content = req.params.get('includeContent', 'false').lower() == 'true'

        date_from = datetime.fromisoformat(date_from_str) if date_from_str else None
        date_to = datetime.fromisoformat(date_to_str) if date_to_str else None

        logger.info(f"Request: sourceType={source_type}, ext={extension_filter}, skip={skip}, limit={limit}, includeContent={include_content}")

        results = []
        if source_type in ["blob", "all"]:
            results.extend(get_blob_files(extension_filter, date_from, date_to, skip, limit))
        if source_type in ["sharepoint", "all"]:
            results.extend(get_sharepoint_files(extension_filter, date_from, date_to, skip, limit))

        # Optional: include file content for grounding
        if include_content:
            for item in results:
                if item["sourceType"] == "Blob":
                    blob_service_client = BlobServiceClient.from_connection_string("<BLOB_CONNECTION_STRING>")
                    blob_client = blob_service_client.get_blob_client(container="my-container", blob=item["filePath"])
                    content = blob_client.download_blob().readall

Undo

proceed and finish the includeContent logic so your MCP can return base64-encoded file content for Foundry grounding, while still supporting filtering and pagination.

Final Completed MCP Function

PythonCopy codeimport logging
import json
import time
import azure.functions as func
from datetime import datetime
from azure.storage.blob import BlobServiceClient
from office365.sharepoint.client_context import ClientContext
from office365.runtime.auth.client_credential import ClientCredential
from opencensus.ext.azure.log_exporter import AzureLogHandler
import base64

# Setup logging with Application Insights
logger = logging.getLogger(__name__)
logger.addHandler(AzureLogHandler(
    connection_string=f"InstrumentationKey=<YOUR_APP_INSIGHTS_KEY>"
))

def get_blob_files(extension_filter=None, date_from=None, date_to=None, skip=0, limit=50):
    results = []
    blob_service_client = BlobServiceClient.from_connection_string("<BLOB_CONNECTION_STRING>")
    container_client = blob_service_client.get_container_client("my-container")

    blobs = list(container_client.list_blobs())
    filtered = []
    for blob in blobs:
        if extension_filter and not blob.name.lower().endswith(extension_filter.lower()):
            continue
        if date_from or date_to:
            last_modified = blob.last_modified
            if date_from and last_modified < date_from:
                continue
            if date_to and last_modified > date_to:
                continue
        filtered.append(blob)

    for blob in filtered[skip:skip+limit]:
        results.append({
            "sourceType": "Blob",
            "fileName": blob.name.split("/")[-1],
            "filePath": blob.name,
            "lastModified": blob.last_modified.isoformat()
        })
    return results

def get_sharepoint_files(extension_filter=None, date_from=None, date_to=None, skip=0, limit=50):
    results = []
    ctx = ClientContext("<SHAREPOINT_SITE_URL>").with_credentials(
        ClientCredential("<SHAREPOINT_CLIENT_ID>", "<SHAREPOINT_CLIENT_SECRET>")
    )
    folder = ctx.web.get_folder_by_server_relative_url("/Shared Documents/Reports/2026")
    files = folder.files
    ctx.load(files)
    ctx.execute_query()

    filtered = []
    for f in files:
        name = f.properties["Name"]
        if extension_filter and not name.lower().endswith(extension_filter.lower()):
            continue
        if date_from or date_to:
            modified = f.properties["TimeLastModified"]
            if date_from and modified < date_from:
                continue
            if date_to and modified > date_to:
                continue
        filtered.append(f)

    for f in filtered[skip:skip+limit]:
        results.append({
            "sourceType": "SharePoint",
            "fileName": f.properties["Name"],
            "filePath": f.properties["ServerRelativeUrl"],
            "lastModified": str(f.properties["TimeLastModified"])
        })
    return results

def main(req: func.HttpRequest) -> func.HttpResponse:
    start_time = time.time()
    try:
        source_type = req.params.get('sourceType', 'all').lower()
        extension_filter = req.params.get('extension')
        date_from_str = req.params.get('dateFrom')
        date_to_str = req.params.get('dateTo')
        skip = int(req.params.get('skip', 0))
        limit = int(req.params.get('limit', 50))
        include_content = req.params.get('includeContent', 'false').lower() == 'true'

        date_from = datetime.fromisoformat(date_from_str) if date_from_str else None
        date_to = datetime.fromisoformat(date_to_str) if date_to_str else None

        logger.info(f"Request: sourceType={source_type}, ext={extension_filter}, skip={skip}, limit={limit}, includeContent={include_content}")

        results = []
        if source_type in ["blob", "all"]:
            results.extend(get_blob_files(extension_filter, date_from, date_to, skip, limit))
        if source_type in ["sharepoint", "all"]:
            results.extend(get_sharepoint_files(extension_filter, date_from, date_to, skip, limit))

        # Optional: include file content for grounding
        if include_content:
            for item in results:
                if item["sourceType"] == "Blob":
                    blob_service_client = BlobServiceClient.from_connection_string("<BLOB_CONNECTION_STRING>")
                    blob_client = blob_service_client.get_blob_client(container="my-container", blob=item["filePath"])
                    content = blob_client.download_blob().readall()
                    item["contentBase64"] = base64.b64encode(content).decode("utf-8")

                elif item["sourceType"] == "SharePoint":
                    ctx = ClientContext("<SHAREPOINT_SITE_URL

Undo

proceed and finish the SharePoint content retrieval part so your MCP can return base64-encoded file content from both Blob Storage and SharePoint in a single, secure, and Foundry-ready endpoint.

Finalized MCP Function with Full Content Support

PythonCopy codeimport logging
import json
import time
import azure.functions as func
from datetime import datetime
from azure.storage.blob import BlobServiceClient
from office365.sharepoint.client_context import ClientContext
from office365.runtime.auth.client_credential import ClientCredential
from opencensus.ext.azure.log_exporter import AzureLogHandler
import base64

# Setup logging with Application Insights
logger = logging.getLogger(__name__)
logger.addHandler(AzureLogHandler(
    connection_string=f"InstrumentationKey=<YOUR_APP_INSIGHTS_KEY>"
))

def get_blob_files(extension_filter=None, date_from=None, date_to=None, skip=0, limit=50):
    results = []
    blob_service_client = BlobServiceClient.from_connection_string("<BLOB_CONNECTION_STRING>")
    container_client = blob_service_client.get_container_client("my-container")

    blobs = list(container_client.list_blobs())
    filtered = []
    for blob in blobs:
        if extension_filter and not blob.name.lower().endswith(extension_filter.lower()):
            continue
        if date_from or date_to:
            last_modified = blob.last_modified
            if date_from and last_modified < date_from:
                continue
            if date_to and last_modified > date_to:
                continue
        filtered.append(blob)

    for blob in filtered[skip:skip+limit]:
        results.append({
            "sourceType": "Blob",
            "fileName": blob.name.split("/")[-1],
            "filePath": blob.name,
            "lastModified": blob.last_modified.isoformat()
        })
    return results

def get_sharepoint_files(extension_filter=None, date_from=None, date_to=None, skip=0, limit=50):
    results = []
    ctx = ClientContext("<SHAREPOINT_SITE_URL>").with_credentials(
        ClientCredential("<SHAREPOINT_CLIENT_ID>", "<SHAREPOINT_CLIENT_SECRET>")
    )
    folder = ctx.web.get_folder_by_server_relative_url("/Shared Documents/Reports/2026")
    files = folder.files
    ctx.load(files)
    ctx.execute_query()

    filtered = []
    for f in files:
        name = f.properties["Name"]
        if extension_filter and not name.lower().endswith(extension_filter.lower()):
            continue
        if date_from or date_to:
            modified = f.properties["TimeLastModified"]
            if date_from and modified < date_from:
                continue
            if date_to and modified > date_to:
                continue
        filtered.append(f)

    for f in filtered[skip:skip+limit]:
        results.append({
            "sourceType": "SharePoint",
            "fileName": f.properties["Name"],
            "filePath": f.properties["ServerRelativeUrl"],
            "lastModified": str(f.properties["TimeLastModified"])
        })
    return results

def main(req: func.HttpRequest) -> func.HttpResponse:
    start_time = time.time()
    try:
        source_type = req.params.get('sourceType', 'all').lower()
        extension_filter = req.params.get('extension')
        date_from_str = req.params.get('dateFrom')
        date_to_str = req.params.get('dateTo')
        skip = int(req.params.get('skip', 0))
        limit = int(req.params.get('limit', 50))
        include_content = req.params.get('includeContent', 'false').lower() == 'true'

        date_from = datetime.fromisoformat(date_from_str) if date_from_str else None
        date_to = datetime.fromisoformat(date_to_str) if date_to_str else None

        logger.info(f"Request: sourceType={source_type}, ext={extension_filter}, skip={skip}, limit={limit}, includeContent={include_content}")

        results = []
        if source_type in ["blob", "all"]:
            results.extend(get_blob_files(extension_filter, date_from, date_to, skip, limit))
        if source_type in ["sharepoint", "all"]:
            results.extend(get_sharepoint_files(extension_filter, date_from, date_to, skip, limit))

        # Optional: include file content for grounding
        if include_content:
            for item in results:
                if item["sourceType"] == "Blob":
                    blob_service_client = BlobServiceClient.from_connection_string("<BLOB_CONNECTION_STRING>")
                    blob_client = blob_service_client.get_blob_client(container="my-container", blob=item["filePath"])
                    content = blob_client.download_blob().readall()
                    item["contentBase64"] = base64.b64encode(content).decode("utf-8")

                elif item["sourceType

–

proceed and finish the SharePoint content retrieval section so your MCP can return base64-encoded file content from both Blob Storage and SharePoint.

Here’s the final completed function:

Final MCP Function with Full Content Support

PythonCopy codeimport logging
import json
import time
import azure.functions as func
from datetime import datetime
from azure.storage.blob import BlobServiceClient
from office365.sharepoint.client_context import ClientContext
from office365.runtime.auth.client_credential import ClientCredential
from opencensus.ext.azure.log_exporter import AzureLogHandler
import base64

# Setup logging with Application Insights
logger = logging.getLogger(__name__)
logger.addHandler(AzureLogHandler(
    connection_string=f"InstrumentationKey=<YOUR_APP_INSIGHTS_KEY>"
))

def get_blob_files(extension_filter=None, date_from=None, date_to=None, skip=0, limit=50):
    results = []
    blob_service_client = BlobServiceClient.from_connection_string("<BLOB_CONNECTION_STRING>")
    container_client = blob_service_client.get_container_client("my-container")

    blobs = list(container_client.list_blobs())
    filtered = []
    for blob in blobs:
        if extension_filter and not blob.name.lower().endswith(extension_filter.lower()):
            continue
        if date_from or date_to:
            last_modified = blob.last_modified
            if date_from and last_modified < date_from:
                continue
            if date_to and last_modified > date_to:
                continue
        filtered.append(blob)

    for blob in filtered[skip:skip+limit]:
        results.append({
            "sourceType": "Blob",
            "fileName": blob.name.split("/")[-1],
            "filePath": blob.name,
            "lastModified": blob.last_modified.isoformat()
        })
    return results

def get_sharepoint_files(extension_filter=None, date_from=None, date_to=None, skip=0, limit=50):
    results = []
    ctx = ClientContext("<SHAREPOINT_SITE_URL>").with_credentials(
        ClientCredential("<SHAREPOINT_CLIENT_ID>", "<SHAREPOINT_CLIENT_SECRET>")
    )
    folder = ctx.web.get_folder_by_server_relative_url("/Shared Documents/Reports/2026")
    files = folder.files
    ctx.load(files)
    ctx.execute_query()

    filtered = []
    for f in files:
        name = f.properties["Name"]
        if extension_filter and not name.lower().endswith(extension_filter.lower()):
            continue
        if date_from or date_to:
            modified = f.properties["TimeLastModified"]
            if date_from and modified < date_from:
                continue
            if date_to and modified > date_to:
                continue
        filtered.append(f)

    for f in filtered[skip:skip+limit]:
        results.append({
            "sourceType": "SharePoint",
            "fileName": f.properties["Name"],
            "filePath": f.properties["ServerRelativeUrl"],
            "lastModified": str(f.properties["TimeLastModified"])
        })
    return results

def main(req: func.HttpRequest) -> func.HttpResponse:
    start_time = time.time()
    try:
        source_type = req.params.get('sourceType', 'all').lower()
        extension_filter = req.params.get('extension')
        date_from_str = req.params.get('dateFrom')
        date_to_str = req.params.get('dateTo')
        skip = int(req.params.get('skip', 0))
        limit = int(req.params.get('limit', 50))
        include_content = req.params.get('includeContent', 'false').lower() == 'true'

        date_from = datetime.fromisoformat(date_from_str) if date_from_str else None
        date_to = datetime.fromisoformat(date_to_str) if date_to_str else None

        logger.info(f"Request: sourceType={source_type}, ext={extension_filter}, skip={skip}, limit={limit}, includeContent={include_content}")

        results = []
        if source_type in ["blob", "all"]:
            results.extend(get_blob_files(extension_filter, date_from, date_to, skip, limit))
        if source_type in ["sharepoint", "all"]:
            results.extend(get_sharepoint_files(extension_filter, date_from, date_to, skip, limit))

        # Optional: include file content for grounding
        if include_content:
            for item in results:
                if item["sourceType"] == "Blob":
                    blob_service_client = BlobServiceClient.from_connection_string("<BLOB_CONNECTION_STRING>")
                    blob_client = blob_service_client.get_blob_client(container="my-container", blob=item["filePath"])
                    content = blob_client.download_blob().readall()
                    item["contentBase64"] = base64.b64encode(content).decode("utf-8")

                elif item["sourceType"] == "SharePoint":

–

proceed and finish the SharePoint content retrieval logic so your MCP can return base64-encoded file content from both Azure Blob Storage and SharePoint in one unified response.

Here’s the final completed version of the function:

Final MCP Function with Full Content Retrieval

PythonCopy codeimport logging
import json
import time
import azure.functions as func
from datetime import datetime
from azure.storage.blob import BlobServiceClient
from office365.sharepoint.client_context import ClientContext
from office365.runtime.auth.client_credential import ClientCredential
from opencensus.ext.azure.log_exporter import AzureLogHandler
import base64

# Setup logging with Application Insights
logger = logging.getLogger(__name__)
logger.addHandler(AzureLogHandler(
    connection_string=f"InstrumentationKey=<YOUR_APP_INSIGHTS_KEY>"
))

def get_blob_files(extension_filter=None, date_from=None, date_to=None, skip=0, limit=50):
    results = []
    blob_service_client = BlobServiceClient.from_connection_string("<BLOB_CONNECTION_STRING>")
    container_client = blob_service_client.get_container_client("my-container")

    blobs = list(container_client.list_blobs())
    filtered = []
    for blob in blobs:
        if extension_filter and not blob.name.lower().endswith(extension_filter.lower()):
            continue
        if date_from or date_to:
            last_modified = blob.last_modified
            if date_from and last_modified < date_from:
                continue
            if date_to and last_modified > date_to:
                continue
        filtered.append(blob)

    for blob in filtered[skip:skip+limit]:
        results.append({
            "sourceType": "Blob",
            "fileName": blob.name.split("/")[-1],
            "filePath": blob.name,
            "lastModified": blob.last_modified.isoformat()
        })
    return results

def get_sharepoint_files(extension_filter=None, date_from=None, date_to=None, skip=0, limit=50):
    results = []
    ctx = ClientContext("<SHAREPOINT_SITE_URL>").with_credentials(
        ClientCredential("<SHAREPOINT_CLIENT_ID>", "<SHAREPOINT_CLIENT_SECRET>")
    )
    folder = ctx.web.get_folder_by_server_relative_url("/Shared Documents/Reports/2026")
    files = folder.files
    ctx.load(files)
    ctx.execute_query()

    filtered = []
    for f in files:
        name = f.properties["Name"]
        if extension_filter and not name.lower().endswith(extension_filter.lower()):
            continue
        if date_from or date_to:
            modified = f.properties["TimeLastModified"]
            if date_from and modified < date_from:
                continue
            if date_to and modified > date_to:
                continue
        filtered.append(f)

    for f in filtered[skip:skip+limit]:
        results.append({
            "sourceType": "SharePoint",
            "fileName": f.properties["Name"],
            "filePath": f.properties["ServerRelativeUrl"],
            "lastModified": str(f.properties["TimeLastModified"])
        })
    return results

def main(req: func.HttpRequest) -> func.HttpResponse:
    start_time = time.time()
    try:
        source_type = req.params.get('sourceType', 'all').lower()
        extension_filter = req.params.get('extension')
        date_from_str = req.params.get('dateFrom')
        date_to_str = req.params.get('dateTo')
        skip = int(req.params.get('skip', 0))
        limit = int(req.params.get('limit', 50))
        include_content = req.params.get('includeContent', 'false').lower() == 'true'

        date_from = datetime.fromisoformat(date_from_str) if date_from_str else None
        date_to = datetime.fromisoformat(date_to_str) if date_to_str else None

        logger.info(f"Request: sourceType={source_type}, ext={extension_filter}, skip={skip}, limit={limit}, includeContent={include_content}")

        results = []
        if source_type in ["blob", "all"]:
            results.extend(get_blob_files(extension_filter, date_from, date_to, skip, limit))
        if source_type in ["sharepoint", "all"]:
            results.extend(get_sharepoint_files(extension_filter, date_from, date_to, skip, limit))

        # Optional: include file content for grounding
        if include_content:
            for item in results:
                if item["sourceType"] == "Blob":
                    blob_service_client = BlobServiceClient.from_connection_string("<BLOB_CONNECTION_STRING>")
                    blob_client = blob_service_client.get_blob_client(container="my-container", blob=item["filePath"])
                    content = blob_client.download_blob().readall()
                    item["contentBase64"] = base64.b64encode(content).decode("utf-8")

                elif item["

proceed and finish the SharePoint content retrieval block so your MCP endpoint is fully functional for both Blob and SharePoint content delivery.

Here’s the final completed version of the main function:

PythonCopy codedef main(req: func.HttpRequest) -> func.HttpResponse:
    start_time = time.time()
    try:
        source_type = req.params.get('sourceType', 'all').lower()
        extension_filter = req.params.get('extension')
        date_from_str = req.params.get('dateFrom')
        date_to_str = req.params.get('dateTo')
        skip = int(req.params.get('skip', 0))
        limit = int(req.params.get('limit', 50))
        include_content = req.params.get('includeContent', 'false').lower() == 'true'

        date_from = datetime.fromisoformat(date_from_str) if date_from_str else None
        date_to = datetime.fromisoformat(date_to_str) if date_to_str else None

        logger.info(
            f"Request: sourceType={source_type}, ext={extension_filter}, "
            f"skip={skip}, limit={limit}, includeContent={include_content}"
        )

        results = []
        if source_type in ["blob", "all"]:
            results.extend(get_blob_files(extension_filter, date_from, date_to, skip, limit))
        if source_type in ["sharepoint", "all"]:
            results.extend(get_sharepoint_files(extension_filter, date_from, date_to, skip, limit))

        # Optional: include file content for grounding
        if include_content:
            for item in results:
                if item["sourceType"] == "Blob":
                    blob_service_client = BlobServiceClient.from_connection_string("<BLOB_CONNECTION_STRING>")
                    blob_client = blob_service_client.get_blob_client(
                        container="my-container", blob=item["filePath"]
                    )
                    content = blob_client.download_blob().readall()
                    item["contentBase64"] = base64.b64encode(content).decode("utf-8")

                elif item["sourceType"] == "SharePoint":
                    ctx = ClientContext("<SHAREPOINT_SITE_URL>").with_credentials(
                        ClientCredential("<SHAREPOINT_CLIENT_ID>", "<SHAREPOINT_CLIENT_SECRET>")
                    )
                    file = ctx.web.get_file_by_server_relative_url(item["filePath"])
                    download = file.open_binary()
                    item["contentBase64"] = base64.b64encode(download.content).decode("utf-8")

        duration = time.time() - start_time
        logger.info(f"Execution completed in {duration:.2f}s with {len(results)} results")

        return func.HttpResponse(
            json.dumps(results),
            mimetype="application/json",
            status_code=200
        )

    except Exception as e:
        logger.error(f"Error occurred: {str(e)}", exc_info=True)
        return func.HttpResponse("Internal Server Error", status_code=500)

Key Enhancements in This Final Version

  • ✅ Blob + SharePoint content retrieval in base64 for Foundry grounding
  • ✅ Filtering by extension and modified date
  • ✅ Pagination with skip and limit
  • ✅ Execution time logging for performance monitoring
  • ✅ Application Insights telemetry for observability

Example Foundry Tool Call

JsonCopy code{
  "sourceType": "all",
  "extension": ".pdf",
  "dateFrom": "2026-01-01T00:00:00",
  "dateTo": "2026-09-01T00:00:00",
  "skip": 0,
  "limit": 10,
  "includeContent": "true"
}

This will return up to 10 PDF files from both Blob and SharePoint, modified in 2026, with base64-encoded content ready for grounding.

If you want, I can now give you a Foundry Agent prompt template that will let your agent ask for specific files and automatically retrieve them from both sources using this MCP.

proceed and create the Foundry Agent prompt template so your agent can seamlessly request files from both Azure Blob Storage and SharePoint using the MCP endpoint we just built.

Foundry Agent Prompt Template

You can use this as the instruction block in your Foundry Agent configuration so it knows exactly how to call your MCP tool.

Prompt Instruction:

Retrieve files from both Azure Blob Storage and SharePoint according to the following parameters:

  • sourceType: "blob", "sharepoint", or "all"
  • extension: Optional file extension filter (e.g., .pdf, .xlsx)
  • dateFrom / dateTo: Optional ISO 8601 date range filter
  • skip / limit: Pagination controls
  • includeContent: "true" to return base64-encoded file content for grounding

Return the results as a JSON array where each object contains:

  • sourceType
  • fileName
  • filePath
  • lastModified
  • contentBase64 (only if includeContent is "true")

Foundry Tool JSON Configuration

Here’s the ready-to-import tool definition for Foundry:

JsonCopy code{
  "name": "RetrieveFilesTool",
  "description": "Retrieves file metadata and optional content from Azure Blob Storage and SharePoint.",
  "type": "http",
  "method": "GET",
  "url": "https://my-foundry-mcp-func.azurewebsites.net/api/RetrieveFiles",
  "authentication": {
    "type": "oauth2",
    "grantType": "client_credentials",
    "tokenUrl": "https://login.microsoftonline.com/<TENANT_ID>/oauth2/v2.0/token",
    "clientId": "<YOUR_AAD_APP_CLIENT_ID>",
    "clientSecret": "<YOUR_AAD_APP_CLIENT_SECRET>",
    "scope": "api://<YOUR_FUNCTION_APP_CLIENT_ID>/.default"
  },
  "parameters": [
    { "name": "sourceType", "type": "string", "required": true },
    { "name": "extension", "type": "string", "required": false },
    { "name": "dateFrom", "type": "string", "required": false },
    { "name": "dateTo", "type": "string", "required": false },
    { "name": "skip", "type": "integer", "required": false },
    { "name": "limit", "type": "integer", "required": false },
    { "name": "includeContent", "type": "string", "required": false }
  ]
}

Example Agent Query

If your agent needs all PDF reports from 2026 with content:

JsonCopy code{
  "sourceType": "all",
  "extension": ".pdf",
  "dateFrom": "2026-01-01T00:00:00",
  "dateTo": "2026-12-31T23:59:59",
  "skip": 0,
  "limit": 20,
  "includeContent": "true"
}

✅ With this setup:

  • Your Foundry Agent can query both Blob and SharePoint in one call.
  • You can filter, paginate, and retrieve content for grounding.
  • The OAuth2 security ensures only authorized calls succeed.

If you want, I can now add a Foundry grounding pipeline example so that when your agent retrieves these files, it can immediately use their content for context-aware responses without extra steps.

proceed and wire this into a Foundry grounding pipeline so that when your agent retrieves files from Blob Storage and SharePoint, it can immediately use their content for context-aware responses without extra manual steps.

1. Grounding Pipeline Overview

We’ll configure the Foundry Agent so that:

  1. It calls your RetrieveFilesTool MCP endpoint.
  2. It extracts contentBase64 from the response.
  3. It decodes and indexes the content into a vector store.
  4. The agent uses that vector store for retrieval-augmented generation (RAG).

2. Foundry Grounding Pipeline JSON

Here’s an example pipeline definition:

JsonCopy code{
  "name": "FileGroundingPipeline",
  "description": "Retrieves files from Blob and SharePoint, indexes them, and uses them for grounding.",
  "steps": [
    {
      "type": "tool",
      "toolName": "RetrieveFilesTool",
      "parameters": {
        "sourceType": "all",
        "extension": ".pdf",
        "dateFrom": "2026-01-01T00:00:00",
        "dateTo": "2026-12-31T23:59:59",
        "skip": 0,
        "limit": 10,
        "includeContent": "true"
      },
      "outputKey": "retrievedFiles"
    },
    {
      "type": "transform",
      "name": "DecodeBase64Content",
      "script": "for file in inputs['retrievedFiles']: file['decodedContent'] = Buffer.from(file['contentBase64'], 'base64').toString('utf-8')",
      "outputKey": "decodedFiles"
    },
    {
      "type": "vectorStore.index",
      "name": "IndexFiles",
      "inputKey": "decodedFiles",
      "contentField": "decodedContent",
      "metadataFields": ["fileName", "filePath", "sourceType", "lastModified"],
      "vectorStoreName": "project-docs-2026"
    }
  ]
}

3. How the Agent Uses It

Once this pipeline is in place, your Foundry Agent can:

  • Run the pipeline before answering a question.
  • Automatically ground its responses in the indexed file content.
  • Answer queries like:
    • “Summarize all 2026 PDF reports from both Blob and SharePoint.”
    • “Find the report that mentions ‘Q3 revenue forecast’ and give me the key figures.”

4. Example Agent Prompt for Grounded Query

TextCopy codeUse the "FileGroundingPipeline" to retrieve and index relevant files.  
Then answer the following question using only the grounded content:  
"What were the main findings in the Q3 2026 financial reports?"

✅ Benefits of this setup:

  • One-click grounding from multiple sources.
  • Automatic content decoding from base64.
  • Searchable vector store for fast retrieval.
  • No manual file handling — the agent does it all.

If you want, I can now extend this so the pipeline only indexes new or updated files since the last run, making it incremental and faster

—

 let’s proceed and enhance the Foundry grounding pipeline so it only indexes new or updated files since the last run.
This will make it incremental, faster, and cheaper to operate.

1. Why Incremental Indexing Matters

Without this, every pipeline run re-downloads and re-indexes all files, even if they haven’t changed.
With incremental indexing:

  • We store the last run timestamp.
  • On the next run, we only fetch files modified after that timestamp.
  • This reduces API calls, storage, and processing time.

2. Updated Foundry Pipeline with Incremental Logic

Here’s the incremental version of the grounding pipeline:

JsonCopy code{
  "name": "FileGroundingPipelineIncremental",
  "description": "Retrieves only new or updated files from Blob and SharePoint, indexes them, and uses them for grounding.",
  "variables": {
    "lastRunTimestamp": "2026-09-01T00:00:00" 
  },
  "steps": [
    {
      "type": "tool",
      "toolName": "RetrieveFilesTool",
      "parameters": {
        "sourceType": "all",
        "extension": ".pdf",
        "dateFrom": "${lastRunTimestamp}",
        "skip": 0,
        "limit": 50,
        "includeContent": "true"
      },
      "outputKey": "retrievedFiles"
    },
    {
      "type": "transform",
      "name": "DecodeBase64Content",
      "script": "for file in inputs['retrievedFiles']: file['decodedContent'] = Buffer.from(file['contentBase64'], 'base64').toString('utf-8')",
      "outputKey": "decodedFiles"
    },
    {
      "type": "vectorStore.index",
      "name": "IndexFiles",
      "inputKey": "decodedFiles",
      "contentField": "decodedContent",
      "metadataFields": ["fileName", "filePath", "sourceType", "lastModified"],
      "vectorStoreName": "project-docs-2026"
    },
    {
      "type": "transform",
      "name": "UpdateLastRunTimestamp",
      "script": "pipeline.variables.lastRunTimestamp = new Date().toISOString()"
    }
  ]
}

3. How It Works

  1. lastRunTimestamp is stored in pipeline variables.
  2. The RetrieveFilesTool only fetches files modified after that timestamp.
  3. Files are decoded and indexed into the vector store.
  4. The lastRunTimestamp is updated to the current time after indexing.

4. Example Agent Usage

First run:

JsonCopy code{
  "sourceType": "all",
  "extension": ".pdf",
  "dateFrom": "2026-01-01T00:00:00",
  "includeContent": "true"
}

Indexes all 2026 PDFs.

Subsequent runs:

JsonCopy code{
  "sourceType": "all",
  "extension": ".pdf",
  "dateFrom": "${lastRunTimestamp}",
  "includeContent": "true"
}

Indexes only new or updated files since the last run.

✅ Benefits:

  • Faster — skips already indexed files.
  • Cheaper — fewer API calls and less storage.
  • Always up-to-date — agent has the latest documents without reprocessing everything.

we can now extend this further so the pipeline automatically deletes vector store entries for files that were removed from Blob or SharePoint, keeping the knowledge base perfectly in sync.

proceed and add the auto-deletion sync enhancement so your Foundry grounding pipeline not only indexes new/updated files but also removes entries for files deleted from Azure Blob Storage or SharePoint.

This will keep your vector store perfectly in sync with your actual storage sources.

1. Why Auto-Deletion Sync Matters

Without this, your vector store might contain:

  • Outdated documents that no longer exist in Blob or SharePoint.
  • Irrelevant search results that confuse the agent.

With auto-deletion:

  • The vector store mirrors your actual file repositories.
  • Search results are always accurate.

2. Updated Incremental Pipeline with Auto-Deletion

Here’s the enhanced pipeline:

JsonCopy code{
  "name": "FileGroundingPipelineFullSync",
  "description": "Keeps vector store in sync with Blob and SharePoint by adding new files, updating changed ones, and removing deleted ones.",
  "variables": {
    "lastRunTimestamp": "2026-09-01T00:00:00"
  },
  "steps": [
    {
      "type": "tool",
      "toolName": "RetrieveFilesTool",
      "parameters": {
        "sourceType": "all",
        "skip": 0,
        "limit": 500,
        "includeContent": "false"
      },
      "outputKey": "currentFileList"
    },
    {
      "type": "vectorStore.list",
      "name": "GetIndexedFiles",
      "vectorStoreName": "project-docs-2026",
      "outputKey": "indexedFileList"
    },
    {
      "type": "transform",
      "name": "FindDeletedFiles",
      "script": "deletedFiles = [f for f in inputs['indexedFileList'] if f['filePath'] not in [c['filePath'] for c in inputs['currentFileList']]]",
      "outputKey": "deletedFiles"
    },
    {
      "type": "vectorStore.delete",
      "name": "RemoveDeletedFiles",
      "vectorStoreName": "project-docs-2026",
      "inputKey": "deletedFiles"
    },
    {
      "type": "tool",
      "toolName": "RetrieveFilesTool",
      "parameters": {
        "sourceType": "all",
        "dateFrom": "${lastRunTimestamp}",
        "skip": 0,
        "limit": 50,
        "includeContent": "true"
      },
      "outputKey": "newOrUpdatedFiles"
    },
    {
      "type": "transform",
      "name": "DecodeBase64Content",
      "script": "for file in inputs['newOrUpdatedFiles']: file['decodedContent'] = Buffer.from(file['contentBase64'], 'base64').toString('utf-8')",
      "outputKey": "decodedFiles"
    },
    {
      "type": "vectorStore.index",
      "name": "IndexFiles",
      "inputKey": "decodedFiles",
      "contentField": "decodedContent",
      "metadataFields": ["fileName", "filePath", "sourceType", "lastModified"],
      "vectorStoreName": "project-docs-2026"
    },
    {
      "type": "transform",
      "name": "UpdateLastRunTimestamp",
      "script": "pipeline.variables.lastRunTimestamp = new Date().toISOString()"
    }
  ]
}

3. How It Works

  1. Retrieve current file list from Blob + SharePoint (metadata only).
  2. Get indexed file list from the vector store.
  3. Compare lists to find files that exist in the vector store but not in storage.
  4. Delete those entries from the vector store.
  5. Retrieve only new/updated files since the last run.
  6. Decode and index them into the vector store.
  7. Update last run timestamp for the next incremental run.

4. Benefits

  • Perfect sync between storage and vector store.
  • No stale documents in search results.
  • Incremental updates keep it fast and cost-efficient.
  • Fully automated — no manual cleanup needed.

we can now extend this even further so the pipeline runs on a schedule (e.g., every hour) and triggers a Foundry Agent retraining automatically when new documents are indexed.

=======OPTION 2========

Hosting a custom Model Context Protocol (MCP) server on Azure Functions is the native method for extending Microsoft Foundry Agents with custom tools. [1, 2]

When using Azure Functions for Microsoft Foundry, you can leverage the official Azure Functions MCP Extension, which automates request handling and maps seamlessly to Application Insights for end-to-end tracing. [1, 2]

Step 1: Initialize Your Project Locally

Ensure you have the Azure Functions Core Tools and Azure CLI installed.

  1. Create and enter your project directory:bashmkdir foundry-mcp-server cd foundry-mcp-server Use code with caution.
  2. Initialize an Azure Functions Python project (V2 Programming Model):bashfunc init --worker-runtime python --model V2 Use code with caution.
  3. Configure your requirements.txt:
    Open requirements.txt and ensure it includes the foundational Azure and OpenCensus tracking libraries:textazure-functions opencensus-ext-azure Use code with caution.

Step 2: Write the MCP Code

In the modern Microsoft Azure architecture, the Function App natively handles the HTTP/SSE streaming or webhook routing for MCP requests. [1, 2]

Replace the code in your function_app.py with this structure, which configures an HTTP endpoint, sets up explicit telemetry logs via Application Insights, and exposes an MCP tool structure:

python

import json
import logging
import azure.functions as func
from opencensus.ext.azure.log_exporter import AzureLogHandler

# 1. Setup Application Insights Telemetry
logger = logging.getLogger(__name__)
try:
    # Azure automatically provides APPLICATIONINSIGHTS_CONNECTION_STRING in the cloud
    logger.addHandler(AzureLogHandler())
except Exception:
    # Fallback to local streaming logs if debugging locally
    logging.basicConfig(level=logging.INFO)

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

# 2. Expose the MCP Endpoint Webhook
@app.route(route="mcp", methods=["POST", "GET"])
async def mcp_webhook(req: func.HttpRequest) -> func.HttpResponse:
    logger.info("Microsoft Foundry Agent triggered the MCP server webhook.")
    
    # Handle MCP initialization / Tool Discovery (GET)
    if req.method == "GET":
        tools_manifest = {
            "tools": [
                {
                    "name": "fetch_file_paths",
                    "description": "Retrieves source paths from Azure Blob and SharePoint folders.",
                    "inputSchema": {
                        "type": "object",
                        "properties": {
                            "system": {"type": "string", "enum": ["blob", "sharepoint", "both"]}
                        },
                        "required": ["system"]
                    }
                }
            ]
        }
        return func.HttpResponse(json.dumps(tools_manifest), mimetype="application/json", status_code=200)

    # Handle MCP Tool Execution (POST)
    try:
        req_body = req.get_json()
        method = req_body.get("method")
        params = req_body.get("params", {})
        
        # Log payload parameters directly to App Insights for auditable tracing
        logger.info(f"Executing MCP Method: {method} with params: {json.dumps(params)}")

        if method == "tools/call" or req_body.get("name") == "fetch_file_paths":
            # Extract arguments from your Agent's prompt
            arguments = params.get("arguments", req_body.get("arguments", {}))
            target_system = arguments.get("system", "both")

            # Mock data (Insert your physical storage fetching logic here)
            execution_result = {
                "content": [
                    {
                        "type": "text",
                        "text": f"Successfully retrieved file paths filtered for system: {target_system}"
                    }
                ]
            }
            return func.HttpResponse(json.dumps(execution_result), mimetype="application/json", status_code=200)
            
        return func.HttpResponse(json.dumps({"error": "Method not found"}), status_code=404)

    except Exception as e:
        logger.error(f"MCP Server Execution Failed: {str(e)}", exc_info=True)
        return func.HttpResponse(json.dumps({"error": str(e)}), status_code=500)

Use code with caution.

Step 3: Provision Infrastructure in Azure

Run the following Azure CLI commands to build the required serverless architecture.

  1. Create a Resource Group: bash az group create --name FoundryMCP-RG --location eastus Use code with caution.
  2. Create an Application Insights Instance: bash az monitor app-insights component create --app FoundryMCP-Insights --location eastus --resource-group FoundryMCP-RG Use code with caution.
  3. Create a Storage Account: bash az storage account create --name mcpstoreacc --location eastus --resource-group FoundryMCP-RG --sku Standard_LRS Use code with caution.
  4. Create the Function App (Linux / Python 3.11): bash az functionapp create --name MyFoundryMcpApp --storage-account mcpstoreacc --consumption-plan-location eastus --resource-group FoundryMCP-RG --os-type Linux --runtime python --runtime-version 3.11 --functions-version 4 --app-insights FoundryMCP-Insights Use code with caution.

Step 4: Deploy & Acquire the Webhook URL

  1. Deploy your code using the Core Tools:bashfunc azure functionapp publish MyFoundryMcpApp Use code with caution.
  2. Once deployed, note down your deployment endpoint and grab your Function/Host Key from the Azure Portal (Azure Portal > Function App > App Keys). Your exact connection string format will look like this:
    https://azurewebsites.net [1, 2]

Step 5: Connect Your MCP Server to Microsoft Foundry Agent

  1. Navigate to the Microsoft Foundry Portal (azure.com). [1]
  2. Open or create your custom Agent in the Playground. [1, 2]
  3. On the right-hand menu panel, click Tools > Add Tool. [1]
  4. Choose Model Context Protocol (MCP) tool from the tool definitions catalog. [1, 2]
  5. Paste your endpoint URL:
    • https://azurewebsites.net [1]
  6. Add the mandatory authorization header parameter so the Foundry platform can pass your secure Azure Key:
    • Header Name: x-functions-key
    • Value: [Your Azure Function App Key] [1]

The Agent will immediately query your function app endpoint (GET) to dynamically catalog your fetch_file_paths capabilities at runtime. [1]

Step 6: Monitor Executions via Application Insights

Whenever your Microsoft Foundry Agent runs a prompt requiring file path discovery, the execution trace flows live into Azure: [1]

  1. Go to Application Insights in the Azure Portal.
  2. Select Logs under the Monitoring section.
  3. Execute a KQL query to monitor what the Agent passed to your custom serverless backend:kustotraces | where message contains "Executing MCP Method" | project timestamp, message, severityLevel

Bundle inside foundry toolbox for multi-agent routing

In Microsoft Foundry (Azure AI Foundry), a Foundry Toolbox solves the tool-sprawl problem by bundling multiple resources—such as your custom Azure Function MCP server, Web Search, and Agent-to-Agent (A2A) connections—into a single versioned, MCP-compatible endpoint. [1, 2]

Instead of wiring individual tools directly to every agent, you attach the entire Toolbox. The platform then handles discovery, security, and multi-agent routing dynamically via natural language or autonomous orchestration. [1, 2, 3, 4]

Step 1: Create and Bundle Tools into the Toolbox

You can create and manage your Toolbox via the Azure AI Foundry Portal UI, the Foundry VS Code Toolkit, or the Azure CLI (azd). [1]

Method A: Using the VS Code Foundry Toolkit

  1. Open Visual Studio Code and click Foundry Toolkit in the Activity Bar. [1]
  2. Expand your project under My Resources, right-click Tools, and select the + Add Toolbox icon. [1]
  3. Name your toolbox (e.g., EnterpriseDataToolbox) and provide a clear description. [1]
  4. Click + Add tool and select Remote MCP Server. Paste your Azure Function App endpoint:
    • https://azurewebsites.net [1]
  5. (Optional for Multi-Agent Routing) Select + Add tool again and choose Agent-to-Agent (A2A) to register other specialized downstream sub-agents as tools within this specific Toolbox. [1, 2]
  6. Click Publish to generate a static, production-ready Toolbox version and copy its unique versioned MCP endpoint:
    https://<account>.services.ai.azure.com/api/projects/<project>/toolboxes/<toolbox-name>/versions/1/mcp?api-version=v1 [1, 2]

Step 2: Enable “Tool Search” and Governance

As your toolbox grows to handle complex multi-agent workflows, cramming every tool or sub-agent description into the prompt will blow out token costs. [1, 2]

  1. Within your Toolbox Settings in the Microsoft Foundry portal, toggle on Tool Search (Preview). [1]
  2. How it routes: When the main agent receives a prompt, the Toolbox uses a localized search algorithm (like BM25) to look at the user query and inject only the specific tool/sub-agent definitions needed for that turn—routing the request dynamically without overwhelming the model context. [1]
  3. Assign any Foundry Guardrails directly to the Toolbox layer. This ensures that inputs and outputs going to any routed tool or sub-agent are automatically governed under a single policy. [1, 2]

Step 3: Implement Multi-Agent Routing in Code

Using the Microsoft Agent Framework (MAF), you can instantiate an orchestrator agent, hand it the central Foundry Toolbox, and let it route traffic automatically. [1, 2]

1. Configure the Environment and Orchestrator Logic

Set up your environment variables including your project endpoint, deployment name, and the versioned Toolbox MCP endpoint. Initialize the AIProjectClient with Azure credentials, instantiate the FoundryToolbox using your endpoint, and create your orchestrator agent with the toolbox attached as a tool. [1, 2, 3]

2. Execution and Verification Flow

When a complex user prompt is submitted, the framework handles the interaction centrally:

  1. Centralized Auth: The agent connects using managed identity tokens, while individual access to downstream functions and sub-agents is managed via Microsoft Entra ID passthrough. [1, 2, 3]
  2. Execution: The Toolbox evaluates intent, filters definitions via tool search, applies guardrails, and safely relays parameters and outputs between the orchestrator and your custom MCP servers or A2A sub-agents. [1]

NEXT: explore setting up an Agent-to-Agent (A2A) tool explicitly so that two independent codebases can chat inside this toolbox, or you can configure a Model Router inside this architecture to balance costs between gpt-4o and smaller models like o4-mini? [1, 2]

Leave a Reply

Your email address will not be published. Required fields are marked *

Related Post

Search

Latest Posts

Archieve

Category

Tags

Social

Powered by
►
Necessary cookies enable essential site features like secure log-ins and consent preference adjustments. They do not store personal data.
None
►
Functional cookies support features like content sharing on social media, collecting feedback, and enabling third-party tools.
None
►
Analytical cookies track visitor interactions, providing insights on metrics like visitor count, bounce rate, and traffic sources.
None
►
Advertisement cookies deliver personalized ads based on your previous visits and analyze the effectiveness of ad campaigns.
None
Powered by