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 can be generated as one async root module
• The client follows the async ConnectorClientBase runtime patterns
• A .NET sample exists to port to Python (optional)

Procedure

Step 1: Gather Connector Information

Collect the following details from the user:

Step 2: Create/Verify Connector Module

Generate one async root module from CodefulSdkGenerator:

• src/azure/connectors/{connector_name}.py

The root connector owns models, metadata, and the async-only client:

# 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.core.credentials import AzureKeyCredential
from azure.core.credentials_async import AsyncTokenCredential
from azure.core.pipeline.transport import AsyncHttpTransport

from azure.connectors.sdk import ConnectorClientBase, 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}"

    # ... asynchronous unsuffixed methods ...

The module defines models, aliases, enums, and TRIGGER_OPERATIONS. Its client uses AsyncTokenCredential and AsyncHttpTransport. Pageable operations return AsyncItemPaged without awaiting construction. Await begin_* operations to obtain an AsyncLROPoller. Do not create a synchronous mirror.

Step 3: Keep Package Initializers Minimal

Consumers import clients and models directly from each root connector module. Regenerate src/azure/connectors/_client_registry.py and src/azure/connectors/__init__.pyi with the connector inventory; do not hand-edit generated package exports. Preserve the lazy loader in src/azure/connectors/__init__.py. Do not add eager generated-client barrel imports or try/except imports that replace failed clients with None.

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}."""

from unittest.mock import AsyncMock, patch

import pytest
from azure.core.credentials import AzureKeyCredential
from azure.connectors.{connector_name} import {ClientClassName}
from azure.connectors.{connector_name} import (
    {RequestModelName},
)
from azure.connectors.sdk import 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",
            AzureKeyCredential("test-key"),
        )

        assert client._connection_runtime_url == "https://example.azure.com/connections/test"
        assert client.connector_name == "{connector_name}"
        assert isinstance(client._http_client._credential, AzureKeyCredential)

    def test_init_with_trailing_slash(self):
        """Test that trailing slash is removed from URL."""
        client = {ClientClassName}(
            "https://example.azure.com/connections/test/",
            AzureKeyCredential("test-key"),
        )

        assert client._connection_runtime_url == "https://example.azure.com/connections/test"

    def test_init_with_custom_credential(self, mock_credential):
        """Test initialization with custom Azure Core credential."""
        client = {ClientClassName}(
            "https://example.azure.com/connections/test",
            credential=mock_credential
        )

        assert client._http_client._credential is mock_credential

    def test_init_with_custom_settings(self, mock_credential):
        """Test initialization with custom pipeline settings."""
        client = {ClientClassName}(
            "https://example.azure.com/connections/test",
            credential=mock_credential,
            timeout_seconds=60.0,
            max_retry_attempts=5,
        )

        assert client._http_client._timeout_seconds == 60.0

    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}("", AzureKeyCredential("test-key"))

    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, AzureKeyCredential("test-key"))

    def test_connector_name_property(self, mock_credential):
        """Test connector_name property returns '{connector_name}'."""
        client = {ClientClassName}(
            "https://example.azure.com/connections/test",
            credential=mock_credential
        )

        assert client.connector_name == "{connector_name}"


class Test{ClientClassName}Lifecycle:
    """Tests for {ClientClassName} lifecycle methods."""

    @pytest.mark.asyncio
    async def test_close(self, mock_credential):
        """Test close method calls http_client.close."""
        client = {ClientClassName}(
            "https://example.azure.com/connections/test",
            credential=mock_credential
        )

        with patch.object(client._http_client, 'close', new_callable=AsyncMock) as mock_close:
            await client.close()
            mock_close.assert_awaited_once()

    @pytest.mark.asyncio
    async def test_context_manager(self, mock_credential):
        """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",
                credential=mock_credential
            ) as client:
                assert isinstance(client, {ClientClassName})

            mock_close.assert_awaited_once()


