Connector Namespace Trigger Registration

Registers polling trigger configs on a Connector Namespace so that connector events (new email, new file, etc.) call back to your application endpoint. Covers scaffolding a Python Function App with the ConnectorTrigger extension binding.

When to Use

• Developer needs a connector trigger (e.g., "when a new email arrives in Office365")
• Developer has an existing Connector Namespace connection (use the connection-setup skill first if not)
• Developer needs to scaffold a Python Function App with ConnectorTrigger in Azure Functions
• Developer needs to wire the callback URL from a deployed or local Function App
• Developer needs to understand which typed payload to use for a given trigger operation

Prerequisites

• Azure CLI installed and authenticated (az login)
• Connector Namespace with a connected connector (see connection-setup skill)
• The Connector Namespace must have a system-assigned managed identity enabled
• Supported regions for Connector Namespace: westcentralus
• A supported Python version for Azure Functions with the azure-connectors package installed

Key Concepts

ConnectorTrigger Extension Binding

Python Decorator and Package Rules

Use this decision logic when scaffolding a Python connector trigger function:

1. Add connector_trigger decorator:
   • @app.connector_trigger
2. Choose packages by trigger operation:
   • Office 365 OnNewEmail → add azurefunctions-extensions-connectors (imports azure-connectors automatically)
   • Any other connector with a typed SDK model → add azure-connectors, use typed class as parameter type
   • No typed model available → use str as parameter type, no extra package needed beyond azure-functions
3. Base package: azure-functions
   • Python 3.13 → >=2.2.0b4
   • Python < 3.13 → >=1.26.0b3

Extension Webhook Endpoint

The ConnectorTrigger extension registers a webhook route on the Function App:

POST /runtime/webhooks/connector?functionName={FunctionName}&code={connector_extension_key}

• functionName must exactly match the @app.function_name(name="...") value
• connector_extension is a system key auto-generated when the extension loads
• Locally (func start), the system key is not enforced

Trigger Config vs Connection

Connector Namespace
├── connections/
│   └── office365-conn         ← auth + runtime URL (connection-setup skill)
└── triggerConfigs/
    └── onnewemail-trigger     ← poll + callback config (THIS skill)

Scaffolding a Python Function App

1. Initialize with azd

azd init -t functions-quickstart-python-http-azd

2. Update host.json for preview extension bundle

{
    "version": "2.0",
    "extensionBundle": {
        "id": "Microsoft.Azure.Functions.ExtensionBundle.Preview",
        "version": "[4.*, 5.0.0)"
    }
}

3. Install packages

Add to requirements.txt (include packages based on your approach):

# >=2.2.0b4 required for Python 3.13+, >=1.26.0b3 for Python < 3.13
azure-functions>=2.2.0b4

# Currently only supports Office 365 OnNewEmail operation
azurefunctions-extensions-connectors

# Required for str payloads, don't include if using azurefunctions-extensions-connectors 
azure-connectors

4. Create a ConnectorTrigger function

With typed SDK payload (Office 365 email example)

import azure.functions as func
import azurefunctions.extensions.connectors.office365 as office365
import logging
from typing import List

app = func.FunctionApp()


@app.function_name(name="OnNewEmail")
@app.connector_trigger(arg_name="emails")
def on_new_email(emails: List[office365.ClientReceiveMessage]) -> None:
    logging.info("OnNewEmail trigger received")

    for email in emails:
        logging.info(f"Subject: {email.subject}")
        logging.info(f"From: {email.from_}")

With connector trigger and raw JSON (when no typed SDK model exists)

import azure.functions as func
import json
import logging

app = func.FunctionApp()


@app.function_name(name="OnNewFile")
@app.connector_trigger(arg_name="payload")
def on_new_file(payload: str) -> None:
    logging.info("OnNewFile trigger received")

    data = json.loads(payload)
    body = data.get("body", {})

    if isinstance(body, dict) and "value" in body:
        # Metadata trigger — batch of items
        for item in body["value"]:
            logging.info(f"File: {item.get('name')}")
    elif isinstance(body, str):
        # Binary trigger — base64-encoded file content
        import base64
        content = base64.b64decode(body)
        logging.info(f"Received file content: {len(content)} bytes")

5. Run locally

Start Azurite (required for AzureWebJobsStorage):

npx azurite

Verify local.settings.json:

• Python 3.13+ example:

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

• Python < 3.13 example - PYTHON_ISOLATE_WORKER_DEPENDENCIES enabled:

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

Start the Function App:

func start

The extension logs the webhook endpoint at startup:

Connector endpoint: http://localhost:7071/runtime/webhooks/connector

Registering a Trigger Config

Step 1: Get the Callback URL

Deployed Function App

$resourceGroup = "<resource-group>"
$functionAppName = "<function-app-name>"
$functionName = "<function-name>"  # must match @app.function_name(name="...")

