How to use queue triggers in Azure Durable Python Functions with managed identity

I'm developing a queue trigger for an Azure Durable Function. When using the "queue_trigger" decorator, I'm required to provide a connection string. However, I need to avoid using a connection string directly and instead use Managed Identity to connect to the Storage Account/Queue Storage, ideally with DefaultAzureCredential() or a similar method. In the example below, I currently have the "QueueConnectionString" set up in the environment variables, but I want to replace this with Managed Identity for secure access to the queue.

import azure.functions as func
import logging
import azure.durable_functions as adf

myApp = adf.DFApp(http_auth_level=func.AuthLevel.ANONYMOUS)
@myApp.durable_client_input(client_name="client")
@myApp.queue_trigger(arg_name="azqueue", queue_name="test", connection="QueueConnectionString")
async def begin_data_entry(azqueue: func.QueueMessage, client):
    logging.info('Python HTTP trigger function processed a request.')
    await client.start_new("activity_function_name", client_input={})

I created a sample queue trigger for an Azure Durable Function with Managed identity using DefaultAzureCredential to process a message to the Azure Storage Queue and it worked fine for me.

• It's better to add the storage URI QueueConnectionString__queueServiceUri in the local.settings.json as below.

local.settings.json :

{
    "IsEncrypted": false,
  "Values": {
    "AzureWebJobsStorage": "UseDevelopmentStorage=true",
    "FUNCTIONS_WORKER_RUNTIME": "python",
    "QueueConnection__queueServiceUri": "https://<storage_name>.queue.core.windows.net/"
  }
}

Add the below connection in the code, it will load the storage URI from local.settings.json.

connection="QueueConnection__queueServiceUri"

Code :

import azure.functions as func
import logging
import azure.durable_functions as adf
from azure.identity import DefaultAzureCredential
from azure.storage.queue import QueueClient

myApp = adf.DFApp(http_auth_level=func.AuthLevel.ANONYMOUS)
@myApp.durable_client_input(client_name="client")
@myApp.queue_trigger(arg_name="azqueue", queue_name="test", connection="QueueConnection__queueServiceUri")
async def begin_data_entry(azqueue: func.QueueMessage, client):
    logging.info('Python HTTP trigger function processed a request.')
    await client.start_new("activity_function_name", client_input={})

I have added the owner role to the service principle and Storage Queue Data Contributor role to the function app in the Azure Storage account as shown below.

Make sure to add the below URI to the Azure Function App > Environment Variables > App settings as shown below.

"QueueConnection__queueServiceUri": "https://<storage_name>.queue.core.windows.net/"