Skip to main content Home Creators azure connectors-python-sdk add-connector
add-connector Add a new connector to the Azure Connectors Python SDK. USE WHEN: adding a new connector client, scaffolding connector files, creating unit tests, updating __init__.py exports, updating README status table, creating sample usage files. Covers the full workflow from connector file creation through test validation. NOT FOR: modifying existing connectors, connection setup, or trigger registration.
Jump to install Skills Marketplace Discover and explore AI skills built by the community.
Install with Codex or Claude Copy this prompt, paste it into Codex, Claude, or another assistant, and let it review the skill page and install it for you.
Copy promptShow prompt details A direct command skips the review prompt. Inspect the source before running it.
npx skills add https://github.com/Azure/connectors-python-sdk --skill add-connectorThe command stays on one line. Scroll horizontally to inspect it before copying.
Prefer a local copy? Download the files currently available to SkillsMP.
Download Zip Downloading... More from this repository
Create and configure Connector Namespace connections for the Azure Connectors Python SDK. USE WHEN: setting up a new connector connection, creating a Connector Namespace, authorizing OAuth consent, adding access policies, or configuring local.settings.json / app settings. Covers Office365, SharePoint, Teams, and any Microsoft.Web/connections connector. Works with Azure Functions, Flask, FastAPI, Django, or any Python app using the SDK. NOT FOR: trigger registration (use trigger-registration skill), or code generation.
Related occupations SOC
Based on SOC occupation classification
name add-connector description Add a new connector to the Azure Connectors Python SDK. USE WHEN: adding a new connector client, scaffolding connector files, creating unit tests, updating __init__.py exports, updating README status table, creating sample usage files. Covers the full workflow from connector file creation through test validation. NOT FOR: modifying existing connectors, connection setup, or trigger registration. argument-hint Provide the connector name (e.g., "dynamics365", "servicenow")
Add New Connector Skill
Automates the complete workflow for adding a new connector to the Azure Connectors Python SDK.
When to Use
Adding a completely new connector to the SDK
Scaffolding all required files for a new connector
Creating comprehensive unit tests for a new connector
Creating sample usage files from .NET examples
Prerequisites
The connector module file (e.g., dynamics365.py) is provided or needs to be created
The connector follows the ConnectorClientBase pattern
A .NET sample exists to port to Python (optional)
Procedure
Step 1: Gather Connector Information
Collect the following details from the user:
Field Description Example connector_nameSnake_case module name dynamics365, servicenowclient_class_name
Dynamics365Client, ServicenowClient
display_nameHuman-readable name Dynamics 365, ServiceNow
package_pathImport path azure.connectors.dynamics365
env_var_nameEnvironment variable DYNAMICS365_CONNECTION_URL
statusValidation status ✅ E2E Validated or 🔄 SDK Generated
test_countNumber of tests 30 tests
Step 2: Create/Verify Connector Module Ensure the connector file exists at src/azure/connectors/{connector_name}.py.
The connector must follow this structure:
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Optional , Any , Dict , List
from urllib.parse import quote
import json
from azure.connectors.sdk import (
ConnectorClientBase,
ConnectorClientOptions,
TokenProvider,
ManagedIdentityTokenProvider,
ConnectorException,
)
class {ClientClassName}(ConnectorClientBase):
"""Client for {DisplayName} connector operations."""
@property
def connector_name (self ) -> str :
return "{connector_name}"
Step 3: Update __init__.py Exports Add the connector import to src/azure/connectors/__init__.py:
try :
from .{connector_name} import {ClientClassName}
except (ImportError, NameError):
{ClientClassName} = None
Also add to the __all__ list if present.
Step 4: Update README Connector Table Add a row to the "Validated Connectors" table in README.md:
| **{DisplayName}** | `{package_path}` | ✅ Complete | {status} | {test_count} |
Insert alphabetically by display name. Update the "Total:" line with the new test count.
Step 5: Create Unit Tests Create tests/test_{connector_name}.py following this template:
"""Unit tests for {ClientClassName}."""
import pytest
from unittest.mock import AsyncMock, patch
from azure.connectors.{connector_name} import (
{ClientClassName},
)
from azure.connectors.sdk import (
ConnectorClientOptions,
ManagedIdentityTokenProvider,
ConnectorException,
)
from tests.conftest import MockResponse
class Test {ClientClassName}Initialization:
"""Tests for {ClientClassName} initialization."""
def test_init_with_valid_url_and_defaults (self ):
"""Test initialization with valid URL and default parameters."""
client = {ClientClassName}("https://example.azure.com/connections/test" )
assert client._connection_runtime_url == "https://example.azure.com/connections/test"
assert client.connector_name == "{connector_name}"
assert isinstance (client._http_client._token_provider, ManagedIdentityTokenProvider)
def test_init_with_trailing_slash (self ):
"""Test that trailing slash is removed from URL."""
client = {ClientClassName}("https://example.azure.com/connections/test/" )
assert client._connection_runtime_url == "https://example.azure.com/connections/test"
def test_init_with_custom_token_provider (self, mock_token_provider ):
"""Test initialization with custom token provider."""
client = {ClientClassName}(
"https://example.azure.com/connections/test" ,
token_provider=mock_token_provider
)
assert client._http_client._token_provider is mock_token_provider
def test_init_with_custom_options (self, mock_token_provider ):
"""Test initialization with custom options."""
options = ConnectorClientOptions(timeout_seconds=60.0 , max_retry_attempts=5 )
client = {ClientClassName}(
"https://example.azure.com/connections/test" ,
token_provider=mock_token_provider,
options=options
)
assert client._options is options
assert client._options.timeout_seconds == 60.0
assert client._options.max_retry_attempts == 5
def test_init_with_empty_url_raises_error (self ):
"""Test that empty URL raises ValueError."""
with pytest.raises(ValueError, match ="connection_runtime_url cannot be None or empty" ):
{ClientClassName}("" )
def test_init_with_none_url_raises_error (self ):
"""Test that None URL raises ValueError."""
with pytest.raises(ValueError, match ="connection_runtime_url cannot be None or empty" ):
{ClientClassName}(None )
def test_connector_name_property (self, mock_token_provider ):
"""Test connector_name property returns '{connector_name}'."""
client = {ClientClassName}(
"https://example.azure.com/connections/test" ,
token_provider=mock_token_provider
)
assert client.connector_name == "{connector_name}"
class Test {ClientClassName}Lifecycle:
"""Tests for {ClientClassName} lifecycle methods."""
@pytest.mark.asyncio
async def test_close (self, mock_token_provider ):
"""Test close method calls http_client.close."""
client = {ClientClassName}(
"https://example.azure.com/connections/test" ,
token_provider=mock_token_provider
)
with patch.object (client._http_client, 'close' , new_callable=AsyncMock) as mock_close:
await client.close()
mock_close.assert_called_once()
@pytest.mark.asyncio
async def test_context_manager (self, mock_token_provider ):
"""Test async context manager functionality."""
with patch.object ({ClientClassName}, 'close' , new_callable=AsyncMock) as mock_close:
async with {ClientClassName}(
"https://example.azure.com/connections/test" ,
token_provider=mock_token_provider
) as client:
assert isinstance (client, {ClientClassName})
mock_close.assert_called_once()
Required test categories:
Initialization tests — constructor, URL handling, options, error cases
Lifecycle tests — close(), context manager
Method tests — one class per public async method with:
Success case with mocked response
Parameter handling tests
Error response handling
Step 6: Create Sample Usage File Create samples/sample_connector_usage/sample_connector_usage_{connector_name}.py:
"""
{DisplayName} Connector SDK Sample
This sample demonstrates how to use the {DisplayName} connector SDK.
Prerequisites:
1. Azure subscription with {DisplayName} connection
2. {DisplayName} connection in Connector Namespaces
3. Connection runtime URL from Azure Portal
Installation:
pip install azure-connectors
Usage:
Set environment variable:
$env:{ENV_VAR_NAME} = "https://[region].azure-apihub.net/apim/{connector_name}/[connection-id]"
python sample_connector_usage_{connector_name}.py
"""
import asyncio
import os
from azure.identity.aio import DefaultAzureCredential
from azure.connectors import ConnectorException
from azure.connectors.{connector_name} import (
{ClientClassName},
)
CONNECTION_RUNTIME_URL = os.environ.get(
"{ENV_VAR_NAME}" ,
""
)
async def example_1_basic_operation ():
"""Example 1: Basic connector operation."""
print ("\n=== Example 1: Basic Operation ===" )
credential = DefaultAzureCredential()
async with {ClientClassName}(CONNECTION_RUNTIME_URL, credential) as client:
try :
result = await client.{example_method}_async ()
print (f"Result: {{result}}" )
except ConnectorException as ex:
print (f"Connector error: {{ex}}" )
except Exception as ex:
print (f"Error: {{ex}}" )
async def example_2_advanced_operation ():
"""Example 2: Advanced connector operation."""
print ("\n=== Example 2: Advanced Operation ===" )
credential = DefaultAzureCredential()
async with {ClientClassName}(CONNECTION_RUNTIME_URL, credential) as client:
try :
print ("Advanced operation completed" )
except ConnectorException as ex:
print (f"Connector error: {{ex}}" )
except Exception as ex:
print (f"Error: {{ex}}" )
async def main ():
"""Run all examples."""
if not CONNECTION_RUNTIME_URL:
print ("Error: {ENV_VAR_NAME} environment variable not set." )
print ("Set it to your connection runtime URL from Azure Portal." )
return
print (f"Using connection URL: {{CONNECTION_RUNTIME_URL[:50]}}..." )
await example_1_basic_operation()
await example_2_advanced_operation()
print ("\n=== All examples completed ===" )
if __name__ == "__main__" :
asyncio.run(main())
Find the corresponding .NET sample
Translate each example method to Python async/await syntax
Convert C# types to Python dataclasses
Use async with for client lifecycle
Handle exceptions with ConnectorException
Step 7: Update Samples README Add a row to samples/sample_connector_usage/README.md:
| `sample_connector_usage_{connector_name}.py` | {DisplayName} | `{ENV_VAR_NAME}` |
Insert alphabetically by connector name.
Step 8: Update CHANGELOG Add the new connector to the [Unreleased] section in CHANGELOG.md under ### Added:
## [Unreleased]
### Added
- **{DisplayName}** (`{connector_name}.py` ) connector client with unit tests and samples
If there are multiple new connectors being added together, combine them in a single bullet point:
- **3 new connector clients** with unit tests and samples:
- {DisplayName}, Other Connector, Another Connector
Step 9: Update Connection Setup Skill Add the connector's API name to the supported SDK connector names list in .github/skills/connection-setup/SKILL.md (Step 2).
Find the line that starts with Supported SDK connector names: and add {connector_name} alphabetically to the list:
Supported SDK connector names: `arm` , `azuread` , ..., `{connector_name}` , ... (and any `Microsoft.Web/connections` connector name).
Step 10: Run Tests Execute the full test suite to validate:
.\.venv\Scripts\python -m pytest tests/test_{connector_name}.py tests/test_code_quality.py -v --tb=short
Step 11: Validate Sample Syntax python -m py_compile samples/sample_connector_usage/sample_connector_usage_{connector_name}.py
.\.venv\Scripts\python.exe -m flake8 samples/sample_connector_usage/sample_connector_usage_{connector_name}.py --max-line-length=100
Checklist Before completing, verify:
Files Modified File Change src/azure/connectors/{connector_name}.pyNew connector module src/azure/connectors/__init__.pyAdd export README.mdAdd to connector table tests/test_{connector_name}.pyNew test file samples/sample_connector_usage/sample_connector_usage_{connector_name}.pyNew sample samples/sample_connector_usage/README.mdAdd to samples table CHANGELOG.mdAdd to [Unreleased] section .github/skills/connection-setup/SKILL.mdAdd connector name to Step 2
Common Issues
Import Error in __init__.py If the connector has undefined type references, the try/except wrapper handles it gracefully. The connector will be None until fixed.
Test Discovery Ensure tests/__init__.py exists. Test files must start with test_.
Sample File Errors Check that all imported types exist in the connector module. Use only public dataclasses.