$connectorExtensionKey = az functionapp keys list -g $resourceGroup -n $functionAppName --query "systemKeys.connector_extension" -o tsv
$callbackUrl = "https://$functionAppName.azurewebsites.net/runtime/webhooks/connector?functionName=$functionName&code=$connectorExtensionKey"

Local development (with dev tunnel)

$tunnelUrl = "<your-tunnel-url>"  # e.g., https://<id>-7071.uks1.devtunnels.ms
$functionName = "<function-name>"
$callbackUrl = "$tunnelUrl/runtime/webhooks/connector?functionName=$functionName"

Step 2: Create Trigger Config

$subscriptionId = "<subscription-id>"
$resourceGroup = "<resource-group>"
$namespaceName = "<namespace-name>"
$nsId = "/subscriptions/$subscriptionId/resourceGroups/$resourceGroup/providers/Microsoft.Web/connectorGateways/$namespaceName"

$triggerName = "<trigger-config-name>"   # e.g., "onnewemail-trigger"
$connectionName = "<connection-name>"    # e.g., "office365-conn"
$connectorName = "<connector-name>"      # e.g., "office365"
$operationName = "<operation-name>"      # e.g., "OnNewEmailV3"

$token = az account get-access-token `
    --resource "https://management.core.windows.net/" `
    --query "accessToken" -o tsv

$body = @{
    properties = @{
        operationName = $operationName
        connectionDetails = @{
            connectorName = $connectorName
            connectionName = $connectionName
        }
        notificationDetails = @{
            callbackUrl = $callbackUrl
            httpMethod = "Post"
        }
        parameters = @(
            # Office 365 OnNewEmailV3:
            # @{ name = "folderPath"; value = "Inbox" }

            # OneDrive for Business OnNewFile:
            # @{ name = "folderId"; value = "root" }
        )
    }
} | ConvertTo-Json -Depth 4

$uri = "https://management.azure.com${nsId}/triggerConfigs/${triggerName}?api-version=2026-05-01-preview"
try {
    $response = Invoke-WebRequest -Uri $uri -Method PUT -Body $body `
        -ContentType "application/json" `
        -Headers @{ Authorization = "Bearer $token" }
    Write-Output "Status: $($response.StatusCode)"
} catch {
    Write-Output "Error: $($_.Exception.Response.StatusCode)"
    $_.ErrorDetails.Message
}

Step 3: Verify Trigger Config

az rest --method GET `
    --uri "https://management.azure.com${nsId}/triggerConfigs/${triggerName}?api-version=2026-05-01-preview" `
    --query "properties.{operation:operationName, state:state, callback:notificationDetails.callbackUrl}" `
    -o table

Expected: state = Enabled.

Step 4: Test the Trigger

Trigger the connector event (e.g., send an email, upload a file). The Connector Namespace polls every 1-5 minutes.

Watch Function App logs:

func start  # local
# or for deployed:
az functionapp log tail -g $resourceGroup -n $functionAppName

Using the SDK for Actions (Beyond Triggers)

The azure-connectors package provides async-only root clients for calling connector actions directly from Python applications. Clients accept caller-owned AsyncTokenCredential or AzureKeyCredential credentials; no synchronous mirror is provided.

Example: Send an email

from azure.identity.aio import DefaultAzureCredential

from azure.connectors.office365 import Office365Client
from azure.connectors.office365 import SendEmailInput


async def main(connection_url: str):
    async with DefaultAzureCredential() as credential:
        async with Office365Client(connection_url, credential) as client:
            await client.send_email(
                input=SendEmailInput(
                    to="recipient@example.com",
                    subject="Hello from Python SDK",
                    body="<p>Sent from any Python app.</p>",
                )
            )

Example: List SharePoint items

from azure.identity.aio import DefaultAzureCredential

from azure.connectors.sharepointonline import SharepointonlineClient


async def list_items(connection_url: str):
    async with DefaultAzureCredential() as credential:
        async with SharepointonlineClient(connection_url, credential) as client:
            items = [
                item
                async for item in client.list_items(
                    dataset="https://contoso.sharepoint.com/sites/MySite",
                    table="MyList",
                )
            ]

    for item in items:
        print(f"Item: {item.get('Title')}")

Authentication Options

Use credentials from azure.identity.aio. Use AzureKeyCredential from azure.core.credentials when the connection runtime expects an API key. The caller owns async credentials and must close them with async with or await credential.close().

E2E Validated Connectors

All SDKs support generating clients for any of the 1,000+ Azure managed connectors.

Trigger Payload Shapes

The Connector Namespace delivers callbacks in two shapes depending on the connector's splitOn setting:

The SDK's TriggerCallbackPayload.from_json() and TriggerCallbackBody.from_dict() methods normalize both shapes into a list, so consumers always iterate:

from azure.connectors.sdk import TriggerCallbackPayload

payload = TriggerCallbackPayload.from_json(raw_json)
for item in payload.body.value:
    print(item)

Common Errors

Reference

For a complete mapping of trigger operations to function signatures across .NET, Python, and TypeScript, see Operations to Functions Signature Match.