# Add test classes for each public method in the connector
# Follow the pattern:
#
# class Test{MethodName}:
#     """Tests for {method_name} on the root async client."""
#
#     @pytest.mark.asyncio
#     async def test_success(self, mock_credential):
#         """Test successful request."""
#         ...
#
#     @pytest.mark.asyncio
#     async def test_error_response_raises_exception(self, mock_credential):
#         """Test that error response raises ConnectorException."""
#         ...

Required test categories:

1. Initialization tests — constructor, URL handling, options, error cases
2. Lifecycle tests — await client.close(), async with, and awaited cleanup
3. Method tests — exercise root async operations with:
   • Success cases with asynchronous mocked responses
   • Keyword-only optional parameters and request controls
   • Azure Core error response handling
   • Models declared in the same root module
   • AsyncItemPaged iteration without awaiting construction, including continuation requests, empty pages, later-page items, and termination
   • Retained Location-backed operations: await begin_* for an AsyncLROPoller, then await poller.result() while the client remains open
4. Export tests — direct imports and lazy package-root clients match the generated registry and type stub; broken imports raise instead of returning None

Step 6: Create Sample Usage File

Create one async sample file:

• samples/sample_connector_usage/sample_connector_usage_{connector_name}.py

Use azure.identity.aio.DefaultAzureCredential, import the client from its root connector module, and retain async with and await:

# 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}
from azure.connectors.{connector_name} import (
    {RequestModelName},
)

# 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 ===")
    
    async with DefaultAzureCredential() as credential, {ClientClassName}(
        CONNECTION_RUNTIME_URL, credential
    ) as client:
        try:
            # Call a basic read operation
            result = await client.{example_method}()
            
            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 ===")
    
    async with DefaultAzureCredential() as credential, {ClientClassName}(
        CONNECTION_RUNTIME_URL, credential
    ) as client:
        try:
            # Call a write or complex operation
            # result = await client.{another_method}(...)
            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())

When porting from .NET:

1. Find the corresponding .NET sample
2. Create one asynchronous example under the conventional root sample path
3. Convert C# types to Python dataclasses
4. Use async with for client and credential 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

Start with focused connector and lazy-export tests from the repository root:

$env:PYTHONPATH = (Join-Path $PWD "src")
.\.venv\Scripts\python -m pytest tests/test_{connector_name}.py tests/test_root_client_exports.py -v --tb=short

Required checks:

• Focused tests pass against this checkout, not another editable installation
• Direct client/model imports work and resolve under this checkout's src directory
• Generated registry and type stub agree; package import does not eagerly load connectors
• Full tests and code quality checks pass before submitting the connector PR

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

Run the sample offline with empty and nonempty mocked responses and assert its output contains no caught-and-printed errors. Test its actual parsed __main__ guard with the example functions mocked by AsyncMock, leaving main() and asyncio.run() intact. Assert each example is awaited once and no unawaited coroutine warning occurs. Keep only asyncio.run(main()) in the guard.

Checklist

Before completing, verify:

• src/azure/connectors/{connector_name}.py exists with proper structure
• One async-only root module owns the client, models, and trigger metadata; no sync or aio mirror exists
• Generated _client_registry.py and __init__.pyi agree with the connector inventory
• Package-root exports remain lazy; no eager barrel imports or failed-client None placeholders
• README.md connector table updated with correct status and test count
• tests/test_{connector_name}.py covers async lifecycle, operations, serialization, and applicable AsyncItemPaged / AsyncLROPoller behavior
• One async sample exists at samples/sample_connector_usage/sample_connector_usage_{connector_name}.py
• 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
• Offline sample checks cover empty and nonempty output and the actual main guard without unawaited coroutines

Files Modified

Common Issues

Import Error in __init__.py

The package root loads a client only when requested. Import failures must propagate; never hide a broken connector behind a try/except import or assign its client to None. Check direct connector imports, fix undefined references in CodefulSdkGenerator, then regenerate the connector, registry, and type stub. Do not hand-edit generated output or eagerly import all clients to bypass the lazy loader.

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.