add-connector

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_name	Snake_case module name	dynamics365, servicenow
client_class_name	PascalCase client class	Dynamics365Client, ServicenowClient
display_name	Human-readable name	Dynamics 365, ServiceNow
package_path	Import path	azure.connectors.dynamics365
env_var_name	Environment variable	DYNAMICS365_CONNECTION_URL
status	Validation status	✅ E2E Validated or 🔄 SDK Generated
test_count	Number 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:

# Copyright (c) Microsoft Corporation. All rights reserved.

# {ClientClassName}.py - Auto-generated DirectClient SDK
# Do not edit this file directly.

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,
)


# Type Definitions
# ... dataclasses for request/response types ...


class {ClientClassName}(ConnectorClientBase):
    """Client for {DisplayName} connector operations."""

    @property
    def connector_name(self) -> str:
        return "{connector_name}"

    # ... async methods ...

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  # type: ignore[assignment,misc]

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:

# Copyright (c) Microsoft Corporation. All rights reserved.

"""Unit tests for {ClientClassName}."""

import pytest
from unittest.mock import AsyncMock, patch
from azure.connectors.{connector_name} import (
    {ClientClassName},
    # Import all public dataclasses from the connector
)
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}(
            ,
            token_provider=mock_token_provider
        )

         client._http_client._token_provider  mock_token_provider

     ():
        
        options = ConnectorClientOptions(timeout_seconds=, max_retry_attempts=)
        client = {ClientClassName}(
            ,
            token_provider=mock_token_provider,
            options=options
        )

         client._options  options
         client._options.timeout_seconds == 
         client._options.max_retry_attempts == 

     ():
        
         pytest.raises(ValueError, =):
            {ClientClassName}()

     ():
        
         pytest.raises(ValueError, =):
            {ClientClassName}()

     ():
        
        client = {ClientClassName}(
            ,
            token_provider=mock_token_provider
        )

         client.connector_name == 


 {ClientClassName}Lifecycle:
    


      ():
        
        client = {ClientClassName}(
            ,
            token_provider=mock_token_provider
        )

         patch.(client._http_client, , new_callable=AsyncMock)  mock_close:
             client.close()
            mock_close.assert_called_once()


      ():
        
         patch.({ClientClassName}, , new_callable=AsyncMock)  mock_close:
              {ClientClassName}(
                ,
                token_provider=mock_token_provider
            )  client:
                 (client, {ClientClassName})

            mock_close.assert_called_once()

Required test categories:

1. Initialization tests — constructor, URL handling, options, error cases
2. Lifecycle tests — close(), context manager
3. 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:

# Copyright (c) Microsoft Corporation. All rights reserved.

"""
{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},
    # Import relevant dataclasses for the examples
)

# Connection runtime URL format:
# https://[region].azure-apihub.net/apim/{connector_name}/[connection-id]
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:
            # Call a basic read operation
            result = await client.{example_method}_async()
            
            print(f"Result: {{result}}")
                
        except ConnectorException as ex:
            print(f"Connector error: {{ex}}")
         Exception  ex:
            ()


  ():
    
    ()
    
    credential = DefaultAzureCredential()
    
      {ClientClassName}(CONNECTION_RUNTIME_URL, credential)  client:
        :
            
            
            ()
            
         ConnectorException  ex:
            ()
         Exception  ex:
            ()


  ():
    
      CONNECTION_RUNTIME_URL:
        ()
        ()
        

    ()
    
     example_1_basic_operation()
     example_2_advanced_operation()
    
    ()


 __name__ == :
    asyncio.run(main())

When porting from .NET:

1. Find the corresponding .NET sample
2. Translate each example method to Python async/await syntax
3. Convert C# types to Python dataclasses
4. Use async with for client lifecycle
5. 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

Required checks:

• All tests pass
• No import errors
• Code quality tests pass (imports, structure)

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:

• src/azure/connectors/{connector_name}.py exists with proper structure
• src/azure/connectors/__init__.py updated with try/except import
• README.md connector table updated with correct status and test count
• tests/test_{connector_name}.py created with comprehensive tests
• samples/sample_connector_usage/sample_connector_usage_{connector_name}.py created
• samples/sample_connector_usage/README.md updated
• CHANGELOG.md updated with new connector in [Unreleased] section
• .github/skills/connection-setup/SKILL.md Step 2 updated with connector API name
• All tests pass
• Sample file compiles without errors

Files Modified

File	Change
src/azure/connectors/{connector_name}.py	New connector module
src/azure/connectors/__init__.py	Add export
README.md	Add to connector table
tests/test_{connector_name}.py	New test file
samples/sample_connector_usage/sample_connector_usage_{connector_name}.py	New sample
samples/sample_connector_usage/README.md	Add to samples table
CHANGELOG.md	Add to [Unreleased] section
.github/skills/connection-setup/SKILL.md	Add 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.